#!/usr/bin/env python3 """moon_dialamoon.py — fetch + cache NASA SVS Dial-a-Moon renders. Dial-a-Moon publishes hourly pre-rendered moon images for the current year that include: - exact phase (terminator with real crater shadows) - real libration (the moon "wobbles" up to ~7° at the limb) - exact illumination fraction - apparent diameter For our purpose we want the image as it would appear to a ground observer at the requested UTC moment, downloaded once per phase event and cached forever under $MOON_DIALAMOON_CACHE_DIR. API: https://svs.gsfc.nasa.gov/api/dialamoon/ The response is JSON with an "image" object that carries one or more URLs (varies by year — keys seen include "url", "tif", "1024", etc.). We pick the largest reasonable JPEG/PNG variant, downsample to MOON_DIALAMOON_TARGET_PX on save to keep the cache tidy. Usage as a library: from moon_dialamoon import fetch_for_time path = fetch_for_time(datetime(2026, 4, 29, 21, 0, tzinfo=timezone.utc)) # path is a local cached PNG sized to MOON_DIALAMOON_TARGET_PX CLI: python3 moon_dialamoon.py 2026-04-29T21:00 # fetch and print path python3 moon_dialamoon.py 2026-04-29T21:00 --info # also print metadata """ from __future__ import annotations import argparse import json import os import pathlib import re import sys import urllib.request from datetime import datetime, timezone _here = pathlib.Path(__file__).resolve().parent def _read_conf(path): conf = {} try: with open(path) as f: for line in f: line = line.strip() if not line or line.startswith('#') or '=' not in line: continue k, v = line.split('=', 1) k = k.strip() v = v.strip() v = re.sub(r'\s+#.*$', '', v) v = v.strip('"').strip("'") m = re.match(r'^\$\{[^}]+:-([^}]*)\}$', v) if m: v = m.group(1).strip('"').strip("'") conf[k] = v except FileNotFoundError: pass return conf _CONF = _read_conf(_here / 'sky-cam.conf') _CONF.update(_read_conf(_here / '.env')) API_BASE = _CONF.get('MOON_DIALAMOON_API', 'https://svs.gsfc.nasa.gov/api/dialamoon') CACHE_DIR = pathlib.Path(_CONF.get('MOON_DIALAMOON_CACHE_DIR') or (_here / 'moon-ref' / 'dialamoon')) TARGET_PX = int(_CONF.get('MOON_DIALAMOON_TARGET_PX', 2048)) USER_AGENT = _CONF.get('MOON_DIALAMOON_USER_AGENT', 'sky-cam/1.0 (+https://github.com/outis1one/sky-cam)') TIMEOUT = int(_CONF.get('MOON_DIALAMOON_TIMEOUT_SEC', 30)) def _hour_key(dt: datetime) -> str: """Cache key — ISO hour, no minutes/seconds. Matches API resolution.""" if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) dt = dt.astimezone(timezone.utc) return dt.strftime('%Y-%m-%dT%H:00') def _cache_path(dt: datetime) -> pathlib.Path: return CACHE_DIR / f'{_hour_key(dt)}.png' def _meta_path(dt: datetime) -> pathlib.Path: return CACHE_DIR / f'{_hour_key(dt)}.json' def _http_get(url: str, accept: str = '*/*') -> bytes: req = urllib.request.Request(url, headers={ 'User-Agent': USER_AGENT, 'Accept': accept, }) with urllib.request.urlopen(req, timeout=TIMEOUT) as r: return r.read() def _api_query(dt: datetime) -> dict: url = f'{API_BASE}/{_hour_key(dt)}' data = _http_get(url, accept='application/json') return json.loads(data) def _pick_image_url(meta: dict) -> str | None: """Walk the response to find the largest JPG/PNG image URL. Dial-a-Moon's response shape has shifted over time — different years publish slightly different keys. We accept any of the documented forms. """ img = meta.get('image') or {} # Candidate URLs in priority order: explicit png/jpg, then anything ending # in those extensions inside nested fields. candidates: list[str] = [] if isinstance(img, dict): for key in ('png', 'jpg', 'jpeg', 'url', '4096', '2048', '1024', '512'): v = img.get(key) if isinstance(v, str): candidates.append(v) elif isinstance(v, dict): for vv in v.values(): if isinstance(vv, str): candidates.append(vv) # walk all string leaves once more in case the schema changed for v in img.values(): if isinstance(v, str): candidates.append(v) elif isinstance(img, str): candidates.append(img) # Prefer png > jpg > tif (we can't decode tif with stdlib, but Pillow can) def score(u: str) -> int: u = u.lower() if u.endswith('.png'): return 3 if u.endswith('.jpg') or u.endswith('.jpeg'): return 2 if u.endswith('.tif') or u.endswith('.tiff'): return 1 return 0 candidates = [c for c in candidates if score(c) > 0] if not candidates: return None candidates.sort(key=score, reverse=True) return candidates[0] def fetch_for_time(dt: datetime, force: bool = False) -> pathlib.Path: """Return the path to a cached dial-a-moon PNG for hour-of-`dt`. Downloads + downsamples on first call, cached forever after. Raises RuntimeError on network / parsing failure. """ CACHE_DIR.mkdir(parents=True, exist_ok=True) cache = _cache_path(dt) if cache.exists() and not force: return cache meta = _api_query(dt) # Persist the metadata next to the image so we can include phase # description, libration, etc. in captions later if we want. _meta_path(dt).write_text(json.dumps(meta, indent=2)) img_url = _pick_image_url(meta) if not img_url: raise RuntimeError( f'Dial-a-Moon API for {_hour_key(dt)} returned no usable image URL.\n' f'Response keys: {sorted(meta.keys())}' ) raw = _http_get(img_url) # Decode + downsample to TARGET_PX so the cache stays manageable from io import BytesIO from PIL import Image try: LANCZOS = Image.Resampling.LANCZOS except AttributeError: LANCZOS = Image.LANCZOS im = Image.open(BytesIO(raw)).convert('RGB') if max(im.size) > TARGET_PX: scale = TARGET_PX / max(im.size) new_size = (int(im.size[0] * scale), int(im.size[1] * scale)) im = im.resize(new_size, LANCZOS) im.save(cache, 'PNG') return cache def get_metadata(dt: datetime) -> dict | None: p = _meta_path(dt) if p.exists(): try: return json.loads(p.read_text()) except Exception: return None return None def _cli(): p = argparse.ArgumentParser() p.add_argument('when_utc', help='ISO 8601 UTC, e.g. 2026-04-29T21:00') p.add_argument('--force', action='store_true') p.add_argument('--info', action='store_true', help='print metadata too') args = p.parse_args() dt = datetime.fromisoformat(args.when_utc.replace('Z', '+00:00')) if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) try: path = fetch_for_time(dt, force=args.force) except Exception as e: print(f'ERROR: {e}', file=sys.stderr) return 1 print(path) if args.info: meta = get_metadata(dt) or {} for k in ('phase', 'subsolar_lon', 'subsolar_lat', 'subearth_lon', 'subearth_lat', 'posangle', 'distance', 'j2000_ra', 'j2000_dec'): if k in meta: print(f' {k}: {meta[k]}') return 0 if __name__ == '__main__': sys.exit(_cli())