sky-cam/moon_dialamoon.py

#!/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/<ISO 8601 UTC, hour precision>

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())