Python Download Script

"""

download_asmc_wis2.py

=====================

Subscribe to ASMC WIS 2.0 topics and automatically download data files.

"""

import json

import logging

import ssl

import sys

import urllib.error

import urllib.request

from datetime import datetime

from pathlib import Path

from urllib.parse import urlparse

try:

import paho.mqtt.client as mqtt

except ImportError:

print("ERROR: paho-mqtt is not installed.")

print("Please run: python -m pip install \"paho-mqtt>=2.0,<3.0\"")

sys.exit(1)

BROKER = "globalbroker.meteo.fr"

PORT = 8883

USERNAME = "everyone"

PASSWORD = "everyone"

TOPIC = "origin/a/wis2/sg-mss-asmc/#"

DOWNLOAD_DIR = Path(__file__).parent / "downloads"

NOTIFY_ONLY = False

SKIP_HOSTS = {"api-open.data.gov.sg", "api.data.gov.sg"}

logging.basicConfig(

level=logging.INFO,

format="%(asctime)s %(levelname)-8s %(message)s",

datefmt="%Y-%m-%d %H:%M:%S",

)

log = logging.getLogger("asmc-wis2")

def get_download_url(payload: dict) -> str | None:

for link in payload.get("links", []):

if link.get("rel") == "canonical" and link.get("href"):

return link["href"]

return None

def is_skipped_host(url: str) -> bool:

return urlparse(url).netloc in SKIP_HOSTS

def build_save_path(topic: str, url: str, pubtime: str) -> Path:

parts = topic.split("/")

try:

asmc_idx = parts.index("sg-mss-asmc")

product_parts = parts[asmc_idx + 3:]

except (ValueError, IndexError):

product_parts = []

product_folder = "/".join(product_parts) if product_parts else "unknown"

filename = url.rstrip("/").split("/")[-1] or "unknown_file"

try:

date_str = pubtime[:10]

datetime.strptime(date_str, "%Y-%m-%d")

except (ValueError, TypeError):

date_str = "unknown-date"

return DOWNLOAD_DIR / product_folder / date_str / filename

def download_file(url: str, dest: Path) -> bool:

try:

dest.parent.mkdir(parents=True, exist_ok=True)

req = urllib.request.Request(url, headers={"User-Agent": "asmc-wis2-downloader/1.0"})

with urllib.request.urlopen(req, timeout=60) as resp:

dest.write_bytes(resp.read())

log.info(" Saved -> %s (%d bytes)", dest, dest.stat().st_size)

return True

except Exception as e:

log.error(" Error downloading %s: %s", url, e)

return False

def on_message(client, userdata, msg):

try:

payload = json.loads(msg.payload.decode("utf-8"))

except Exception as e:

log.warning("Could not parse message on %s: %s", msg.topic, e)

return

props = payload.get("properties", {})

pub_time = props.get("pubtime", "")

url = get_download_url(payload)

log.info("New notification: %s", msg.topic)

log.info(" Published : %s", pub_time or "(unknown)")

log.info(" URL : %s", url or "(none)")

if not url:

return

if is_skipped_host(url):

log.info(" Skipped : %s requires a separate API key", urlparse(url).netloc)

return

if NOTIFY_ONLY:

log.info(" NOTIFY_ONLY mode — skipping download")

return

dest = build_save_path(msg.topic, url, pub_time)

if dest.exists():

log.info(" Already exists — skipping")

return

download_file(url, dest)

def on_connect(client, userdata, flags, reason_code, properties):

if reason_code == 0:

log.info("Connected to %s:%s", BROKER, PORT)

client.subscribe(TOPIC, qos=1)

log.info("Subscribed to: %s", TOPIC)

log.info("Waiting for data... (Press Ctrl+C to stop)")

else:

log.error("Connection failed with code: %s", reason_code)

def main():

DOWNLOAD_DIR.mkdir(parents=True, exist_ok=True)

log.info("ASMC WIS 2.0 Downloader Activated")

client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2)

client.username_pw_set(USERNAME, PASSWORD)

client.tls_set()

client.on_connect = on_connect

client.on_message = on_message

try:

client.connect(BROKER, PORT, keepalive=60)

client.loop_forever()

except KeyboardInterrupt:

log.info("Stopped by user.")

finally:

client.disconnect()

if name == "__main__":

main()