From bf09ad76c0cbcb5f3a8e2fd559b37d50889d5954 Mon Sep 17 00:00:00 2001 From: Joey Date: Fri, 28 Aug 2026 16:35:02 -0500 Subject: [PATCH] first commit --- Dockerfile | 18 + config.json | 21 ++ docker-compose.yml | 19 + recorder.py | 875 +++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + 5 files changed, 934 insertions(+) create mode 100644 Dockerfile create mode 100644 config.json create mode 100644 docker-compose.yml create mode 100644 recorder.py create mode 100644 requirements.txt diff --git a/Dockerfile b/Dockerfile new file mode 100644 index 0000000..320ff8a --- /dev/null +++ b/Dockerfile @@ -0,0 +1,18 @@ +FROM python:3.12-slim + +RUN apt-get update && apt-get install -y \ + ffmpeg \ + rsync \ + tzdata \ + && rm -rf /var/lib/apt/lists/* + +WORKDIR /app + +COPY requirements.txt . +RUN pip install --no-cache-dir -r requirements.txt + +COPY recorder.py . +COPY config.json . + +CMD ["python", "recorder.py"] + diff --git a/config.json b/config.json new file mode 100644 index 0000000..241ada8 --- /dev/null +++ b/config.json @@ -0,0 +1,21 @@ +{ + "mqtt": { + "host": "10.40.0.2", + "port": 1883 + }, + + "recording": { + "local_root": "/recordings", + "nas_root": "/nas" + }, + + "stream": { + "listen_url": "rtmp://0.0.0.0:1935/live", + "listen_key": "service", + + "publish_enabled": false, + + "publish_url": "rtmp://ingest-va.streamspot.com/ingest?7c318a95a3&9cc6fd63c0", + "publish_key": "7c318a95a3" + } +} diff --git a/docker-compose.yml b/docker-compose.yml new file mode 100644 index 0000000..b029ac4 --- /dev/null +++ b/docker-compose.yml @@ -0,0 +1,19 @@ +services: + stream-recorder: + build: . + container_name: stream-recorder + restart: unless-stopped + network_mode: host + user: "1001:1001" + group_add: + - 1000 + volumes: + - /home/stream/Recordings:/recordings:Z + - /mnt/media/Videos/recordings:/nas + - /etc/localtime:/etc/localtime:ro + environment: + MQTT_HOST: 10.40.0.2 + LOCAL_ROOT: /recordings + NAS_ROOT: /nas + PUBLISH_URL: "" + TZ: America/Chicago \ No newline at end of file diff --git a/recorder.py b/recorder.py new file mode 100644 index 0000000..ec76ae2 --- /dev/null +++ b/recorder.py @@ -0,0 +1,875 @@ +import json +import shutil +import signal +import subprocess +import threading +import time +from datetime import datetime +from pathlib import Path +from zoneinfo import ZoneInfo + +import paho.mqtt.client as mqtt + + +CONFIG = json.loads(Path("/app/config.json").read_text()) + +MQTT_HOST = CONFIG["mqtt"]["host"] +MQTT_PORT = CONFIG["mqtt"]["port"] + +LOCAL_ROOT = Path(CONFIG["recording"]["local_root"]) +NAS_ROOT = Path(CONFIG["recording"]["nas_root"]) + +publish_enabled = CONFIG["stream"]["publish_enabled"] + +RTMP_URL = ( + f'{CONFIG["stream"]["listen_url"]}/' + f'{CONFIG["stream"]["listen_key"]}' +) + +PUBLISH_URL = ( + f'{CONFIG["stream"]["publish_url"]}/' + f'{CONFIG["stream"]["publish_key"]}' +) + +LOCAL_TZ = ZoneInfo(CONFIG.get("timezone", "America/Chicago")) + +KNOWN_MARKERS = [ + "full_service", + "worship", + "offering", + "announcements", + "sermon", + "altar", + "testimony", +] + +armed = False +recording = False +running = True + +session_dir = None +recording_file = None +markers_file = None +metadata_file = None +exports_dir = None + +session_id = None +recording_started_at = None +recording_ended_at = None + +sections = [] +marker_counts = {} + +ffmpeg = None +current_status = "offline" + + +def pub(topic, payload, retain=True): + client.publish(topic, str(payload), retain=retain) + + +def set_status(status): + global current_status + + if status != current_status: + current_status = status + print(f"STATUS: {status}", flush=True) + pub("stream/recorder/status", status) + + +def now_local(): + return datetime.now(LOCAL_TZ) + + +def format_duration(seconds): + seconds = max(0, int(seconds)) + mins = seconds // 60 + secs = seconds % 60 + hours = mins // 60 + mins = mins % 60 + + if hours: + return f"{hours}:{mins:02d}:{secs:02d}" + + return f"{mins}:{secs:02d}" + +def parse_time_to_seconds(value): + parts = value.strip().split(":") + + if len(parts) == 1: + return float(parts[0]) + + if len(parts) == 2: + minutes, seconds = parts + return int(minutes) * 60 + float(seconds) + + if len(parts) == 3: + hours, minutes, seconds = parts + return int(hours) * 3600 + int(minutes) * 60 + float(seconds) + + raise ValueError("bad time") + +def parse_timecode(value): + value = str(value).strip() + + if value.replace(".", "", 1).isdigit(): + return float(value) + + parts = value.split(":") + + if len(parts) == 2: + mins, secs = parts + return int(mins) * 60 + float(secs) + + if len(parts) == 3: + hours, mins, secs = parts + return int(hours) * 3600 + int(mins) * 60 + float(secs) + + raise ValueError(f"Bad timecode: {value}") + + +def offset(): + if recording_started_at is None: + return 0 + + return round(time.time() - recording_started_at, 3) + + +def active_sections(): + return [s["name"] for s in sections if s["end"] is None] + + +def folder_name(): + now = now_local() + + weekday = now.strftime("%A").upper() + ampm = "AM" if now.hour < 12 else "PM" + stamp = now.strftime("%I-%M-%S") + + return f"{now:%Y-%m-%d}-{weekday}_{ampm}_RECORDING-{stamp}" + + +def marker_label(section): + start = format_duration(section["start"]) + end = "ACTIVE" if section["end"] is None else format_duration(section["end"]) + + return f'#{section["number"]} {start}-{end}' + + +def marker_public(section): + return { + "id": section["id"], + "name": section["name"], + "number": section["number"], + "label": marker_label(section), + "start": section["start"], + "start_text": format_duration(section["start"]), + "end": section["end"], + "end_text": "ACTIVE" if section["end"] is None else format_duration(section["end"]), + "duration": ( + round(offset() - section["start"], 3) + if section["end"] is None + else round(section["end"] - section["start"], 3) + ), + "duration_text": format_duration( + offset() - section["start"] + if section["end"] is None + else section["end"] - section["start"] + ), + } + + +def publish_marker_lists(): + groups = {} + + for section in sections: + groups.setdefault(section["name"], []).append(marker_public(section)) + + all_group_names = sorted(set(KNOWN_MARKERS) | set(groups.keys())) + + for name in all_group_names: + items = groups.get(name, []) # keep #1, #2, #3 order + + pub(f"stream/marker/list/{name}", json.dumps(items)) + pub(f"stream/marker/list/{name}/count", len(items)) + + for i in range(1, 21): + if i <= len(items): + item = items[i - 1] + pub(f"stream/marker/list/{name}/{i}/id", item["id"]) + pub(f"stream/marker/list/{name}/{i}/label", item["label"]) + else: + pub(f"stream/marker/list/{name}/{i}/id", "") + pub(f"stream/marker/list/{name}/{i}/label", "—") + + active_group_names = sorted(groups.keys()) + pub("stream/marker/groups", json.dumps(active_group_names)) + + +def save_markers(): + if not markers_file or not markers_file.parent.exists(): + return + + markers_file.write_text( + json.dumps( + { + "session_id": session_id, + "recording": "recording.mkv", + "markers_file": "markers.json", + "recording_started_at": recording_started_at, + "recording_ended_at": recording_ended_at, + "status": current_status, + "active_sections": active_sections(), + "sections": [marker_public(s) for s in sections], + }, + indent=2, + ) + ) + + +def save_metadata(): + if not metadata_file or not metadata_file.parent.exists(): + return + + duration = offset() + + metadata_file.write_text( + json.dumps( + { + "session_id": session_id, + "folder": session_dir.name if session_dir else None, + "recording": "recording.mkv", + "markers": "markers.json", + "exports": "exports", + "duration": duration, + "duration_text": format_duration(duration), + "publish_enabled": publish_enabled, + "created_at": now_local().isoformat(), + "sections": list(sorted(set(s["name"] for s in sections))), + }, + indent=2, + ) + ) + + +def clear_marker_feedbacks(): + for name in KNOWN_MARKERS: + pub(f"stream/marker/{name}/active", "false") + pub(f"stream/marker/{name}/duration", "0:00") + pub(f"stream/marker/list/{name}", "[]") + + pub("stream/marker/groups", "[]") + pub("stream/marker/active", "[]") + pub("stream/marker/active_text", "none") + pub("stream/marker/status", "ready") + + +def publish_active(): + active = [] + + for section in sections: + if section["end"] is None: + duration = offset() - section["start"] + active.append(f'{section["name"]} {format_duration(duration)}') + + pub("stream/marker/active", json.dumps(active)) + pub("stream/marker/active_text", ", ".join(active) or "none") + + +def publish_marker_durations(): + names = set(KNOWN_MARKERS) | {s["name"] for s in sections} + + for name in names: + active_for_name = [ + s for s in sections + if s["name"] == name and s["end"] is None + ] + + if active_for_name: + latest = active_for_name[-1] + duration = offset() - latest["start"] + + pub(f"stream/marker/{name}/active", "true") + pub(f"stream/marker/{name}/duration", format_duration(duration)) + else: + pub(f"stream/marker/{name}/active", "false") + + +def publish_stream_duration(): + if not recording_started_at: + pub("stream/recorder/duration_seconds", "0") + pub("stream/recorder/duration", "0:00") + return + + end_time = recording_ended_at or time.time() + duration = int(end_time - recording_started_at) + + pub("stream/recorder/duration_seconds", duration) + pub("stream/recorder/duration", format_duration(duration)) + + +def find_marker_by_id(marker_id): + marker_id = marker_id.strip().lower() + + for section in sections: + if section["id"] == marker_id: + return section + + return None + + +def refresh_marker_outputs(): + save_markers() + publish_active() + publish_marker_lists() + publish_marker_durations() + +def clear_marker_lists(): + for name in KNOWN_MARKERS: + + pub(f"stream/marker/list/{name}", "[]") + + pub(f"stream/marker/list/{name}/count", "0") + + for i in range(1, 21): + pub(f"stream/marker/list/{name}/{i}/id", "") + pub(f"stream/marker/list/{name}/{i}/label", "—") + +def start_section(name): + name = name.strip().lower() + + if not recording: + pub("stream/marker/status", f"ignored:not_recording:{name}") + return + + number = marker_counts.get(name, 0) + 1 + marker_counts[name] = number + marker_id = f"{name}-{number:03d}" + + sections.append( + { + "id": marker_id, + "name": name, + "number": number, + "start": offset(), + "end": None, + } + ) + + pub(f"stream/marker/{name}/active", "true") + pub(f"stream/marker/{name}/duration", "0:00") + pub("stream/marker/status", f"started:{marker_id}") + + refresh_marker_outputs() + + print(f"START {marker_id}", flush=True) + + +def end_section(name): + name = name.strip().lower() + + if name == "all": + end_all_sections() + return + + for section in reversed(sections): + if section["name"] == name and section["end"] is None: + section["end"] = offset() + duration = section["end"] - section["start"] + + pub(f"stream/marker/{name}/active", "false") + pub(f"stream/marker/{name}/duration", format_duration(duration)) + pub("stream/marker/status", f"ended:{section['id']}") + + refresh_marker_outputs() + + print(f"END {section['id']}", flush=True) + return + + pub("stream/marker/status", f"no_open_marker:{name}") + + +def end_all_sections(): + closed = [] + + for section in sections: + if section["end"] is None: + section["end"] = offset() + name = section["name"] + duration = section["end"] - section["start"] + + pub(f"stream/marker/{name}/active", "false") + pub(f"stream/marker/{name}/duration", format_duration(duration)) + closed.append(section["id"]) + + refresh_marker_outputs() + + pub("stream/marker/status", "ended:all") + print(f"Closed markers: {closed}", flush=True) + +def set_marker_start(payload): + # format: marker-id:15:00 + marker_id, value = payload.split(":", 1) + + marker = find_marker_by_id(marker_id) + + if not marker: + pub("stream/marker/edit/status", f"missing:{marker_id}") + return + + try: + marker["start"] = max(0, parse_time_to_seconds(value)) + except ValueError: + pub("stream/marker/edit/status", "bad_set_start") + return + + if marker["end"] is not None: + marker["end"] = max(marker["start"], marker["end"]) + + save_markers() + publish_active() + publish_marker_lists() + pub("stream/marker/edit/status", f"set_start:{marker_id}") + + +def set_marker_end(payload): + # format: marker-id:45:30 + marker_id, value = payload.split(":", 1) + + marker = find_marker_by_id(marker_id) + + if not marker: + pub("stream/marker/edit/status", f"missing:{marker_id}") + return + + try: + marker["end"] = max(marker["start"], parse_time_to_seconds(value)) + except ValueError: + pub("stream/marker/edit/status", "bad_set_end") + return + + save_markers() + publish_active() + publish_marker_lists() + pub("stream/marker/edit/status", f"set_end:{marker_id}") + +def adjust_marker(payload): + # payload format: + # marker-id:start:-60 + # marker-id:end:+30 + + parts = payload.split(":") + + if len(parts) != 3: + pub("stream/marker/edit/status", "bad_adjust") + return + + marker_id, field, amount = parts + marker_id = marker_id.strip().lower() + field = field.strip().lower() + + marker = find_marker_by_id(marker_id) + + if not marker: + pub("stream/marker/edit/status", f"missing:{marker_id}") + return + + try: + delta = float(amount) + except ValueError: + pub("stream/marker/edit/status", "bad_amount") + return + + if field == "start": + marker["start"] = max(0, marker["start"] + delta) + + if marker["end"] is not None: + marker["start"] = min(marker["start"], marker["end"]) + + elif field == "end": + if marker["end"] is None: + marker["end"] = offset() + + marker["end"] = max(marker["start"], marker["end"] + delta) + + else: + pub("stream/marker/edit/status", "bad_field") + return + + refresh_marker_outputs() + pub("stream/marker/edit/status", f"adjusted:{marker_id}") + + +def set_marker_time(payload, field): + # payload format: + # marker-id:15:00 + # marker-id:1:05:30 + # marker-id:900 + + marker_id, sep, value = payload.partition(":") + + if not sep: + pub("stream/marker/edit/status", f"bad_set_{field}") + return + + marker_id = marker_id.strip().lower() + marker = find_marker_by_id(marker_id) + + if not marker: + pub("stream/marker/edit/status", f"missing:{marker_id}") + return + + try: + seconds = parse_timecode(value) + except ValueError: + pub("stream/marker/edit/status", "bad_timecode") + return + + seconds = max(0, seconds) + + if field == "start": + marker["start"] = seconds + + if marker["end"] is not None: + marker["start"] = min(marker["start"], marker["end"]) + + elif field == "end": + marker["end"] = max(marker["start"], seconds) + + refresh_marker_outputs() + pub("stream/marker/edit/status", f"set_{field}:{marker_id}") + + +def delete_marker(marker_id): + global sections + + marker_id = marker_id.strip().lower() + marker = find_marker_by_id(marker_id) + + if not marker: + pub("stream/marker/edit/status", f"missing:{marker_id}") + return + + sections = [s for s in sections if s["id"] != marker_id] + + refresh_marker_outputs() + pub("stream/marker/edit/status", f"deleted:{marker_id}") + + +def create_session(): + global session_dir, recording_file, markers_file, metadata_file, exports_dir + global session_id, recording_started_at, recording_ended_at, sections, marker_counts + + session_id = folder_name() + + session_dir = LOCAL_ROOT / "current" / session_id + exports_dir = session_dir / "exports" + + session_dir.mkdir(parents=True, exist_ok=True) + exports_dir.mkdir(parents=True, exist_ok=True) + + recording_file = session_dir / "recording.mkv" + markers_file = session_dir / "markers.json" + metadata_file = session_dir / "metadata.json" + + recording_started_at = None + recording_ended_at = None + sections = [] + marker_counts = {} + + pub("stream/recorder/session", session_id) + pub("stream/recorder/folder", session_id) + pub("stream/recorder/file", "recording.mkv") + pub("stream/recorder/markers_file", "markers.json") + + clear_marker_feedbacks() + clear_marker_lists() + save_markers() + save_metadata() + publish_marker_lists() + + +def ffmpeg_args(): + if publish_enabled and PUBLISH_URL: + return [ + "ffmpeg", + "-listen", "1", + "-i", RTMP_URL, + "-map", "0", + "-c", "copy", + "-f", "tee", + f"[f=matroska]{recording_file}|[f=flv]{PUBLISH_URL}", + ] + + return [ + "ffmpeg", + "-listen", "1", + "-i", RTMP_URL, + "-c", "copy", + "-f", "matroska", + str(recording_file), + ] + + +def export_clips(): + if not recording_file or not recording_file.exists(): + return + + for section in sections: + if section["end"] is None: + continue + + name = section["name"] + marker_id = section["id"] + start = section["start"] + duration = section["end"] - section["start"] + + if duration <= 1: + continue + + output = exports_dir / f"{marker_id}.mp4" + + cmd = [ + "ffmpeg", + "-y", + "-ss", str(start), + "-i", str(recording_file), + "-t", str(duration), + "-c", "copy", + str(output), + ] + + print(f"Exporting {output.name}", flush=True) + subprocess.run(cmd) + + +def move_to_nas(): + if not session_dir or not session_dir.exists(): + return + + if not NAS_ROOT.exists(): + print("NAS path missing; leaving local copy", flush=True) + pub("stream/recorder/nas_status", "missing") + return + + target = NAS_ROOT / session_dir.name + + print(f"Moving to NAS: {target}", flush=True) + + result = subprocess.run( + [ + "rsync", + "-a", + f"{session_dir}/", + f"{target}/", + ] + ) + + if result.returncode != 0: + print("NAS move failed; leaving local copy", flush=True) + pub("stream/recorder/nas_status", "failed") + return + + try: + shutil.rmtree(session_dir) + except Exception as e: + print(f"Local cleanup failed: {e}", flush=True) + + pub("stream/recorder/nas_status", "moved") + pub("stream/recorder/nas_folder", str(target)) + + +def reset_session(): + global session_dir, recording_file, markers_file, metadata_file, exports_dir + global session_id, recording_started_at, recording_ended_at, sections, marker_counts + + session_dir = None + recording_file = None + markers_file = None + metadata_file = None + exports_dir = None + + session_id = None + recording_started_at = None + recording_ended_at = None + + sections = [] + marker_counts = {} + + marker_counts.clear() + clear_marker_lists() + + +def finalize_session(): + set_status("finalizing") + time.sleep(5) + + end_all_sections() + save_markers() + save_metadata() + + set_status("exporting") + export_clips() + save_metadata() + + move_to_nas() + + clear_marker_feedbacks() + set_status("complete") + + reset_session() + publish_stream_duration() + + time.sleep(2) + set_status("disarmed") + + +def run_listener(): + global ffmpeg, recording, recording_started_at, recording_ended_at, armed + + create_session() + set_status("armed") + + ffmpeg = subprocess.Popen( + ffmpeg_args(), + stderr=subprocess.PIPE, + stdout=subprocess.DEVNULL, + ) + + for raw in ffmpeg.stderr: + line = raw.decode(errors="ignore").strip() + + if "Opening" in line or "Input #0" in line or "time=" in line: + if recording_started_at is None: + recording_started_at = time.time() + recording = True + set_status("recording") + save_markers() + save_metadata() + + if "time=" in line: + pub("stream/recorder/heartbeat", int(time.time()), retain=False) + + if line: + print(line, flush=True) + + code = ffmpeg.wait() + + if recording: + recording = False + recording_ended_at = time.time() + finalize_session() + else: + if code != 0: + set_status("error") + else: + set_status("disarmed") + + armed = False + + +def arm(): + global armed + + if armed or recording: + return + + armed = True + set_status("armed") + + threading.Thread(target=run_listener, daemon=True).start() + + +def disarm(): + global armed + + armed = False + + if ffmpeg and ffmpeg.poll() is None: + ffmpeg.terminate() + + set_status("disarmed") + + +def on_connect(client, userdata, flags, reason_code, properties=None): + client.subscribe("stream/recorder/cmd") + client.subscribe("stream/publish/cmd") + client.subscribe("stream/marker/start") + client.subscribe("stream/marker/end") + + client.subscribe("stream/marker/edit/adjust") + client.subscribe("stream/marker/edit/delete") + client.subscribe("stream/marker/edit/set_start") + client.subscribe("stream/marker/edit/set_end") + + pub("stream/publish/enabled", str(publish_enabled).lower()) + clear_marker_feedbacks() + publish_marker_lists() + set_status("disarmed") + + +def on_message(client, userdata, msg): + global publish_enabled + + payload = msg.payload.decode().strip().lower() + + if msg.topic == "stream/recorder/cmd": + if payload == "arm": + arm() + elif payload == "disarm": + disarm() + + elif msg.topic == "stream/publish/cmd": + if payload in ["on", "true", "1", "yes"]: + publish_enabled = True + elif payload in ["off", "false", "0", "no"]: + publish_enabled = False + + pub("stream/publish/enabled", str(publish_enabled).lower()) + + elif msg.topic == "stream/marker/start": + start_section(payload) + + elif msg.topic == "stream/marker/end": + end_section(payload) + + elif msg.topic == "stream/marker/edit/adjust": + adjust_marker(payload) + + elif msg.topic == "stream/marker/edit/delete": + delete_marker(payload) + + elif msg.topic == "stream/marker/edit/set_start": + set_marker_time(payload, "start") + + elif msg.topic == "stream/marker/edit/set_end": + set_marker_time(payload, "end") + +def shutdown(*args): + global running + + running = False + disarm() + client.loop_stop() + + +signal.signal(signal.SIGINT, shutdown) +signal.signal(signal.SIGTERM, shutdown) + +client = mqtt.Client(mqtt.CallbackAPIVersion.VERSION2) +client.on_connect = on_connect +client.on_message = on_message +client.connect(MQTT_HOST, MQTT_PORT, 60) +client.loop_start() + +set_status("disarmed") +clear_marker_feedbacks() +publish_marker_lists() + +print("Stream recorder ready", flush=True) + +while running: + publish_stream_duration() + publish_marker_durations() + save_markers() + time.sleep(1) diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..8579e8b --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +paho-mqtt