Add nightly moon-track timelapse and monthly moon-phase composites

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
This commit is contained in:
Claude
2026-04-30 11:40:11 +00:00
parent 499bafdcc2
commit 5fb7ff09a3
10 changed files with 1484 additions and 0 deletions
+47
View File
@@ -5,6 +5,8 @@ Automated sky / timelapse camera scripts that produce:
- **Daily sunrise clip** — a speed-adjusted video of the sunrise window, uploaded to Mattermost each morning
- **Four Seasons timelapse** — daily clips sized to each Vivaldi movement's music duration, assembled automatically into per-movement montages (with music + attribution overlay) and a full-year video
- **Full-day timelapse** — a fixed-fps timelapse of every image captured that day, kept for a configurable retention window
- **Nightly moon-track timelapse** — every east night frame where the moon is visible is cropped around the moon and stitched into a short mp4 (the moon roughly held still while clouds and stars drift past)
- **Monthly moon-phase close-ups** — one composite per phase (full moon, first quarter, third quarter), framed as if east took it through a 65× telephoto. The sky/halo/parallactic-angle/timing are real-from-east; the lunar surface texture is borrowed from a cached high-res reference (no software can recover detail your camera didn't capture). Posted to Mattermost.
---
@@ -15,6 +17,7 @@ Automated sky / timelapse camera scripts that produce:
```bash
sudo apt install ffmpeg bc fonts-dejavu curl python3-pip
pip3 install suntime pytz requests
pip3 install skyfield Pillow numpy scipy # moon jobs (moon-track + moon-phase-monthly)
```
| Package | Purpose |
@@ -24,6 +27,8 @@ pip3 install suntime pytz requests
| `fonts-dejavu` | Text overlays (sunrise time, attribution) |
| `python3` + `suntime pytz` | Astronomical sunrise calculation — pure math, no internet, works indefinitely |
| `python3` + `requests` | Mattermost upload |
| `python3` + `skyfield` | Moon phases + altitude/azimuth + parallactic angle (offline after first ephemeris download) |
| `python3` + `Pillow numpy scipy` | Moon detection + phase compositing |
### 2. Download
@@ -262,6 +267,48 @@ Replace the date list with whatever range you need. Each run produces one `*-fi
./year-end-join.sh 2025 east
```
**Moon jobs**:
```bash
# Re-run last night's moon-track timelapse for east
./moon-track.sh east
# Back-fill moon-track for a specific past night
./moon-track.sh east 2026-04-15
# Auto mode — runs whichever phase composite is due today (no-ops otherwise)
./moon-phase-monthly.sh
# Force a single phase, picking the most recent occurrence
./moon-phase-monthly.sh --phase full
./moon-phase-monthly.sh --phase first-quarter
./moon-phase-monthly.sh --phase third-quarter
# Build but skip the Mattermost post
./moon-phase-monthly.sh --phase full --no-upload
# Dry run — find best frame, log it, build nothing
./moon-phase-monthly.sh --phase third-quarter --dry-run
# Back-fill a specific event by exact UTC moment
./moon-phase-monthly.sh --phase full --target 2026-04-01T15:51:00Z
# Inspect any moon-related stats for a frame
python3 moon_detect.py BASE_DIR/east/2026-04-29/21-07-00.jpg --debug /tmp/dbg.png
python3 moon_phase.py info 2026-04-29T21:07:00Z
```
**Posting schedule**:
| Job | When the timer fires | When the artifact actually appears |
|---|---|---|
| Sunrise video | `SCHEDULE_SUNRISE` (default 03:00 local) | A few minutes after sunrise + `SUNRISE_POST_MIN` |
| Moon-track timelapse | `SCHEDULE_MOON_TRACK` (default 02:30 local) | ~5 min after the timer, covers the previous night |
| 🌕 Full Moon composite | `SCHEDULE_MOON_PHASE` (default 09:30 local) | 3 days after exact full moon (configurable: `MOON_FULL_POST_DELAY_DAYS`) |
| 🌓 First Quarter composite | same timer | 2 days after exact first quarter (configurable: `MOON_QUARTER_POST_DELAY_DAYS`) — best-effort, see geometry note below |
| 🌗 Third Quarter composite | same timer | 2 days after exact third quarter |
The moon-phase timer fires every day; the script no-ops on days that aren't a post-day for any phase, so you'll see exactly three posts per lunar cycle in Mattermost (sometimes only two if first quarter detection fails — see geometry note in `sky-cam.conf`).
**Check capture status**:
```bash
systemctl --user status sky-cam-capture-east.service
+1
View File
@@ -23,6 +23,7 @@ echo "Done. Next steps:"
echo " 1. Install system packages (if not already present):"
echo " sudo apt install ffmpeg bc fonts-dejavu"
echo " pip3 install suntime pytz requests"
echo " pip3 install skyfield Pillow numpy scipy # moon jobs (moon-track, moon-phase-monthly)"
echo ""
echo " 2. Edit $TARGET/sky-cam.conf"
echo " — SCRIPT_DIR full path to the directory you'll run scripts from"
+66
View File
@@ -225,6 +225,72 @@ EOF
done
fi
# ── Moon jobs: nightly tracker + monthly full-moon composite ────────────────
# Both jobs target the SUNRISE_CAM (the east-facing camera). They share three
# Python helpers (moon_phase.py, moon_detect.py, moon_composite.py) and one
# downloaded asset (the lunar reference texture).
if [ "${MOON_TRACK_ENABLED:-true}" = "true" ] || [ "${MOON_FULL_ENABLED:-true}" = "true" ]; then
# Pre-download skyfield ephemeris so first run doesn't hit the network at
# an inopportune moment. Stored next to the scripts.
if [ ! -f "$HERE/de421.bsp" ]; then
echo "Downloading skyfield ephemeris (de421.bsp, ~17 MB)..."
if curl -fsSL -o "$HERE/de421.bsp" \
"https://ssd.jpl.nasa.gov/ftp/eph/planets/bsp/de421.bsp"; then
echo " ok"
else
echo " WARNING: ephemeris download failed — moon jobs will retry on first run"
rm -f "$HERE/de421.bsp"
fi
fi
# Lunar reference texture for moon_phase_monthly.py. Default URL is the
# Wikipedia "FullMoon2010" by Gregory H. Revera (CC BY-SA 3.0), 3500×3500.
# Override MOON_REFERENCE_URL in .env to use a different source — any
# high-res photo of a full moon on a black background works.
ref_dir="${MOON_REFERENCE_DIR:-$HERE/moon-ref}"
ref_path="${MOON_REFERENCE_PATH:-$ref_dir/full-moon.jpg}"
ref_url="${MOON_REFERENCE_URL:-https://upload.wikimedia.org/wikipedia/commons/e/e1/FullMoon2010.jpg}"
if [ ! -f "$ref_path" ]; then
mkdir -p "$(dirname "$ref_path")"
echo "Downloading lunar reference texture..."
if curl -fsSL -o "$ref_path" "$ref_url"; then
echo " ok → $ref_path"
else
echo " WARNING: lunar reference download failed"
echo " Drop a high-res full-moon JPEG at $ref_path manually,"
echo " or set MOON_REFERENCE_URL in .env and re-run install.sh."
rm -f "$ref_path"
fi
fi
fi
# Nightly moon-track job — runs on the SUNRISE_CAM only.
if [ "${MOON_TRACK_ENABLED:-true}" = "true" ]; then
write_service "sky-cam-moon-track" \
"$SUNRISE_CAM: nightly moon tracking timelapse" \
"moon-track.sh $SUNRISE_CAM"
write_timer "sky-cam-moon-track" \
"$SUNRISE_CAM: nightly moon tracking timelapse" \
"${SCHEDULE_MOON_TRACK:-02:30:00}"
timers+=("sky-cam-moon-track")
fi
# Moon-phase composite job — runs daily; the Python helper internally checks
# full / first-quarter / third-quarter and only acts when today is the
# matching post-day for one of them. One timer covers all three phases.
if [ "${MOON_FULL_ENABLED:-true}" = "true" ] \
|| [ "${MOON_FIRST_QUARTER_ENABLED:-true}" = "true" ] \
|| [ "${MOON_THIRD_QUARTER_ENABLED:-true}" = "true" ]; then
write_service "sky-cam-moon-phase" \
"$SUNRISE_CAM: monthly moon-phase composite (full + quarters)" \
"moon-phase-monthly.sh" \
$'After=network-online.target\nWants=network-online.target\n'
write_timer "sky-cam-moon-phase" \
"$SUNRISE_CAM: monthly moon-phase composite" \
"${SCHEDULE_MOON_PHASE:-09:30:00}"
timers+=("sky-cam-moon-phase")
fi
# ── Per-camera Four Seasons jobs ─────────────────────────────────────────────
# Reads CAMERAS and SCHEDULE_SEASONS_<cam> from sky-cam.conf.
# Script receives the camera name as $1 so it knows which camera to process.
+25
View File
@@ -0,0 +1,25 @@
#!/bin/bash
# moon-phase-monthly.sh — thin wrapper around moon_phase_monthly.py.
#
# Run via systemd daily. In auto mode (no flags) the Python helper checks
# all three phases (full, first-quarter, third-quarter) and runs the composite
# for any whose post-day equals today UTC. If you scheduled the timer for
# 09:30 local, you'll see posts hit Mattermost at:
#
# Full Moon 3 days after exact full moon
# First Quarter 2 days after exact first quarter (best-effort — daytime)
# Third Quarter 2 days after exact third quarter
#
# Manual:
# ./moon-phase-monthly.sh # auto mode
# ./moon-phase-monthly.sh --dry-run # auto, no post
# ./moon-phase-monthly.sh --phase full # force most recent full
# ./moon-phase-monthly.sh --phase third-quarter --target 2026-04-09T11:51:00Z
set -euo pipefail
SCRIPT_DIR="$(dirname "$(realpath "$0")")"
source "$SCRIPT_DIR/sky-cam.conf"
export TZ="$TIMEZONE"
exec python3 "$SCRIPT_DIR/moon_phase_monthly.py" "$@"
Executable
+106
View File
@@ -0,0 +1,106 @@
#!/bin/bash
# moon-track.sh — nightly: detect the moon in each east night frame, crop a
# fixed-size box around it, stitch the cropped sequence into an mp4 that holds
# the moon roughly centred while clouds, halo and stars drift past.
#
# Run once per day from systemd around 02:00, after the night the moon was up.
# Processes YESTERDAY's frames by default; pass YYYY-MM-DD to back-fill.
#
# Output: $MOVIES_DIR/<cam>/moon-track/YYYY/YYYY-MM-DD-moon-track.mp4
set -euo pipefail
SCRIPT_DIR="$(dirname "$(realpath "$0")")"
source "$SCRIPT_DIR/sky-cam.conf"
export TZ="$TIMEZONE"
CAM_NAME="${1:-${SUNRISE_CAM:-east}}"
DATE_ARG="${2:-}"
if [ -n "$DATE_ARG" ]; then
target_date=$(date --date="$DATE_ARG" +%Y-%m-%d)
else
target_date=$(date --date="yesterday" +%Y-%m-%d)
fi
if [ "${MOON_TRACK_ENABLED:-true}" != "true" ]; then
echo "MOON_TRACK_ENABLED=false — skipping"
exit 0
fi
image_dir="$BASE_DIR/$CAM_NAME/$target_date"
if [ ! -d "$image_dir" ]; then
echo "no images for $CAM_NAME on $target_date — skipping"
exit 0
fi
year="${target_date:0:4}"
out_dir="$MOVIES_DIR/$CAM_NAME/moon-track/$year"
mkdir -p "$out_dir"
final="$out_dir/${target_date}-moon-track.mp4"
# Crop size around the detected moon, in source pixels. At 4K with a ~38 px
# moon, 480 px gives a comfortable surround with halo and any clouds visible.
CROP_PX="${MOON_TRACK_CROP_PX:-480}"
FPS="${MOON_TRACK_FPS:-12}"
CRF="${MOON_TRACK_CRF:-24}"
work_dir=$(mktemp -d)
trap 'rm -rf "$work_dir"' EXIT
export SCRIPT_DIR
echo "Scanning $image_dir for night frames with the moon visible..."
python3 - "$image_dir" "$work_dir" "$CROP_PX" <<'PY'
import os, sys
sys.path.insert(0, os.environ['SCRIPT_DIR'])
from PIL import Image
from moon_detect import detect_moon
image_dir = sys.argv[1]
work_dir = sys.argv[2]
crop_px = int(sys.argv[3])
half = crop_px // 2
frames = sorted(f for f in os.listdir(image_dir) if f.endswith('.jpg'))
n = len(frames)
written = 0
for i, name in enumerate(frames):
p = os.path.join(image_dir, name)
det = detect_moon(p)
if det is None:
continue
cx, cy = det.centroid_xy
im = Image.open(p)
w, h = im.size
left = max(0, min(int(cx - half), w - crop_px))
top = max(0, min(int(cy - half), h - crop_px))
crop = im.crop((left, top, left + crop_px, top + crop_px))
out = os.path.join(work_dir, f'{written:06d}.jpg')
crop.save(out, quality=88)
written += 1
if (i + 1) % 200 == 0 or i + 1 == n:
print(f' scanned {i + 1}/{n} frames, kept {written}', flush=True)
print(f'moon-detected frames: {written}')
PY
count=$(find "$work_dir" -maxdepth 1 -name '*.jpg' | wc -l)
if [ "$count" -lt 10 ]; then
echo "only $count moon-visible frames on $target_date — skipping video build"
exit 0
fi
echo "encoding $count frames into $final ..."
ffmpeg -loglevel warning -y \
-framerate "$FPS" \
-i "$work_dir/%06d.jpg" \
-c:v libx264 -pix_fmt yuv420p -preset "${ENCODE_PRESET:-slow}" \
-crf "$CRF" \
"$final"
echo "moon-track: $final ($(du -h "$final" | cut -f1))"
# Retention sweep
RETENTION="${MOON_TRACK_RETENTION_DAYS:-90}"
if [ "$RETENTION" -gt 0 ]; then
find "$MOVIES_DIR/$CAM_NAME/moon-track" -type f -name '*-moon-track.mp4' \
-mtime +"$RETENTION" -delete 2>/dev/null || true
fi
+237
View File
@@ -0,0 +1,237 @@
#!/usr/bin/env python3
"""moon_composite.py — paint a high-res lunar texture into a sky-cam frame.
Goal: from a frame east captured of a small white blob (~38 px), produce a
1920x1080 image that looks like east took it through a 65x telephoto.
What's real, from east:
- Sky color, atmospheric halo, any clouds drifting past
- Time, parallactic angle (orientation of "up" on the moon)
- Position of the moon in the frame at that instant
What's borrowed:
- The lunar surface texture (one cached high-res reference image)
Library entry point:
from moon_composite import composite_full_moon
composite_full_moon(
source_jpg='/data/east/2026-04-29/21-07-00.jpg',
detection=detect_moon(...),
when_utc=datetime(2026, 4, 29, 21, 7, 0, tzinfo=timezone.utc),
ref_moon_path='/path/to/full-moon.jpg',
out_path='/movies/east/full-moons/2026-04-full-moon.jpg',
)
CLI:
python3 moon_composite.py SOURCE.jpg WHEN_UTC REF_MOON.jpg OUT.jpg
"""
from __future__ import annotations
import argparse
import math
import sys
from datetime import datetime, timezone
from pathlib import Path
import numpy as np
from PIL import Image, ImageDraw, ImageFilter
# Pillow ≥ 10 renamed resampling constants
try:
LANCZOS = Image.Resampling.LANCZOS
BICUBIC = Image.Resampling.BICUBIC
except AttributeError: # Pillow < 10
LANCZOS = Image.LANCZOS
BICUBIC = Image.BICUBIC
def _square_crop_to_disk(ref: Image.Image, threshold: int = 25) -> Image.Image:
"""Tight-crop a reference moon image to the disk's bounding square.
Most lunar reference photos have the disk on a large black field with
significant padding. We threshold on luminance and crop to bbox + small
margin so the disk fills our target square evenly.
"""
g = np.asarray(ref.convert('L'))
mask = g >= threshold
ys, xs = np.where(mask)
if len(xs) == 0:
return ref
x0, x1 = int(xs.min()), int(xs.max())
y0, y1 = int(ys.min()), int(ys.max())
cx = (x0 + x1) // 2
cy = (y0 + y1) // 2
half = max(x1 - x0, y1 - y0) // 2 + 4 # tiny margin
left = max(0, cx - half)
top = max(0, cy - half)
right = min(ref.width, cx + half)
bottom = min(ref.height, cy + half)
return ref.crop((left, top, right, bottom))
def _disk_mask(size: int, feather_px: int = 6) -> Image.Image:
"""Soft circular alpha mask the size of the reference moon image."""
m = Image.new('L', (size, size), 0)
draw = ImageDraw.Draw(m)
# Inset slightly so the feather sits inside the disk edge
draw.ellipse((feather_px, feather_px, size - feather_px, size - feather_px),
fill=255)
if feather_px > 0:
m = m.filter(ImageFilter.GaussianBlur(radius=feather_px))
return m
def _apply_phase_shadow(disk: Image.Image, phase_angle_deg: float,
waxing: bool) -> Image.Image:
"""Darken the un-lit portion of the disk based on phase.
phase_angle_deg: 0 = full, 90 = quarter, 180 = new.
waxing: True = lit on right, False = lit on left.
Implementation: the terminator is an ellipse whose semi-minor axis is
cos(phase_angle). Pixels on the un-lit side are multiplied by a small
factor (not zero, so the un-lit limb stays visible like real earthshine).
"""
if phase_angle_deg < 1.0:
return disk # full enough that shadow would be a single-pixel sliver
w, h = disk.size
# Build a mask: 1.0 in lit area, 0.04 in un-lit area, soft transition near
# the terminator.
cx, cy = w / 2.0, h / 2.0
r = min(w, h) / 2.0
yy, xx = np.mgrid[0:h, 0:w].astype(np.float32)
# Normalise to disk coords (-1..1)
nx = (xx - cx) / r
ny = (yy - cy) / r
# Distance from disk centre (we still want to clip to the disk)
in_disk = (nx * nx + ny * ny) <= 1.0
# Terminator equation: x_norm = cos(phase) on the appropriate side.
# For waxing moon, the lit portion is right of the terminator (nx > x_t).
cos_p = math.cos(math.radians(phase_angle_deg))
# When the moon is more than half lit (cos_p > 0), terminator is on the
# un-lit side and the lit portion is broader. When less than half
# (cos_p < 0), terminator is on the lit side.
# Distance from terminator (positive = lit side)
if waxing:
d = nx - (-cos_p)
else:
d = -(nx - cos_p)
# Smooth step around the terminator (~2% of radius)
soft_px = max(1.5 / r, 0.01)
lit = np.clip(0.5 + d / (2 * soft_px), 0.04, 1.0)
lit = np.where(in_disk, lit, 1.0) # leave outside-disk untouched
arr = np.asarray(disk).astype(np.float32)
arr = arr * lit[..., None]
return Image.fromarray(np.clip(arr, 0, 255).astype(np.uint8), disk.mode)
def composite_full_moon(
source_jpg: str,
detection, # MoonDetection from moon_detect
when_utc: datetime,
ref_moon_path: str,
out_path: str,
output_size: tuple[int, int] = (1920, 1080),
moon_height_pct: float = 0.70,
caption: str | None = None,
):
"""Build the full-moon close-up composite and write it to out_path."""
# Lazy import — avoids loading skyfield when caller doesn't need it
import moon_phase
src = Image.open(source_jpg).convert('RGB')
cx, cy = detection.centroid_xy
src_diam = detection.diameter_px
# ── Crop east around the moon, sized so the moon fills moon_height_pct ──
crop_h = int(round(src_diam / moon_height_pct))
crop_w = int(round(crop_h * output_size[0] / output_size[1]))
sw, sh = src.size
crop_w = min(crop_w, sw)
crop_h = min(crop_h, sh)
left = int(round(cx - crop_w / 2))
top = int(round(cy - crop_h / 2))
left = max(0, min(left, sw - crop_w))
top = max(0, min(top, sh - crop_h))
crop = src.crop((left, top, left + crop_w, top + crop_h))
bg = crop.resize(output_size, LANCZOS)
# ── Where is the moon's center within the upscaled background? ──
moon_x_in_crop = cx - left
moon_y_in_crop = cy - top
scale = output_size[1] / crop_h
out_moon_cx = moon_x_in_crop * scale
out_moon_cy = moon_y_in_crop * scale
# ── Load reference texture, tight-crop to disk ──
ref = Image.open(ref_moon_path).convert('RGB')
ref = _square_crop_to_disk(ref)
target_size = int(round(output_size[1] * moon_height_pct))
target_size += target_size % 2 # even
ref_resized = ref.resize((target_size, target_size), LANCZOS)
# ── Phase shadow (skip when essentially full) ──
illum = moon_phase.illumination(when_utc)
pa = moon_phase.phase_angle(when_utc)
if illum < 0.995:
wax = moon_phase.waxing(when_utc)
ref_resized = _apply_phase_shadow(ref_resized, pa, wax)
# ── Parallactic-angle rotation ──
par = moon_phase.parallactic_angle(when_utc)
# PIL rotates counter-clockwise for positive angles; we want celestial
# north to end up "up" in the camera image. Negate so the rotation
# direction matches image-space y-down convention.
ref_rot = ref_resized.rotate(-par, resample=BICUBIC, expand=False)
# ── Composite with feathered circular mask ──
feather = max(4, target_size // 200)
mask = _disk_mask(target_size, feather_px=feather)
paste_x = int(round(out_moon_cx - target_size / 2))
paste_y = int(round(out_moon_cy - target_size / 2))
# Clamp so the disk stays fully on canvas (recenter if needed)
paste_x = max(0, min(paste_x, output_size[0] - target_size))
paste_y = max(0, min(paste_y, output_size[1] - target_size))
bg.paste(ref_rot, (paste_x, paste_y), mask)
# ── Caption ──
if caption:
draw = ImageDraw.Draw(bg)
# Drop shadow for legibility
draw.text((24, output_size[1] - 44), caption, fill=(0, 0, 0))
draw.text((22, output_size[1] - 46), caption, fill=(220, 220, 220))
Path(out_path).parent.mkdir(parents=True, exist_ok=True)
bg.save(out_path, quality=92)
return out_path
def _cli():
p = argparse.ArgumentParser()
p.add_argument('source')
p.add_argument('when_utc', help='ISO 8601 UTC, e.g. 2026-04-29T21:07:00Z')
p.add_argument('ref_moon')
p.add_argument('out')
p.add_argument('--caption', default=None)
p.add_argument('--moon-pct', type=float, default=0.70)
args = p.parse_args()
from moon_detect import detect_moon
det = detect_moon(args.source)
if det is None:
print('ERROR: no moon detected in source', file=sys.stderr)
return 2
when = datetime.fromisoformat(args.when_utc.replace('Z', '+00:00'))
composite_full_moon(
args.source, det, when, args.ref_moon, args.out,
moon_height_pct=args.moon_pct, caption=args.caption,
)
print(f'wrote {args.out}')
return 0
if __name__ == '__main__':
sys.exit(_cli())
Executable
+217
View File
@@ -0,0 +1,217 @@
#!/usr/bin/env python3
"""moon_detect.py — find the moon disk in a sky-cam frame.
Returns centroid, radius, and a quality score derived from:
- roundness (area / (pi * r^2))
- isolation (no comparable bright blob within IGNORE_RADIUS_PX)
- sky (low halo extent → clearer sky around the moon)
Used as a library:
from moon_detect import detect_moon
result = detect_moon('/path/to/frame.jpg')
if result is not None:
cx, cy = result['centroid']
r = result['radius']
score = result['quality']
CLI (handy for tuning):
python3 moon_detect.py /path/to/frame.jpg
python3 moon_detect.py /path/to/frame.jpg --debug debug.png
The detector ignores the bottom OVERLAY_PX rows because capture frames carry a
burnt-in timestamp that contains saturated pixels.
"""
from __future__ import annotations
import argparse
import json
import math
import sys
from dataclasses import dataclass
import numpy as np
from PIL import Image, ImageDraw
from scipy import ndimage
# --- Tuning constants --------------------------------------------------------
SATURATED_THRESHOLD = 240 # 0..255 — pixels at or above this count as "moon disk"
HALO_THRESHOLD = 100 # 0..255 — pixels above this count toward halo extent
OVERLAY_PX = 200 # bottom rows to ignore (timestamp overlay)
MIN_DIAMETER_PX = 10 # blob smaller than this is noise
MAX_DIAMETER_PX = 120 # blob bigger than this is probably the sun, not the moon
MIN_ROUNDNESS = 0.55 # area / (pi * r^2) — perfect circle = 1
ISOLATION_PX = 200 # other comparable blob this close → reject
# Halo extent — typical moon halo is ~3.5x disk radius; >5x means thick clouds
MAX_HALO_RATIO = 6.0
@dataclass
class MoonDetection:
centroid_xy: tuple[float, float]
radius_px: float
diameter_px: float
blob_pixels: int
roundness: float
halo_radius_px: float
halo_ratio: float
isolation_px: float
quality: float
def asdict(self) -> dict:
return {
'centroid': list(self.centroid_xy),
'radius': self.radius_px,
'diameter': self.diameter_px,
'blob_pixels': self.blob_pixels,
'roundness': self.roundness,
'halo_radius': self.halo_radius_px,
'halo_ratio': self.halo_ratio,
'isolation': self.isolation_px,
'quality': self.quality,
}
def _grayscale_array(image_path: str) -> np.ndarray:
im = Image.open(image_path).convert('L')
return np.asarray(im)
def _largest_round_blob(mask: np.ndarray) -> tuple[int, np.ndarray, np.ndarray] | None:
labels, n = ndimage.label(mask)
if n == 0:
return None
sizes = ndimage.sum(mask, labels, range(1, n + 1))
# Sort blobs by size descending; check roundness on the top few
order = np.argsort(sizes)[::-1]
for idx in order[:8]:
label_id = idx + 1
ys, xs = np.where(labels == label_id)
if len(xs) < 4:
continue
bbw = xs.max() - xs.min() + 1
bbh = ys.max() - ys.min() + 1
diam = max(bbw, bbh)
if diam < MIN_DIAMETER_PX or diam > MAX_DIAMETER_PX:
continue
radius = diam / 2.0
roundness = len(xs) / (math.pi * radius * radius)
if roundness < MIN_ROUNDNESS:
continue
return label_id, ys, xs
return None
def detect_moon(image_path: str) -> MoonDetection | None:
"""Return MoonDetection or None if no acceptable moon is found."""
a = _grayscale_array(image_path)
if OVERLAY_PX > 0:
a = a.copy()
a[-OVERLAY_PX:, :] = 0
mask = a >= SATURATED_THRESHOLD
if not mask.any():
return None
found = _largest_round_blob(mask)
if found is None:
return None
_, ys, xs = found
cx = float(xs.mean())
cy = float(ys.mean())
blob_pixels = int(len(xs))
bbw = xs.max() - xs.min() + 1
bbh = ys.max() - ys.min() + 1
diameter = float(max(bbw, bbh))
radius = diameter / 2.0
roundness = blob_pixels / (math.pi * radius * radius)
# Isolation: any other saturated blob nearby of comparable size?
labels_full, n_full = ndimage.label(mask)
own_label = labels_full[int(round(cy)), int(round(cx))]
isolation = float('inf')
for label_id in range(1, n_full + 1):
if label_id == own_label:
continue
ys2, xs2 = np.where(labels_full == label_id)
if len(xs2) < blob_pixels * 0.3:
continue
d = math.hypot(xs2.mean() - cx, ys2.mean() - cy)
if d < isolation:
isolation = d
if isolation < ISOLATION_PX:
return None
# Halo: connected component above HALO_THRESHOLD that contains the centroid
halo_mask = a >= HALO_THRESHOLD
halo_labels, _ = ndimage.label(halo_mask)
halo_id = halo_labels[int(round(cy)), int(round(cx))]
if halo_id == 0:
halo_radius = radius
else:
yh, xh = np.where(halo_labels == halo_id)
halo_radius = max(xh.max() - xh.min(), yh.max() - yh.min()) / 2.0
halo_ratio = halo_radius / radius if radius > 0 else 1.0
if halo_ratio > MAX_HALO_RATIO:
return None # too much glow → probably thick cloud cover
# Quality score: roundness (0..1), low halo (1 = clear, 0 = thick cloud),
# isolation factor (1 if very isolated, less if close to other lights).
halo_clean = max(0.0, min(1.0, (MAX_HALO_RATIO - halo_ratio) / (MAX_HALO_RATIO - 1.5)))
iso_factor = 1.0 if isolation == float('inf') else min(1.0, isolation / 600.0)
quality = 0.5 * roundness + 0.35 * halo_clean + 0.15 * iso_factor
return MoonDetection(
centroid_xy=(cx, cy),
radius_px=radius,
diameter_px=diameter,
blob_pixels=blob_pixels,
roundness=roundness,
halo_radius_px=halo_radius,
halo_ratio=halo_ratio,
isolation_px=isolation if isolation != float('inf') else -1.0,
quality=quality,
)
def _draw_debug(image_path: str, detection: MoonDetection | None, out_path: str):
im = Image.open(image_path).convert('RGB')
draw = ImageDraw.Draw(im)
if detection is None:
draw.text((20, 20), 'NO MOON DETECTED', fill='red')
else:
cx, cy = detection.centroid_xy
r = detection.radius_px
hr = detection.halo_radius_px
draw.ellipse((cx - r, cy - r, cx + r, cy + r), outline='yellow', width=4)
draw.ellipse((cx - hr, cy - hr, cx + hr, cy + hr), outline='cyan', width=2)
draw.text((20, 20), f'q={detection.quality:.2f} r={r:.1f} halo={hr:.1f}',
fill='yellow')
im.save(out_path)
def _cli():
p = argparse.ArgumentParser()
p.add_argument('image')
p.add_argument('--debug', help='write annotated PNG to this path')
p.add_argument('--json', action='store_true', help='emit JSON for scripting')
args = p.parse_args()
det = detect_moon(args.image)
if args.json:
print(json.dumps(det.asdict() if det else None))
elif det is None:
print('no moon detected')
else:
d = det.asdict()
for k, v in d.items():
print(f'{k}={v}')
if args.debug:
_draw_debug(args.image, det, args.debug)
print(f'debug image written: {args.debug}', file=sys.stderr)
return 0 if det else 2
if __name__ == '__main__':
sys.exit(_cli())
Executable
+264
View File
@@ -0,0 +1,264 @@
#!/usr/bin/env python3
"""moon_phase.py — moon ephemeris lookups for sky-cam.
Provides:
- full_moons_in_range(start, end) list of UTC datetimes of full moons
- nearest_full_moon(when) UTC datetime of full moon nearest 'when'
- illumination(when) 0.0..1.0 fraction lit
- phase_angle(when) degrees, 0=full, 180=new
- waxing(when) True if moon is waxing
- altaz(when, lat, lon) (altitude_deg, azimuth_deg)
- parallactic_angle(when, lat, lon) degrees, rotation to put celestial north up
Used as a library by other scripts and as a CLI:
python3 moon_phase.py next-full # next full moon UTC + local
python3 moon_phase.py altaz <ISO-UTC> # alt/az from configured location
python3 moon_phase.py info <ISO-UTC> # everything for one timestamp
All times assume UTC unless tagged otherwise. Local timezone comes from
sky-cam.conf TIMEZONE for display only.
"""
from __future__ import annotations
import math
import os
import pathlib
import re
import sys
from datetime import datetime, timedelta, 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'))
LATITUDE = float(_conf.get('LATITUDE') or 0.0)
LONGITUDE = float(_conf.get('LONGITUDE') or 0.0)
TIMEZONE = _conf.get('TIMEZONE', 'UTC')
# Cached ephemeris — bootstrap.sh pre-downloads de421.bsp into SCRIPT_DIR so
# this never needs to hit the network at run time.
_EPH_PATH = _here / 'de421.bsp'
_ts = None
_eph = None
_observer = None
def _lazy():
"""Defer skyfield import + ephemeris load until first use.
Lets the module be imported by tests and by --help paths even when
skyfield is missing or the ephemeris hasn't been downloaded yet.
"""
global _ts, _eph, _observer
if _ts is not None:
return
from skyfield.api import Loader, wgs84
loader = Loader(str(_here), verbose=False)
_ts = loader.timescale()
if _EPH_PATH.exists():
_eph = loader('de421.bsp')
else:
_eph = loader('de421.bsp') # downloads on first run
_observer = _eph['earth'] + wgs84.latlon(LATITUDE, LONGITUDE)
def _to_utc(when: datetime) -> datetime:
if when.tzinfo is None:
return when.replace(tzinfo=timezone.utc)
return when.astimezone(timezone.utc)
def _t(when: datetime):
_lazy()
w = _to_utc(when)
return _ts.from_datetime(w)
def full_moons_in_range(start: datetime, end: datetime) -> list[datetime]:
"""Return UTC datetimes of every full moon between start and end (inclusive)."""
return phase_events_in_range(start, end, 2)
def phase_events_in_range(start: datetime, end: datetime, phase_index: int) -> list[datetime]:
"""Return UTC datetimes of every occurrence of `phase_index` between start and end.
skyfield phase indices: 0 = new, 1 = first quarter, 2 = full, 3 = last quarter.
"""
_lazy()
from skyfield.almanac import find_discrete, moon_phases
t0 = _t(start)
t1 = _t(end)
times, phases = find_discrete(t0, t1, moon_phases(_eph))
return [t.utc_datetime() for t, p in zip(times, phases) if p == phase_index]
def nearest_full_moon(when: datetime) -> datetime:
"""Full moon UTC nearest 'when' — searches a 45-day window centred on it."""
w = _to_utc(when)
candidates = full_moons_in_range(w - timedelta(days=45), w + timedelta(days=45))
return min(candidates, key=lambda d: abs(d - w))
def illumination(when: datetime) -> float:
"""Fraction of the moon's disk that is illuminated (0..1)."""
_lazy()
from skyfield.almanac import fraction_illuminated
return float(fraction_illuminated(_eph, 'moon', _t(when)))
def phase_angle(when: datetime) -> float:
"""Sun-Moon-Earth phase angle in degrees: 0=full, 90=quarter, 180=new."""
_lazy()
earth = _eph['earth']
sun = _eph['sun']
moon = _eph['moon']
t = _t(when)
e = earth.at(t)
s_from_moon = (sun - moon).at(t)
e_from_moon = (e.position.au - moon.at(t).position.au)
# angle between sun→moon→earth
a = s_from_moon.position.au
b = e_from_moon
cosang = (a[0]*b[0] + a[1]*b[1] + a[2]*b[2]) / (
math.sqrt(a[0]**2 + a[1]**2 + a[2]**2)
* math.sqrt(b[0]**2 + b[1]**2 + b[2]**2)
)
cosang = max(-1.0, min(1.0, cosang))
return math.degrees(math.acos(cosang))
def waxing(when: datetime) -> bool:
"""True if moon is waxing (illumination growing)."""
now = illumination(when)
later = illumination(when + timedelta(hours=6))
return later > now
def altaz(when: datetime, lat: float | None = None, lon: float | None = None):
"""Apparent altitude/azimuth in degrees as seen from (lat, lon).
Falls back to configured LATITUDE/LONGITUDE if not specified.
"""
_lazy()
from skyfield.api import wgs84
if lat is None and lon is None:
obs = _observer
else:
obs = _eph['earth'] + wgs84.latlon(
LATITUDE if lat is None else lat,
LONGITUDE if lon is None else lon,
)
t = _t(when)
alt, az, _ = obs.at(t).observe(_eph['moon']).apparent().altaz()
return float(alt.degrees), float(az.degrees)
def parallactic_angle(when: datetime, lat: float | None = None, lon: float | None = None) -> float:
"""Parallactic angle in degrees — rotates a moon image so celestial north is up.
sin(q) = sin(H) * cos(lat) / cos(alt)
where H is the hour angle and alt is the altitude.
Returned value is the angle to rotate the lunar texture clockwise (in the
image sense, y-down) so celestial north points up in the camera frame.
"""
_lazy()
from skyfield.api import wgs84
if lat is None:
lat = LATITUDE
if lon is None:
lon = LONGITUDE
obs = _eph['earth'] + wgs84.latlon(lat, lon)
t = _t(when)
apparent = obs.at(t).observe(_eph['moon']).apparent()
alt, _, _ = apparent.altaz()
ra, dec, _ = apparent.radec(epoch='date')
# local sidereal time at observer's longitude → hour angle
lst = t.gast * 15.0 + lon # gast in hours → degrees, plus longitude
H = math.radians(lst - ra._degrees)
phi = math.radians(lat)
delta = math.radians(dec.degrees)
# Standard parallactic angle formula
q = math.atan2(math.sin(H), math.tan(phi) * math.cos(delta) - math.sin(delta) * math.cos(H))
return math.degrees(q)
def _format_local(dt_utc: datetime) -> str:
try:
import pytz
tz = pytz.timezone(TIMEZONE)
return dt_utc.astimezone(tz).strftime('%Y-%m-%d %H:%M:%S %Z')
except Exception:
return dt_utc.strftime('%Y-%m-%d %H:%M:%S UTC')
def _cli():
if len(sys.argv) < 2:
print(__doc__.strip())
return 1
cmd = sys.argv[1]
if cmd == 'next-full':
now = datetime.now(timezone.utc)
fm = full_moons_in_range(now, now + timedelta(days=45))
if not fm:
print('No full moon found in the next 45 days?', file=sys.stderr)
return 1
next_fm = fm[0]
print(f'next_full_moon_utc={next_fm.strftime("%Y-%m-%dT%H:%M:%SZ")}')
print(f'next_full_moon_local={_format_local(next_fm)}')
return 0
if cmd == 'nearest-full':
when = datetime.fromisoformat(sys.argv[2].replace('Z', '+00:00'))
fm = nearest_full_moon(when)
print(f'nearest_full_moon_utc={fm.strftime("%Y-%m-%dT%H:%M:%SZ")}')
return 0
if cmd == 'altaz':
when = datetime.fromisoformat(sys.argv[2].replace('Z', '+00:00'))
alt, az = altaz(when)
print(f'altitude_deg={alt:.3f}')
print(f'azimuth_deg={az:.3f}')
return 0
if cmd == 'info':
when = datetime.fromisoformat(sys.argv[2].replace('Z', '+00:00'))
alt, az = altaz(when)
print(f'utc={when.strftime("%Y-%m-%dT%H:%M:%SZ")}')
print(f'local={_format_local(when)}')
print(f'illumination={illumination(when):.4f}')
print(f'phase_angle_deg={phase_angle(when):.2f}')
print(f'waxing={waxing(when)}')
print(f'altitude_deg={alt:.3f}')
print(f'azimuth_deg={az:.3f}')
print(f'parallactic_angle_deg={parallactic_angle(when):.3f}')
return 0
print(f'unknown command: {cmd}', file=sys.stderr)
return 1
if __name__ == '__main__':
sys.exit(_cli())
+412
View File
@@ -0,0 +1,412 @@
#!/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())
+109
View File
@@ -47,6 +47,18 @@
# on the last day of a movement, auto-triggers
# montage-mvt.sh
#
# moon-track.sh — nightly at SCHEDULE_MOON_TRACK (SUNRISE_CAM only)
# detects the moon in each east night frame, crops
# a tracked sequence, stitches into an mp4
#
# moon-phase-monthly.sh — daily at SCHEDULE_MOON_PHASE (SUNRISE_CAM only)
# no-ops except on phase post-days, when it picks
# the best east frame from the collection window,
# composites a high-res lunar texture into it,
# and posts to Mattermost. Handles full moon,
# first quarter (waxing half), and third quarter
# (waning half).
#
#
# install.sh — run once at setup, and again if this file changes
# generates and installs systemd units from conf
@@ -135,6 +147,25 @@
# ← run manually only; no script calls it
# → safe to rename with no other changes needed
#
# moon-track.sh
# ← called by systemd sky-cam-moon-track.timer (SCHEDULE_MOON_TRACK)
# → if renamed: update install.sh (the moon-track write_service block)
# calls → moon_detect.py (Python lib via heredoc), ffmpeg
#
# moon-phase-monthly.sh
# ← called by systemd sky-cam-moon-phase.timer (SCHEDULE_MOON_PHASE)
# → if renamed: update install.sh (the moon-phase write_service block)
# calls → moon_phase_monthly.py
#
# moon_phase_monthly.py
# ← called by moon-phase-monthly.sh
# → if renamed: update moon-phase-monthly.sh (the exec line at the bottom)
# calls → moon_phase.py, moon_detect.py, moon_composite.py, notify.sh
#
# moon_phase.py / moon_detect.py / moon_composite.py
# ← Python libraries used by moon-track.sh and moon_phase_monthly.py
# → if renamed: update moon-track.sh, moon_phase_monthly.py, README
#
# =============================================================================
# ── Install location ──────────────────────────────────────────────────────────
@@ -235,6 +266,10 @@ SUNRISE_CAM=east # which camera faces east (gets the sunrise v
#
SCHEDULE_SUNRISE=03:00:00
# Moon jobs (see "Moon jobs" section further down)
SCHEDULE_MOON_TRACK=02:30:00 # nightly tracker — runs after midnight, before seasons
SCHEDULE_MOON_PHASE=09:30:00 # daily check; no-ops except on phase post-days
# Four Seasons daily clip — one per camera, staggered 30 min apart.
# Processes yesterday's images; on the last day of a movement auto-triggers
# montage-mvt.sh. Space cameras at least 30 min apart to avoid disk contention.
@@ -397,6 +432,80 @@ AMBIENT_RETENTION_DAYS=30 # global default; 0 = keep forever
#AMBIENT_RETENTION_DAYS_north=30
#AMBIENT_RETENTION_DAYS_south=60 # keep south longer for nature sound library
# ── Moon jobs (nightly tracker + monthly phase close-ups) ────────────────────
# All moon jobs operate on the SUNRISE_CAM (the east-facing camera that already
# has a clear view of the eastern sky). They share three Python helpers
# (moon_phase.py, moon_detect.py, moon_composite.py) and one cached lunar
# reference texture downloaded by install.sh.
#
# moon-track.sh nightly batch — detects the moon in each frame,
# crops a 480×480 box around it, stitches the
# tracked sequence into an mp4. Output:
# $MOVIES_DIR/<cam>/moon-track/YYYY/YYYY-MM-DD-moon-track.mp4
#
# moon-phase-monthly.sh daily check; runs whichever phase composite is
# due today. Three phases handled:
# 🌕 Full Moon posts D + MOON_FULL_POST_DELAY_DAYS
# 🌓 First Quarter posts D + MOON_QUARTER_POST_DELAY_DAYS
# 🌗 Third Quarter posts D + MOON_QUARTER_POST_DELAY_DAYS
# Each picks the frame closest in time to the
# exact phase moment that meets quality / altitude
# / illumination thresholds, composites the
# cached lunar texture into it (sky/halo/angle
# real from east, surface detail borrowed),
# uploads to Mattermost. Output:
# $MOVIES_DIR/<cam>/full-moons/YYYY-MM-full.jpg
# $MOVIES_DIR/<cam>/first-quarter/YYYY-MM-first-quarter.jpg
# $MOVIES_DIR/<cam>/third-quarter/YYYY-MM-third-quarter.jpg
#
# Honest-by-design: the surface texture is not from your camera (no software
# can recover detail your camera didn't capture). The framing — sky color,
# halo, parallactic angle, exact moment of capture — is all real-from-east.
#
# Geometry caveat: east only sees the eastern sky. At first quarter the moon
# is up only from noon to midnight, transiting south at sunset, so east only
# catches it during DAYTIME. The brightness-based detector often fails on
# daytime moon shots — first quarter is best-effort. Full moon and third
# quarter both rise after dark and stay in east's view; those should land
# cleanly most months.
#
# Toggles (any can be disabled independently):
MOON_TRACK_ENABLED=true
MOON_FULL_ENABLED=true
MOON_FIRST_QUARTER_ENABLED=true # set false if daytime-detection misses are noisy
MOON_THIRD_QUARTER_ENABLED=true
# Nightly tracker tuning ──────────────────────────────────────────────────────
MOON_TRACK_CROP_PX=480 # pixels — box size around the moon (source coords)
MOON_TRACK_FPS=12 # output mp4 framerate
MOON_TRACK_CRF=24 # output mp4 CRF
MOON_TRACK_RETENTION_DAYS=90 # delete tracker mp4s older than this; 0 = forever
# Full-moon monthly tuning ────────────────────────────────────────────────────
# How many days after the exact full moon to post. 3 = waits for D-1..D+2
# nights to be on disk, then runs the morning of D+3.
MOON_FULL_POST_DELAY_DAYS=3
# Frame-acceptance thresholds — a candidate must beat all three to qualify.
# Lower = more permissive (accept hazier nights / lower moon). If you find
# the script never finds a clear shot, loosen these.
MOON_FULL_MIN_QUALITY=0.55 # 0..1 from moon_detect (roundness × halo × isolation)
MOON_FULL_MIN_ALTITUDE_DEG=15 # below this the moon is in trees / on the horizon
MOON_FULL_MIN_ILLUMINATION=0.95 # 0..1 — ~95% lit covers ±2 days from exact full
# Output frame. Default 1920×1080 to match the sunrise videos.
MOON_FULL_OUTPUT_W=1920
MOON_FULL_OUTPUT_H=1080
MOON_FULL_HEIGHT_PCT=0.70 # moon disk fills this fraction of frame height
# Lunar reference texture — downloaded once by install.sh, reused forever.
# Override MOON_REFERENCE_URL in .env to use a different image. Any high-res
# photo of a full moon on a black background works. Default is the Wikipedia
# "FullMoon2010" by Gregory H. Revera (CC BY-SA 3.0), 3500×3500.
#MOON_REFERENCE_URL=https://upload.wikimedia.org/wikipedia/commons/e/e1/FullMoon2010.jpg
#MOON_REFERENCE_DIR="$SCRIPT_DIR/moon-ref"
#MOON_REFERENCE_PATH="$MOON_REFERENCE_DIR/full-moon.jpg"
# ── Mattermost — daily sunrise upload ─────────────────────────────────────────
# mattermost_url, access_token, channel_id go in .env (see bottom of this file).