Claude
9e2852fd73
Previous approach: scan a 3-day window, find best moon detection,
use that timestamp. Problems: fuzzy composite (multi-day scan could
pick a frame far from actual full moon), no clear tie between the
NASA render and a specific observable moment.
New approach — MOON_OBS_TIME_LOCAL (default 22:30 local):
On the night of the exact phase event, look at east frames in a
±MOON_OBS_WINDOW_MIN (default 30 min) window around the configured
time. The frame closest to that time determines atmosphere opacity;
the same time (rounded to hour) drives the NASA Dial-a-Moon fetch.
This gives one definitive moment per phase per month.
moon_phase_monthly.py:
- PHASE_SPEC stripped to just post_delay + enabled_key (all window/
illumination filtering removed — obs time is the only selector)
- run_phase() replaced with obs-time logic; opacity derived from
detect_moon quality on that single frame
- MOON_FULL_POST_DELAY_DAYS / MOON_QUARTER_POST_DELAY_DAYS default
changed to 1 (post morning after the phase, frames already on disk)
moon_composite.py:
- atmosphere blur default: output_width//10 (192px) → output_width//20
(96px) so cloud shapes survive the blur
sky-cam.conf:
- MOON_OBS_TIME_LOCAL=22:30, MOON_OBS_WINDOW_MIN=30
- Post delay defaults updated to 1
test_moon_composite.py:
- Tries real NASA SVS Dial-a-Moon API first; procedural disc only as
fallback if API unreachable
- Runs detect_moon on the east frame to compute real opacity
- Single output (test_composite_out.jpg)
https://claude.ai/code/session_01HkTxpNSTWtViZzxbrbKytR
475 lines
18 KiB
Python
Executable File
475 lines
18 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""moon_phase_monthly.py — build the monthly moon-phase close-up.
|
|
|
|
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) and pick
|
|
the frame closest in time to exact phase UTC where east successfully
|
|
detected the moon and standard quality / altitude / illumination
|
|
thresholds are met. East is the WITNESS — it confirms you actually had
|
|
a chance to see the moon that night.
|
|
3. Round east's capture timestamp to the nearest hour and fetch the NASA
|
|
SVS Dial-a-Moon render for that hour. This gives a real-physics moon
|
|
image with correct phase, libration and crater shadows.
|
|
4. Render full-screen on a black background (moon fills ~92% of frame
|
|
height), drop a caption naming the phase / month / capture moment /
|
|
attribution, and 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', 0.92))
|
|
REQUIRE_EAST_VERIFY = CONF.get('MOON_REQUIRE_EAST_VERIFY', 'true').lower() != 'false'
|
|
|
|
ATMOSPHERE_BLUR = int(CONF.get('MOON_ATMOSPHERE_BLUR', 0))
|
|
CLOUDY_POST_ENABLED = CONF.get('MOON_CLOUDY_POST_ENABLED', 'true').lower() != 'false'
|
|
|
|
_obs_parts = CONF.get('MOON_OBS_TIME_LOCAL', '22:30').split(':')
|
|
OBS_TIME_H = int(_obs_parts[0])
|
|
OBS_TIME_M = int(_obs_parts[1]) if len(_obs_parts) > 1 else 0
|
|
OBS_WINDOW_MIN = int(CONF.get('MOON_OBS_WINDOW_MIN', 30))
|
|
|
|
|
|
def _atmosphere_opacity_from_quality(quality: float | None) -> float:
|
|
"""Map moon detection quality to atmospheric overlay opacity.
|
|
|
|
quality >= 0.85 → 0.00 clear sky, no overlay
|
|
quality 0.70 → 0.20 light haze
|
|
quality 0.55 → 0.40 noticeable cloud, moon still detected
|
|
quality < 0.55 → up to 0.65 (overcast fallback frames)
|
|
quality is None → 0.68 no detection at all, heavy overcast
|
|
|
|
The overlay is applied OVER the NASA moon disk, so higher opacity means
|
|
more of the disk is obscured — physically correct for clouds above us.
|
|
"""
|
|
if quality is None:
|
|
return 0.68
|
|
if quality >= 0.85:
|
|
return 0.0
|
|
# Linear: 0.0 at quality=0.85, 0.40 at quality=0.55
|
|
raw = (0.85 - quality) / 0.30 * 0.40
|
|
return min(0.65, raw)
|
|
|
|
|
|
PHASE_SPEC = {
|
|
'full': {
|
|
'index': 2,
|
|
'label': 'Full Moon',
|
|
'emoji': '🌕',
|
|
'post_delay': int(CONF.get('MOON_FULL_POST_DELAY_DAYS', 1)),
|
|
'enabled_key': 'MOON_FULL_ENABLED',
|
|
},
|
|
'first-quarter': {
|
|
'index': 1,
|
|
'label': 'First Quarter (Waxing Half)',
|
|
'emoji': '🌓',
|
|
'post_delay': int(CONF.get('MOON_QUARTER_POST_DELAY_DAYS', 1)),
|
|
'enabled_key': 'MOON_FIRST_QUARTER_ENABLED',
|
|
},
|
|
'third-quarter': {
|
|
'index': 3,
|
|
'label': 'Third Quarter (Waning Half)',
|
|
'emoji': '🌗',
|
|
'post_delay': int(CONF.get('MOON_QUARTER_POST_DELAY_DAYS', 1)),
|
|
'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)
|
|
|
|
# If east-verification is disabled the user wants a post regardless of
|
|
# whether east could see the moon that night. Fetch dial-a-moon for the
|
|
# exact phase moment, render full-screen, post. Skips all east scanning.
|
|
if not REQUIRE_EAST_VERIFY:
|
|
print('MOON_REQUIRE_EAST_VERIFY=false — skipping east scan, using exact phase UTC')
|
|
return _render_and_post(phase, spec, target_utc, target_utc, target_local,
|
|
cam, dry_run, no_upload, out_path,
|
|
witness_text='not requiring east verification (set MOON_REQUIRE_EAST_VERIFY=true to require)')
|
|
|
|
# ── Fixed observation time on the night of the phase event ──────────────
|
|
# Instead of scanning a multi-day window for the best detection, we look
|
|
# at a short window around a configured local time (default 22:30) on the
|
|
# night of the exact phase. That single timestamp drives both the NASA
|
|
# Dial-a-Moon fetch (rounded to the nearest hour) and the atmosphere check.
|
|
obs_naive = datetime(
|
|
target_local.year, target_local.month, target_local.day,
|
|
OBS_TIME_H, OBS_TIME_M, 0,
|
|
)
|
|
if hasattr(tz, 'localize'):
|
|
obs_utc = tz.localize(obs_naive).astimezone(timezone.utc)
|
|
else:
|
|
obs_utc = obs_naive.replace(tzinfo=tz).astimezone(timezone.utc)
|
|
print(f'obs time: {obs_utc.isoformat()} ({_format_local(obs_utc)})')
|
|
|
|
# Check moon altitude at the observation time
|
|
try:
|
|
alt_at_obs, _ = moon_phase.altaz(obs_utc)
|
|
except Exception as e:
|
|
print(f'ERROR: moon_phase.altaz: {e}', file=sys.stderr)
|
|
return 2
|
|
print(f'moon altitude at obs time: {alt_at_obs:.1f}°')
|
|
|
|
# Gather east frames in the ±OBS_WINDOW_MIN window
|
|
obs_start = obs_utc - timedelta(minutes=OBS_WINDOW_MIN)
|
|
obs_end = obs_utc + timedelta(minutes=OBS_WINDOW_MIN)
|
|
obs_dates: list[str] = []
|
|
d = obs_start.astimezone(tz).date()
|
|
while d <= obs_end.astimezone(tz).date():
|
|
obs_dates.append(d.strftime('%Y-%m-%d'))
|
|
d += timedelta(days=1)
|
|
all_frames = _candidate_frames(cam, obs_dates)
|
|
window_frames = [(p, dt) for p, dt in all_frames if obs_start <= dt <= obs_end]
|
|
print(f'east frames in ±{OBS_WINDOW_MIN} min window: {len(window_frames)}')
|
|
|
|
# Determine atmosphere opacity from the frame closest to obs_utc
|
|
east_frame: str | None = None
|
|
opacity = 0.0
|
|
|
|
if alt_at_obs < MIN_ALTITUDE:
|
|
print('moon below horizon at obs time — posting clean NASA image')
|
|
_notify(
|
|
f'{spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} '
|
|
f'— moon below horizon at {OBS_TIME_H:02d}:{OBS_TIME_M:02d} local',
|
|
f'Adjust MOON_OBS_TIME_LOCAL in sky-cam.conf. Posting clean NASA render.',
|
|
)
|
|
elif window_frames:
|
|
best_f = min(window_frames, key=lambda t: abs(t[1] - obs_utc))
|
|
east_frame, best_dt = best_f
|
|
det = detect_moon(east_frame)
|
|
quality = det.quality if det is not None else None
|
|
opacity = _atmosphere_opacity_from_quality(quality)
|
|
offset_min = (best_dt - obs_utc).total_seconds() / 60.0
|
|
print(f'atmosphere frame: {east_frame}')
|
|
print(f' quality={quality} opacity={opacity:.2f} '
|
|
f'offset={offset_min:+.1f} min from obs time')
|
|
else:
|
|
print('no east frames in obs window — posting clean NASA image')
|
|
|
|
# Label atmospheric conditions for the caption
|
|
if opacity == 0.0:
|
|
atm_label = 'clear sky'
|
|
elif opacity < 0.15:
|
|
atm_label = 'slight haze'
|
|
elif opacity < 0.40:
|
|
atm_label = 'cloud cover'
|
|
else:
|
|
atm_label = 'heavy overcast'
|
|
|
|
obs_local_str = _format_local(obs_utc)
|
|
witness_text = (
|
|
f'sky-cam {cam} — {atm_label} at {obs_local_str} — NASA SVS Dial-a-Moon'
|
|
)
|
|
|
|
if dry_run:
|
|
print(f'dry-run: would post with opacity={opacity:.2f} ({atm_label})')
|
|
return 0
|
|
|
|
return _render_and_post(
|
|
phase, spec, target_utc, obs_utc, obs_utc.astimezone(tz),
|
|
cam, dry_run, no_upload, out_path,
|
|
witness_text=witness_text,
|
|
east_frame_path=east_frame,
|
|
atmosphere_opacity=opacity,
|
|
)
|
|
|
|
|
|
def _render_and_post(phase, spec, target_utc, when_utc, local_dt, cam,
|
|
dry_run, no_upload, out_path, witness_text,
|
|
east_frame_path=None, atmosphere_opacity=0.0):
|
|
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))
|
|
|
|
# Fetch the NASA SVS Dial-a-Moon render for the hour east captured the
|
|
# moon (or the exact phase moment if east-verification is off). The
|
|
# render carries the correct phase, libration and crater shadows for
|
|
# that UTC moment — the strongest possible match for what east "saw,"
|
|
# and free of the white-blob limitation.
|
|
import moon_dialamoon
|
|
try:
|
|
nasa_path = moon_dialamoon.fetch_for_time(when_utc)
|
|
except Exception as e:
|
|
msg = (
|
|
f'NASA SVS Dial-a-Moon fetch failed for '
|
|
f'{when_utc.strftime("%Y-%m-%dT%HZ")}: {e}'
|
|
)
|
|
_notify(f'{spec["emoji"]} {spec["label"]} — dial-a-moon fetch failed', msg)
|
|
print(msg, file=sys.stderr)
|
|
return 4
|
|
print(f'dial-a-moon: {nasa_path}')
|
|
|
|
from moon_composite import render_phase_closeup
|
|
caption = (
|
|
f"{spec['label']} — {target_utc.strftime('%B %Y')} — "
|
|
f"sky-cam {cam} {witness_text} — render: NASA SVS Dial-a-Moon"
|
|
)
|
|
render_phase_closeup(
|
|
str(nasa_path), out_path,
|
|
output_size=(OUT_W, OUT_H), moon_height_pct=MOON_PCT,
|
|
caption=caption,
|
|
east_frame_path=east_frame_path,
|
|
atmosphere_opacity=atmosphere_opacity,
|
|
atmosphere_blur=ATMOSPHERE_BLUR,
|
|
)
|
|
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} — {witness_text}',
|
|
)
|
|
return 0
|
|
|
|
posted = _post_to_mattermost(
|
|
out_path,
|
|
f"{spec['emoji']} {spec['label']} — {target_utc.strftime('%B %Y')}\n"
|
|
f"sky-cam {cam} {witness_text}.\n"
|
|
f"Surface render from NASA SVS Dial-a-Moon for that hour.",
|
|
)
|
|
if posted:
|
|
_notify(
|
|
f'{spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} posted',
|
|
f'{witness_text} — {out_path}',
|
|
)
|
|
else:
|
|
_notify(
|
|
f'FAILED: {spec["emoji"]} {spec["label"]} {target_utc.strftime("%B %Y")} upload',
|
|
f'Image 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())
|