Merge pull request #57 from outis1one/clDemoe/digital-zoom-moon-capture-DprtQ

Add nightly moon-track timelapse and monthly moon-phase composites
This commit is contained in:
Outis
2026-04-30 18:02:04 -04:00
committed by GitHub
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).