Claude
5fb7ff09a3
Two new jobs operate on the SUNRISE_CAM, sharing three Python helpers (moon_phase, moon_detect, moon_composite) and one cached lunar texture: - moon-track.sh: nightly batch detects the moon in each east frame, crops a 480x480 box around it, stitches into an mp4 that holds the moon roughly centred while clouds and stars drift past. - moon-phase-monthly.sh: daily check that runs whichever phase composite is due that day. Handles full moon (posts D+3), first quarter (D+2, best-effort due to daytime-only geometry from east), and third quarter (D+2). Picks the frame closest in time to the exact phase moment that meets quality / altitude / illumination thresholds, then composites the cached lunar texture into it -- sky/halo/parallactic-angle/timing real from east, surface detail borrowed from the reference image. Honest-by-design: a 38-px white blob from a wide-field IP camera cannot be enhanced into crater detail by software. The composite makes the borrowing explicit and constrains everything else (when, where, sky, orientation) to match what east actually saw. install.sh now downloads the skyfield ephemeris (de421.bsp) and the default lunar reference (Wikipedia CC BY-SA full-moon photo) on first run. Both can be overridden via .env. https://claude.ai/code/session_015PBVDESC3KLMbq1LpA6qLn
413 lines
15 KiB
Python
Executable File
413 lines
15 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""moon_phase_monthly.py — build the monthly moon-phase composite.
|
|
|
|
Handles three phases, controlled by --phase:
|
|
|
|
full ~100% lit, posted MOON_FULL_POST_DELAY_DAYS after exact full
|
|
first-quarter ~50% lit waxing (right half lit in northern hemisphere)
|
|
third-quarter ~50% lit waning (left half lit in northern hemisphere)
|
|
|
|
Algorithm (per phase):
|
|
1. Find the most recent occurrence of the target phase (or honour --target).
|
|
2. Scan east frames across the collection window (D-Δb .. D+Δa).
|
|
3. Filter to frames where:
|
|
quality >= MOON_MIN_QUALITY
|
|
altitude_deg >= MOON_MIN_ALTITUDE_DEG
|
|
illumination is within the phase's illumination band
|
|
waxing-state matches the phase target
|
|
4. Pick the qualifying frame closest in time to exact phase UTC.
|
|
5. Composite the cached lunar texture into it (real sky + halo from east,
|
|
borrowed surface detail from the cached reference).
|
|
6. Optionally upload to Mattermost.
|
|
|
|
Geometry note — first-quarter from east is HARD: at first quarter the moon is
|
|
up from noon to midnight, but east only sees the eastern sky, so it captures
|
|
the moon during DAYTIME only (with a bright sky background). The detector is
|
|
brightness-based and may often fail to find a daytime moon. Set
|
|
MOON_FIRST_QUARTER_ENABLED=false in sky-cam.conf if you'd rather not chase it.
|
|
Full moon and third-quarter both rise after dark and stay in east's view —
|
|
those should land cleanly most months.
|
|
|
|
Usage:
|
|
moon_phase_monthly.py # auto: run any phase whose post-day = today
|
|
moon_phase_monthly.py --phase full # force a single phase
|
|
moon_phase_monthly.py --phase third-quarter --target 2026-04-09T11:51:00Z --dry-run
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
import pathlib
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
_here = pathlib.Path(__file__).resolve().parent
|
|
sys.path.insert(0, str(_here))
|
|
|
|
|
|
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'))
|
|
|
|
BASE_DIR = CONF.get('BASE_DIR') or str(_here / 'data')
|
|
MOVIES_DIR = CONF.get('MOVIES_DIR') or f'{BASE_DIR}/movies'
|
|
SUNRISE_CAM = CONF.get('SUNRISE_CAM', 'east')
|
|
TIMEZONE = CONF.get('TIMEZONE', 'UTC')
|
|
|
|
MIN_QUALITY = float(CONF.get('MOON_MIN_QUALITY', CONF.get('MOON_FULL_MIN_QUALITY', 0.55)))
|
|
MIN_ALTITUDE = float(CONF.get('MOON_MIN_ALTITUDE_DEG', CONF.get('MOON_FULL_MIN_ALTITUDE_DEG', 15.0)))
|
|
OUT_W = int(CONF.get('MOON_OUTPUT_W', CONF.get('MOON_FULL_OUTPUT_W', 1920)))
|
|
OUT_H = int(CONF.get('MOON_OUTPUT_H', CONF.get('MOON_FULL_OUTPUT_H', 1080)))
|
|
MOON_PCT = float(CONF.get('MOON_HEIGHT_PCT', CONF.get('MOON_FULL_HEIGHT_PCT', 0.70)))
|
|
REF_PATH = CONF.get('MOON_REFERENCE_PATH') or str(_here / 'moon-ref' / 'full-moon.jpg')
|
|
|
|
|
|
PHASE_SPEC = {
|
|
'full': {
|
|
'index': 2,
|
|
'label': 'Full Moon',
|
|
'emoji': '🌕',
|
|
'illum_min': float(CONF.get('MOON_FULL_MIN_ILLUMINATION', 0.95)),
|
|
'illum_max': 1.01,
|
|
'waxing': None,
|
|
'window_before': int(CONF.get('MOON_FULL_WINDOW_BEFORE_DAYS', 1)),
|
|
'window_after': int(CONF.get('MOON_FULL_WINDOW_AFTER_DAYS', 2)),
|
|
'post_delay': int(CONF.get('MOON_FULL_POST_DELAY_DAYS', 3)),
|
|
'enabled_key': 'MOON_FULL_ENABLED',
|
|
},
|
|
'first-quarter': {
|
|
'index': 1,
|
|
'label': 'First Quarter (Waxing Half)',
|
|
'emoji': '🌓',
|
|
'illum_min': float(CONF.get('MOON_QUARTER_MIN_ILLUMINATION', 0.40)),
|
|
'illum_max': float(CONF.get('MOON_QUARTER_MAX_ILLUMINATION', 0.65)),
|
|
'waxing': True,
|
|
'window_before': int(CONF.get('MOON_QUARTER_WINDOW_BEFORE_DAYS', 1)),
|
|
'window_after': int(CONF.get('MOON_QUARTER_WINDOW_AFTER_DAYS', 1)),
|
|
'post_delay': int(CONF.get('MOON_QUARTER_POST_DELAY_DAYS', 2)),
|
|
'enabled_key': 'MOON_FIRST_QUARTER_ENABLED',
|
|
},
|
|
'third-quarter': {
|
|
'index': 3,
|
|
'label': 'Third Quarter (Waning Half)',
|
|
'emoji': '🌗',
|
|
'illum_min': float(CONF.get('MOON_QUARTER_MIN_ILLUMINATION', 0.40)),
|
|
'illum_max': float(CONF.get('MOON_QUARTER_MAX_ILLUMINATION', 0.65)),
|
|
'waxing': False,
|
|
'window_before': int(CONF.get('MOON_QUARTER_WINDOW_BEFORE_DAYS', 1)),
|
|
'window_after': int(CONF.get('MOON_QUARTER_WINDOW_AFTER_DAYS', 1)),
|
|
'post_delay': int(CONF.get('MOON_QUARTER_POST_DELAY_DAYS', 2)),
|
|
'enabled_key': 'MOON_THIRD_QUARTER_ENABLED',
|
|
},
|
|
}
|
|
|
|
_FRAME_RE = re.compile(r'^(\d{2})-(\d{2})-(\d{2})\.jpg$')
|
|
|
|
|
|
def _local_tz():
|
|
try:
|
|
import pytz
|
|
return pytz.timezone(TIMEZONE)
|
|
except Exception:
|
|
return timezone.utc
|
|
|
|
|
|
def _frame_local_dt(date_str: str, fname: str):
|
|
m = _FRAME_RE.match(fname)
|
|
if not m:
|
|
return None
|
|
h, mn, s = (int(x) for x in m.groups())
|
|
y, mo, d = (int(x) for x in date_str.split('-'))
|
|
naive = datetime(y, mo, d, h, mn, s)
|
|
tz = _local_tz()
|
|
if hasattr(tz, 'localize'):
|
|
return tz.localize(naive)
|
|
return naive.replace(tzinfo=tz)
|
|
|
|
|
|
def _candidate_frames(cam: str, dates: list[str]) -> list[tuple[str, datetime]]:
|
|
out = []
|
|
for d in dates:
|
|
fdir = pathlib.Path(BASE_DIR) / cam / d
|
|
if not fdir.is_dir():
|
|
continue
|
|
for fname in sorted(os.listdir(fdir)):
|
|
if not fname.endswith('.jpg'):
|
|
continue
|
|
local_dt = _frame_local_dt(d, fname)
|
|
if local_dt is None:
|
|
continue
|
|
out.append((str(fdir / fname), local_dt.astimezone(timezone.utc)))
|
|
return out
|
|
|
|
|
|
def _format_local(dt_utc: datetime) -> str:
|
|
return dt_utc.astimezone(_local_tz()).strftime('%Y-%m-%d %H:%M:%S %Z')
|
|
|
|
|
|
def _notify(title: str, body: str):
|
|
try:
|
|
subprocess.run([str(_here / 'notify.sh'), title, body], check=False)
|
|
except FileNotFoundError:
|
|
pass
|
|
|
|
|
|
def _post_to_mattermost(image_path: str, message: str) -> bool:
|
|
import requests
|
|
base = CONF.get('mattermost_url', '').rstrip('/')
|
|
token = CONF.get('access_token', '')
|
|
channel_id = CONF.get('channel_id', '')
|
|
if not all([base, token, channel_id]):
|
|
print('mattermost credentials missing — skipping upload', file=sys.stderr)
|
|
return False
|
|
headers = {'Authorization': f'Bearer {token}'}
|
|
with open(image_path, 'rb') as f:
|
|
r = requests.post(
|
|
f'{base}/api/v4/files',
|
|
headers=headers,
|
|
files={'files': f},
|
|
data={'channel_id': channel_id},
|
|
)
|
|
if r.status_code != 201:
|
|
print(f'mattermost upload failed: {r.status_code} {r.text}', file=sys.stderr)
|
|
return False
|
|
file_id = r.json()['file_infos'][0]['id']
|
|
r = requests.post(
|
|
f'{base}/api/v4/posts',
|
|
headers=headers,
|
|
json={'channel_id': channel_id, 'message': message, 'file_ids': [file_id]},
|
|
)
|
|
if r.status_code != 201:
|
|
print(f'mattermost post failed: {r.status_code} {r.text}', file=sys.stderr)
|
|
return False
|
|
return True
|
|
|
|
|
|
def _output_subdir(phase: str) -> str:
|
|
return {
|
|
'full': 'full-moons',
|
|
'first-quarter': 'first-quarter',
|
|
'third-quarter': 'third-quarter',
|
|
}[phase]
|
|
|
|
|
|
def _output_filename(phase: str, target_utc: datetime) -> str:
|
|
slug = {'full': 'full', 'first-quarter': 'first-quarter', 'third-quarter': 'third-quarter'}[phase]
|
|
return f"{target_utc.strftime('%Y-%m')}-{slug}.jpg"
|
|
|
|
|
|
def run_phase(phase: str, target_utc: datetime | None, cam: str,
|
|
dry_run: bool, no_upload: bool, out_path: str | None) -> int:
|
|
spec = PHASE_SPEC[phase]
|
|
if CONF.get(spec['enabled_key'], 'true').lower() == 'false':
|
|
print(f'{spec["enabled_key"]}=false — skipping {phase}')
|
|
return 0
|
|
|
|
import moon_phase
|
|
from moon_detect import detect_moon
|
|
|
|
if target_utc is None:
|
|
now = datetime.now(timezone.utc)
|
|
events = moon_phase.phase_events_in_range(
|
|
now - timedelta(days=45), now, spec['index'])
|
|
if not events:
|
|
print(f'no recent {phase} found in past 45 days', file=sys.stderr)
|
|
return 1
|
|
target_utc = events[-1]
|
|
|
|
print(f'target {phase}: {target_utc.isoformat()} ({_format_local(target_utc)})')
|
|
|
|
tz = _local_tz()
|
|
target_local = target_utc.astimezone(tz)
|
|
dates = []
|
|
for i in range(-spec['window_before'], spec['window_after'] + 1):
|
|
d = (target_local + timedelta(days=i)).date()
|
|
dates.append(d.strftime('%Y-%m-%d'))
|
|
print(f'scanning dates: {dates}')
|
|
|
|
candidates = _candidate_frames(cam, dates)
|
|
print(f'frame count in window: {len(candidates)}')
|
|
if not candidates:
|
|
msg = (
|
|
f'No frames for {cam} in window {dates[0]}..{dates[-1]} '
|
|
f'around {phase} {target_utc.strftime("%Y-%m-%d %H:%MZ")}.'
|
|
)
|
|
_notify(f'{spec["emoji"]} {spec["label"]} — no frames available', msg)
|
|
print(msg)
|
|
return 0
|
|
|
|
qualifying = []
|
|
for path, utc_dt in candidates:
|
|
try:
|
|
alt, _ = moon_phase.altaz(utc_dt)
|
|
except Exception as e:
|
|
print(f'ERROR: moon_phase.altaz failed: {e}', file=sys.stderr)
|
|
return 2
|
|
if alt < MIN_ALTITUDE:
|
|
continue
|
|
det = detect_moon(path)
|
|
if det is None or det.quality < MIN_QUALITY:
|
|
continue
|
|
illum = moon_phase.illumination(utc_dt)
|
|
if illum < spec['illum_min'] or illum > spec['illum_max']:
|
|
continue
|
|
if spec['waxing'] is not None:
|
|
if moon_phase.waxing(utc_dt) != spec['waxing']:
|
|
continue
|
|
qualifying.append((path, utc_dt, det, alt, illum))
|
|
|
|
print(f'qualifying frames: {len(qualifying)}')
|
|
if not qualifying:
|
|
msg = (
|
|
f'No clear-shot {phase} frame in window {dates[0]}..{dates[-1]} '
|
|
f'(need quality≥{MIN_QUALITY}, altitude≥{MIN_ALTITUDE}°, '
|
|
f'illumination {spec["illum_min"]:.2f}-{spec["illum_max"]:.2f}). '
|
|
)
|
|
if phase == 'first-quarter':
|
|
msg += ('First quarter from east is best-effort because the moon is '
|
|
'only up during daylight hours — daytime detection often '
|
|
'fails. Lower MOON_MIN_QUALITY or accept that some months '
|
|
'will skip.')
|
|
else:
|
|
msg += 'Likely cloudy across the whole window.'
|
|
_notify(f'{spec["emoji"]} {spec["label"]} — no clear shot {target_utc.strftime("%B %Y")}', msg)
|
|
print(msg)
|
|
return 0
|
|
|
|
best = min(qualifying, key=lambda t: abs(t[1] - target_utc))
|
|
path, when_utc, det, alt, illum = best
|
|
delta_min = (when_utc - target_utc).total_seconds() / 60.0
|
|
local_dt = when_utc.astimezone(tz)
|
|
print(f'picked: {path}')
|
|
print(f' when_utc={when_utc.isoformat()} local={local_dt} '
|
|
f'altitude={alt:.1f}° illum={illum:.4f} quality={det.quality:.3f} '
|
|
f'Δtarget={delta_min:+.1f} min')
|
|
|
|
if dry_run:
|
|
return 0
|
|
|
|
if not pathlib.Path(REF_PATH).is_file():
|
|
msg = (f'Lunar reference image missing at {REF_PATH}. Re-run install.sh '
|
|
f'or drop a high-res full moon JPEG there manually.')
|
|
_notify(f'{spec["emoji"]} {spec["label"]} — reference missing', msg)
|
|
print(msg, file=sys.stderr)
|
|
return 3
|
|
|
|
if out_path is None:
|
|
out_dir = pathlib.Path(MOVIES_DIR) / cam / _output_subdir(phase)
|
|
out_dir.mkdir(parents=True, exist_ok=True)
|
|
out_path = str(out_dir / _output_filename(phase, target_utc))
|
|
|
|
from moon_composite import composite_full_moon
|
|
caption = (
|
|
f"{spec['label']} — {target_utc.strftime('%B %Y')} — "
|
|
f"sky-cam {cam} {local_dt.strftime('%Y-%m-%d %H:%M:%S %Z')}"
|
|
)
|
|
composite_full_moon(
|
|
path, det, when_utc, REF_PATH, out_path,
|
|
output_size=(OUT_W, OUT_H), moon_height_pct=MOON_PCT,
|
|
caption=caption,
|
|
)
|
|
print(f'wrote {out_path}')
|
|
|
|
if no_upload:
|
|
_notify(
|
|
f'{spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} (built, not posted)',
|
|
f'{out_path} — picked {local_dt}, Δtarget {delta_min:+.0f} min',
|
|
)
|
|
return 0
|
|
|
|
posted = _post_to_mattermost(
|
|
out_path,
|
|
f"{spec['emoji']} {spec['label']} — {target_utc.strftime('%B %Y')}\n"
|
|
f"Captured by sky-cam {cam} at {local_dt.strftime('%Y-%m-%d %H:%M:%S %Z')} "
|
|
f"({delta_min:+.0f} min from exact {phase}).",
|
|
)
|
|
if posted:
|
|
_notify(
|
|
f'{spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} posted',
|
|
f'Picked {local_dt} — Δtarget {delta_min:+.0f} min — {out_path}',
|
|
)
|
|
else:
|
|
_notify(
|
|
f'FAILED: {spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} upload',
|
|
f'Composite built at {out_path} but Mattermost upload failed.',
|
|
)
|
|
return 0
|
|
|
|
|
|
def auto_run(cam: str, dry_run: bool, no_upload: bool) -> int:
|
|
"""Daily check: run any phase whose post-day equals today (UTC)."""
|
|
import moon_phase
|
|
today_utc = datetime.now(timezone.utc).date()
|
|
ran_any = False
|
|
rc = 0
|
|
for phase, spec in PHASE_SPEC.items():
|
|
if CONF.get(spec['enabled_key'], 'true').lower() == 'false':
|
|
print(f'-- {phase}: {spec["enabled_key"]}=false → skip')
|
|
continue
|
|
events = moon_phase.phase_events_in_range(
|
|
datetime.combine(today_utc - timedelta(days=45), datetime.min.time(), tzinfo=timezone.utc),
|
|
datetime.now(timezone.utc),
|
|
spec['index'])
|
|
if not events:
|
|
continue
|
|
last_event = events[-1]
|
|
days_since = (today_utc - last_event.date()).days
|
|
if days_since == spec['post_delay']:
|
|
print(f'== running {phase} (last event {last_event.date()}, +{spec["post_delay"]} days = today) ==')
|
|
sub = run_phase(phase, last_event, cam, dry_run, no_upload, None)
|
|
rc = rc or sub
|
|
ran_any = True
|
|
else:
|
|
print(f'-- {phase}: last {last_event.date()}, days_since={days_since}, post_delay={spec["post_delay"]} → skip')
|
|
if not ran_any:
|
|
print('no phase scheduled for today')
|
|
return rc
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser()
|
|
p.add_argument('--phase', choices=list(PHASE_SPEC.keys()),
|
|
help='Run a single phase regardless of schedule')
|
|
p.add_argument('--target', help='Override phase event UTC, ISO 8601 (requires --phase)')
|
|
p.add_argument('--cam', default=SUNRISE_CAM)
|
|
p.add_argument('--dry-run', action='store_true')
|
|
p.add_argument('--no-upload', action='store_true')
|
|
p.add_argument('--out', help='Override output path (requires --phase)')
|
|
args = p.parse_args()
|
|
|
|
if args.phase:
|
|
target = None
|
|
if args.target:
|
|
target = datetime.fromisoformat(args.target.replace('Z', '+00:00'))
|
|
return run_phase(args.phase, target, args.cam, args.dry_run, args.no_upload, args.out)
|
|
return auto_run(args.cam, args.dry_run, args.no_upload)
|
|
|
|
|
|
if __name__ == '__main__':
|
|
sys.exit(main())
|