first commit
This commit is contained in:
+875
@@ -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)
|
||||
Reference in New Issue
Block a user