8000 irl4you-pip/pipbox_send.py at v1.0.71 · IRL4YOU/irl4you-pip · GitHub
[go: up one dir, main page]
Skip to content

pipbox_send.py

Latest commit

 

History

History
936 lines (872 loc) · 38 KB

pipbox_send.py

File metadata and controls

936 lines (872 loc) · 38 KB
Raw
#!/usr/bin/env python3
import collections
import json
import os
import re
import shutil
import signal
import stat
import subprocess
import sys
import threading
import time
sys.path.insert(0, "/opt/pipbox")
import server
import pipbox_always
import pipbox_live
try:
import pipbox_watch
except ImportError:
pipbox_watch = None
STATE = "/var/lib/pipbox"
BELACODER = "/opt/pipbox/bin/belacoder"
RUN = "/run/pipbox-send"
WORK = "/var/tmp/pipbox"
STATUS = f"{RUN}/status.json"
BC_STATS = f"{RUN}/belacoder-stats.json"
EVENTS = f"{RUN}/belacoder-events.txt"
EVENT_RX = re.compile(r"^(Aligned |Re-aligned |Aligning |Feed sbf[0-9]|Dynamic feed |Controller parts|Bitrate controller|control: no element)")
STATS = f"{RUN}/belacoder-stats.txt"
STATS_KEEP = 3000
LISTEN_PORT = 9100
VIEW_LIVE = False
AUDIO_LIVE = False
SOFT_START_KBPS = 0
SOFT_FPS_MIN, SOFT_FPS_MAX, SOFT_STEADY = 15.0, 65.0, 1.25
SOFT_STABLE_S = 3
SOFT_FLOOR_S = 20.0
SOFT_FILE_WAIT_S = 90.0
SOFT_DATA_WAIT_S = 60.0
BC_MIN_BPS = 300000
DELAY_LIVE = False
DELAY_LIVE_PIPS = False
SWAP_BASE = None
PLUGIN_DIR = "/opt/pipbox/gst"
DOWN_S = 5
UP_S = 60
UP_FIRST_S = 5
NOTABLE = (
(re.compile(r"Failed to establish an SRT connection"), "Verbindung zum SRT-Server fehlgeschlagen, neuer Versuch"),
(re.compile(r"The SRT connection.*exiting"), "Verbindung zum Server kurz unterbrochen, wird automatisch neu aufgebaut"),
(re.compile(r"Pipeline stall detected \(output\)"), "Der Ausgang stockte (der Encoder lieferte nichts mehr), er wird neu gestartet"),
(re.compile(r"Pipeline stall detected"), "Das Eingangsbild stockte, der Encoder wird neu gestartet"),
(re.compile(r"Failed to establish any initial connections"), "Keine Verbindung zum SRTLA-Server, neuer Versuch"),
(re.compile(r"no available connections"), "Alle Sendewege waren ausgefallen, Verbindung wird neu aufgebaut"),
(re.compile(r"gstreamer error from (\w+)"), "Fehler im Bildaufbau (Kamera oder Encoder)"),
)
class Refuse(Exception):
pass
def exit_text(rc):
if rc is not None and rc < 0:
try:
return f"Code {rc}, Signal {signal.Signals(-rc).name}"
except ValueError:
pass
return f"Code {rc}"
def belacoder_running():
for d in os.listdir("/proc"):
if d.isdigit():
try:
with open(f"/proc/{d}/comm") as f:
if f.read().strip() == "belacoder":
return True
except OSError:
pass
return False
def load_json(path):
try:
with open(path) as f:
return json.load(f)
except (OSError, ValueError):
return {}
def stream_live(key):
import urllib.request
try:
x = urllib.request.urlopen("http://127.0.0.1:1936/", timeout=3).read().decode()
except OSError:
return None
for m in re.finditer(r"<stream>.*?</stream>", x, re.S):
if re.search(rf"<name>\s*{re.escape(key)}\s*</name>", m.group(0)) and "<publishing/>" in m.group(0):
return True
return False
def live_keys():
import urllib.request
try:
x = urllib.request.urlopen("http://127.0.0.1:1936/", timeout=3).read().decode()
except OSError:
return None
out = set()
for m in re.finditer(r"<stream>.*?</stream>", x, re.S):
n = re.search(r"<name>([^<]+)</name>", m.group(0))
if n and "<publishing/>" in m.group(0):
out.add(n.group(1).strip())
return out
def configured_keys(cfg):
ks = [cfg.get("main", "")]
if cfg.get("type") == "pip":
ks += [cfg.get("pip", ""), cfg.get("pip2", ""), cfg.get("pip3", "")]
out = []
for k in ks:
if k and k not in out:
out.append(k)
return out
def effective_cfg(cfg, use):
ks = [k for k in configured_keys(cfg) if k in use]
if not ks:
return None, ()
ina = cfg.get("inactive")
off = {k for k in ina if isinstance(k, str)} if isinstance(ina, list) else set()
main = next((k for k in ks if k == cfg.get("main") or k not in off), None)
if main is None:
return None, ()
ks = [main] + [k for k in ks if k != main]
delay = {}
for key, name in (("main", "main_delay_ms"), ("pip", "pip_delay_ms"), ("pip2", "pip2_delay_ms"), ("pip3", "pip3_delay_ms")):
if cfg.get(key):
delay[cfg[key]] = cfg.get(name, 0)
main, smalls = ks[0], ks[1:4]
out = dict(cfg)
out.update(main=main, pip=smalls[0] if smalls else "", pip2=smalls[1] if len(smalls) > 1 else "", pip3=smalls[2] if len(smalls) > 2 else "",
type="pip" if smalls else "single", main_delay_ms=delay.get(main, 0),
pip_delay_ms=delay.get(smalls[0], 0) if smalls else 0,
pip2_delay_ms=delay.get(smalls[1], 0) if len(smalls) > 1 else 0,
pip3_delay_ms=delay.get(smalls[2], 0) if len(smalls) > 2 else 0)
if smalls:
audio_key = cfg.get(cfg["audio"]) if cfg.get("audio") in ("pip", "pip2", "pip3") else cfg.get("main")
names = ("pip", "pip2", "pip3")
out["audio"] = next((names[i] for i, k in enumerate(smalls) if k == audio_key), "main")
return out, tuple(ks[:4])
class Failover:
def __init__(self, cfg, layout):
self.cfg, self.keys = cfg, configured_keys(cfg)
self.layout = tuple(layout)
self.up, self.down = {}, {}
def wait_left(self, now):
need = UP_S if self.layout else UP_FIRST_S
return {k: max(0, int(need - (now - t) + 0.999)) for k, t in self.up.items() if k not in self.layout}
def step(self, now, live, gone=False):
if live is None:
return None
for k in self.keys:
if k in live:
self.up.setdefault(k, now)
self.down.pop(k, None)
else:
self.down.setdefault(k, now)
self.up.pop(k, None)
need = UP_S if self.layout else UP_FIRST_S
use = set()
for k in self.keys:
if k in self.layout:
if k in live or (not gone and now - self.down[k] < DOWN_S):
use.add(k)
elif k in live and now - self.up[k] >= need:
use.add(k)
new = effective_cfg(self.cfg, use)[1]
if new != self.layout:
self.layout = new
return new
return None
def ensure_work():
try:
st = os.lstat(WORK)
if stat.S_ISLNK(st.st_mode) or not stat.S_ISDIR(st.st_mode) or st.st_uid != os.geteuid():
if stat.S_ISDIR(st.st_mode) and not stat.S_ISLNK(st.st_mode):
shutil.rmtree(WORK)
else:
os.unlink(WORK)
except FileNotFoundError:
pass
os.makedirs(WORK, mode=0o700, exist_ok=True)
os.chmod(WORK, 0o700)
def put_state_file(name, text):
tmp = f"{STATE}/{name}.tmp"
try:
os.unlink(tmp)
except OSError:
pass
fd = os.open(tmp, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o644)
with os.fdopen(fd, "w") as f:
f.write(text)
os.replace(tmp, f"{STATE}/{name}")
def write_pipeline(cfg):
text = server.PipelineStore(os.devnull).build(cfg)
if not text:
return ""
ensure_work()
with open(f"{WORK}/pipeline", "w") as f:
f.write(text)
try:
vals = server.PipelineStore.delay_values(cfg)
put_state_file("main-delay-ms", " ".join(map(str, vals)) + "\n")
except OSError:
pass
global DELAY_LIVE, DELAY_LIVE_PIPS, SWAP_BASE, VIEW_LIVE, AUDIO_LIVE
DELAY_LIVE = "pbctl" in text
DELAY_LIVE_PIPS = "pip-queue=" in text or "cam1=" in text
plan = (server.PipelineStore.always_plan(cfg) or server.PipelineStore.swap_plan(cfg)) if "pbpipsel" in text else None
SWAP_BASE = {"cams": plan["cams"], "group": plan["group"]} if plan else None
try:
os.unlink(server.SWAP_STATE)
except OSError:
pass
try:
if plan:
put_state_file(server.SWAP_SELECT, plan["line"] + "\n")
except OSError:
pass
VIEW_LIVE = "name=avol" in text and "pbctl" in text
AUDIO_LIVE = VIEW_LIVE and "pbpipsel name=asel" in text
if "name=sbf0_src" in text:
DELAY_LIVE = DELAY_LIVE_PIPS = VIEW_LIVE = AUDIO_LIVE = True
SWAP_BASE = None
try:
os.unlink(server.VIEW_STATE)
except OSError:
pass
try:
if VIEW_LIVE:
mute = 0
try:
parts = open(f"{STATE}/{server.VIEW_FILE}").read().split()
mute = 1 if len(parts) >= 3 and parts[2] == "1" else 0
except OSError:
pass
put_state_file(server.VIEW_FILE, server.PipelineStore.view_line(cfg, mute) + "\n")
except OSError:
pass
return text
def prepare():
srt = load_json(f"{STATE}/srtla.json")
pipe = load_json(f"{STATE}/pipeline.json")
servers = srt.get("servers", [])
cur = next((s for s in servers if s.get("id") == srt.get("selected")), None)
if not cur:
raise Refuse("Kein SRTLA-Server ausgewählt")
try:
sv = server.SrtlaStore.check(cur)
except ValueError as e:
raise Refuse(f"SRTLA-Server ungültig: {e}")
st = {**server.SrtlaStore.DEFAULT_SETTINGS, **srt.get("settings", {})}
try:
mn, mx, lat = int(st["min_kbps"]), int(st["max_kbps"]), int(st["latency_ms"])
except (TypeError, ValueError):
raise Refuse("Sendeeinstellungen ungültig")
if not (100 <= mn < mx <= 20000 and 100 <= lat <= 10000):
raise Refuse("Sendeeinstellungen außerhalb des erlaubten Bereichs")
if not pipe.get("main"):
raise Refuse("Kein Bildaufbau gespeichert (Hauptkamera fehlt)")
cfg = {**server.PipelineStore.DEFAULT, **pipe}
if cfg["type"] == "pip" and not os.path.exists(f"{PLUGIN_DIR}/libgstpbpip.so"):
raise Refuse("Bild-in-Bild braucht den Überlagerungs-Baustein, der noch nicht installiert ist")
if not server.PipelineStore(os.devnull).build(cfg):
raise Refuse("Der Bildaufbau konnte nicht erzeugt werden (Kamera-Schlüssel ungültig)")
keys = configured_keys(cfg)
always = bool(server.PipelineStore.always_plan(cfg))
note = server.PipelineStore.always_blocker(cfg)
if note:
print(f"send: {note}", flush=True)
auto = bool(cfg.get("auto_failover", True)) and len(keys) > 1 and not always
live = live_keys() if (auto or always) else None
if always:
if live is not None and not (set(keys) & live) and not pipbox_live.enabled(server.PipelineStore._safe_cfg(cfg)):
raise Refuse("Keine Kamera sendet gerade an die Box")
auto = False
if always:
eff, used = cfg, tuple(keys)
elif auto and live is not None:
eff, used = effective_cfg(cfg, live)
if eff is None:
raise Refuse("Es sendet nur eine deaktivierte Kamera an die Box" if live else "Keine Kamera sendet gerade an die Box")
else:
eff, used = cfg, tuple(keys)
for key in keys:
if not auto and stream_live(key) is False:
raise Refuse(f"Die Kamera {key} sendet gerade nicht an die Box")
have = {o["iface"]: o["ip"] for o in server.iface_ips()}
ips = [have[u] for u in st.get("uplinks", []) if u in have]
if not ips:
raise Refuse("Keiner der gewählten Sendewege hat eine IP-Adresse")
ensure_work()
if not write_pipeline(eff):
raise Refuse("Der Bildaufbau konnte nicht erzeugt werden (Kamera-Schlüssel ungültig)")
with open(f"{WORK}/bitrate", "w") as f:
f.write(f"{mn * 1000}\n{mx * 1000}")
with open(f"{WORK}/ips", "w") as f:
f.write("\n".join(ips) + "\n")
return sv, lat, mn, mx, ips, {"cfg": cfg, "layout": used, "auto": auto, "always": always, "always_note": note,
"spread": "all" if st.get("spread") == "all" else "best",
"sig": server.srtla_signature(srt)}
class Sender:
def __init__(self, sv, lat, ips, plan=None):
self.sv, self.lat, self.ips = sv, lat, ips
self.bitrate_full = None
self.soft_lock = threading.Lock()
plan = plan or {"cfg": {}, "layout": (), "auto": False}
self.plan = plan
self.fo = Failover(plan["cfg"], plan["layout"]) if plan["auto"] else None
self.layout = tuple(plan["layout"])
self.always = None
self.live = bool(plan.get("always") and plan["cfg"] and pipbox_live.enabled(server.PipelineStore._safe_cfg(plan["cfg"])))
self.boost = pipbox_live.CpuBoost(log=lambda m: print(m, flush=True))
if self.live:
self.always = pipbox_live.LiveController(
load_json=lambda name: load_json(f"{STATE}/{name}"), put_state_file=put_state_file, state_dir=STATE,
safe_cfg=server.PipelineStore._safe_cfg, view_values=server.PipelineStore.view_values, log=lambda m: print(m, flush=True),
start_cfg=plan["cfg"])
elif plan.get("always"):
self.always = pipbox_always.Controller(
load_json=lambda name: load_json(f"{STATE}/{name}"), put_state_file=put_state_file,
delay_values=server.PipelineStore.delay_values, cam_live_path=server.CAM_LIVE,
select_name=server.SWAP_SELECT, delay_name="main-delay-ms",
feeders=pipbox_always.Feeders(log=lambda m: print(m, flush=True)), log=lambda m: print(m, flush=True))
self.always_state = {}
self.waiting = False
self.senv = None
self._ups_t = 0
self._cfg_mt = None
self._inact_mt = None
self.stats = collections.deque(maxlen=STATS_KEEP)
self.events = collections.deque(maxlen=120)
self.links = collections.deque(maxlen=300)
self.watch = pipbox_watch.HangWatch() if pipbox_watch else None
self.hang_n = 0
self._hang_t = 0.0
self.stop_ev = threading.Event()
self.procs = {}
self.lock = threading.Lock()
self.restarts = {"srtla_send": 0, "belacoder": 0}
self.last = ""
self.last_src = ""
self.last_at = 0
self.since = int(time.time())
self.state = "starting"
def note(self, line):
for rx, msg in NOTABLE:
m = rx.search(line)
if m:
with self.lock:
fresh = msg != self.last or time.time() - self.last_at > 10
self.last = msg
self.last_at = time.time()
self.last_src = m.group(1) if m.groups() else ""
if fresh:
print(f"send: {msg}", flush=True)
return
@staticmethod
def soft_rate(mn, mx):
if SOFT_START_KBPS <= 0 or int(mn) < BC_MIN_BPS:
return None
start = max(int(mn), SOFT_START_KBPS * 1000)
return start if start < int(mx) else None
def soft_start_begin(self):
try:
with self.soft_lock:
with open(f"{WORK}/bitrate") as f:
a, b = f.read().split()[:2]
if self.bitrate_full is None:
self.bitrate_full = (int(a), int(b))
mn, mx = self.bitrate_full
start = self.soft_rate(mn, mx)
if start is None:
return None
self.write_bitrate(start, start)
return start
except (OSError, ValueError):
return None
@staticmethod
def write_bitrate(mn, mx):
tmp = f"{WORK}/bitrate.tmp"
with open(tmp, "w") as f:
f.write(f"{int(mn)}\n{int(mx)}")
os.replace(tmp, f"{WORK}/bitrate")
@staticmethod
def hup_ready(p):
pid = getattr(p, "pid", None)
if not pid:
return True
try:
with open(f"/proc/{pid}/status") as f:
for line in f:
key, _, val = line.partition(":")
if key.lower() == "sigcgt":
return bool(int(val.split()[0], 16) & 1)
except (OSError, ValueError, IndexError):
pass
return True
def bc_stats(self, t0):
try:
with open(BC_STATS) as f:
d = json.load(f)
return d if isinstance(d, dict) and int(d.get("time") or 0) >= int(t0) else None
except (OSError, ValueError, TypeError, OverflowError):
return None
def soft_alive(self, p):
return p.poll() is None and self.procs.get("belacoder") is p and not self.stop_ev.is_set()
@staticmethod
def soft_steady(win):
return len(win) >= SOFT_STABLE_S and min(win) >= SOFT_FPS_MIN and max(win) <= SOFT_FPS_MAX and max(win) <= SOFT_STEADY * min(win)
def soft_start_wait(self, p, t0):
seen, win, pre, post = None, [], 0, 0
while self.soft_alive(p):
d = self.bc_stats(t0)
if d is None:
pre += 1
if pre * 0.5 >= SOFT_FILE_WAIT_S:
return
else:
post += 1
if post * 0.5 >= SOFT_DATA_WAIT_S:
return
t = d.get("time")
if t != seen:
win = win[-(SOFT_STABLE_S - 1):] if (seen is not None and isinstance(t, (int, float)) and 0 < t - seen <= 2) else []
seen = t
try:
win.append(float(d.get("fps") or 0))
except (TypeError, ValueError):
win = []
if self.soft_steady(win):
return
if self.stop_ev.wait(0.5):
return
def soft_start_run(self, p, start, t0):
try:
self.soft_start_wait(p, t0)
if not self.soft_alive(p):
return
self.soft_start_phase(p, 1, start)
if self.stop_ev.wait(SOFT_FLOOR_S) or not self.soft_alive(p):
return
self.soft_start_phase(p, 2, start)
except Exception as e:
print(f"send: Weicher Start: Fehler im Zeitplan ({type(e).__name__}), die Bitrate wird freigegeben", flush=True)
try:
self.soft_start_phase(p, 1, start)
self.soft_start_phase(p, 2, start)
except Exception:
pass
def soft_start_phase(self, p, phase, start):
mn, mx = self.bitrate_full or (None, None)
if mx is None:
return
try:
waited, ready = 0.0, self.hup_ready(p)
while p.poll() is None and not ready and waited < 20.0 and not self.stop_ev.is_set():
self.stop_ev.wait(0.5)
waited += 0.5
ready = self.hup_ready(p)
with self.soft_lock:
if not self.soft_alive(p):
return
self.write_bitrate(start if phase == 1 else mn, mx)
if ready:
p.send_signal(signal.SIGHUP)
if phase == 1:
print(f"send: Weicher Start: Bitrate darf steigen (Untergrenze {start // 1000} kbit/s)", flush=True)
else:
print(f"send: Weicher Start beendet, Bitrate {mn // 1000} bis {mx // 1000} kbit/s", flush=True)
else:
print("send: Weicher Start: belacoder kennt das Signal nach 20 s noch nicht, die Bitrate bleibt bis zum nächsten Start bei der Startbitrate", flush=True)
except (OSError, ValueError):
pass
def spawn(self, name, args, env=None):
soft = self.soft_start_begin() if name == "belacoder" else None
t0 = time.time()
p = subprocess.Popen(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
errors="replace", env=env, restore_signals=(name != "belacoder"))
with self.lock:
self.procs[name] = p
if soft:
threading.Thread(target=self.soft_start_run, args=(p, soft, t0), daemon=True).start()
if name == "belacoder" and self.watch is not None:
self.watch.started_now()
if name == "belacoder" and self.live and self.always is not None:
self.always.on_restart()
threading.Thread(target=self.pump, args=(name, p), daemon=True).start()
return p
def pump(self, name, p):
dbg = open(os.environ["PIPBOX_BC_LOG"], "a") if os.environ.get("PIPBOX_BC_LOG") and name == "belacoder" else None
for line in p.stdout:
if dbg and not line.startswith("b:"):
dbg.write(time.strftime("%H:%M:%S ") + line[:300].rstrip() + "\n")
dbg.flush()
self.note(line)
if name == "srtla_send" and line.startswith("links:"):
self.links.append((time.strftime("%H:%M:%S", time.gmtime()), line.strip()[:200]))
continue
if name == "belacoder":
ev = line.strip()
if EVENT_RX.match(ev):
self.events.append((time.strftime("%H:%M:%S", time.gmtime()), ev[:200]))
t = line.strip()
if t.startswith("bs:"):
self._dbg_n = (getattr(self, "_dbg_n", 0) + 1) % 5
if self._dbg_n:
continue
elif not (t.startswith("set bitrate") or re.match(r"^b:\s*-?\d+/", t)):
continue
self.stats.append((time.strftime("%H:%M:%S", time.gmtime()), t[:200]))
def flush_stats(self):
try:
os.makedirs(RUN, exist_ok=True)
for path, rows in ((STATS, self.stats), (f"{RUN}/srtla-links.txt", self.links), (EVENTS, self.events)):
tmp = path + ".tmp"
with open(tmp, "w") as f:
f.write("".join(f"{t} {l}\n" for t, l in list(rows)))
os.chmod(tmp, 0o644)
os.replace(tmp, path)
except OSError:
pass
def args(self, name):
sv = self.sv
if name == "srtla_send":
return ["srtla_send", str(LISTEN_PORT), sv["host"], str(sv["port"]), f"{WORK}/ips"]
pre = []
if self.live:
big = pipbox_live.big_cpus()
if big and shutil.which("taskset"):
pre = ["taskset", "-c", ",".join(map(str, big))]
a = pre + ["stdbuf", "-oL", "-eL", BELACODER if os.access(BELACODER, os.X_OK) else "belacoder", f"{WORK}/pipeline", "127.0.0.1", str(LISTEN_PORT), "-d", "0",
"-b", f"{WORK}/bitrate", "-l", str(self.lat)]
if self.live:
a += ["-C", pipbox_live.CTL_FIFO, "-S", pipbox_live.LIVE_STATS, "-A", str(pipbox_live.buffer_ms(server.PipelineStore._safe_cfg(self.plan["cfg"])))]
if sv.get("streamid"):
a += ["-s", sv["streamid"]]
return a
def write_status(self, extra=None):
os.makedirs(RUN, exist_ok=True)
with self.lock:
data = {"state": self.state, "since": self.since, "server": self.sv["name"],
"restarts": dict(self.restarts), "last": self.last, "last_age": int(time.time() - self.last_at) if self.last_at else None, "time": int(time.time()),
"delay_live": DELAY_LIVE, "delay_live_pips": DELAY_LIVE_PIPS, "swap": SWAP_BASE,
"view_live": VIEW_LIVE, "audio_live": AUDIO_LIVE,
"applied": self.plan.get("sig"), "always_note": self.plan.get("always_note"), "hang_restarts": self.hang_n,
"buffer_ms": pipbox_live.buffer_ms(server.PipelineStore._safe_cfg(self.plan["cfg"])) if self.live else None}
if self.always is not None:
keys = configured_keys(self.plan["cfg"])
st = self.always_state or {}
fps_slots = pipbox_live.read_fps() if self.live else {}
slots = list(st.get("slots", []))
data["failover"] = {"auto": True, "always": True, "layout": [k for k in slots if k], "configured": keys,
"inactive": list(self.plan["cfg"].get("inactive") or []), "waiting": False, "degraded": False, "wait": {},
"live": st.get("alive", []), "main": st.get("main"),
"fps": {k: fps_slots[i] for i, k in enumerate(slots) if k and i in fps_slots}}
data["always"] = True
if self.fo is not None:
keys = configured_keys(self.plan["cfg"])
data["failover"] = {"auto": True, "layout": list(self.layout), "configured": keys, "inactive": list(self.plan["cfg"].get("inactive") or []),
"waiting": self.waiting, "degraded": self.waiting or set(self.layout) != set(keys),
"wait": self.fo.wait_left(time.time())}
if extra:
data.update(extra)
tmp = STATUS + ".tmp"
with open(tmp, "w") as f:
json.dump(data, f)
os.chmod(tmp, 0o644)
os.replace(tmp, STATUS)
def stop_proc(self, name):
p = self.procs.get(name)
if p is not None and p.poll() is None:
p.terminate()
deadline = time.time() + 5
while p.poll() is None and time.time() < deadline:
time.sleep(0.1)
if p.poll() is None:
p.kill()
self.procs[name] = None
def switch(self, layout, env):
eff, used = effective_cfg(self.plan["cfg"], set(layout))
if eff is None:
self.stop_proc("belacoder")
self.waiting, self.layout = True, ()
print("send: keine Kamera sendet, warte auf ein Signal", flush=True)
return
if not write_pipeline(eff):
print("send: Umschaltung abgebrochen (Pipeline nicht erzeugbar)", flush=True)
return
self.stop_proc("belacoder")
self.waiting, self.layout = False, used
self.spawn("belacoder", self.args("belacoder"), env=env)
with self.lock:
self.state = "running"
print(f"send: umgeschaltet auf {len(used)} Kamera(s): {', '.join(used)}", flush=True)
def encoder_died(self, env):
if self.fo is None:
return False
try:
live = live_keys()
if live is not None:
print(f"send: Kameras laut RTMP-Server: {len(live & set(self.fo.keys))} von {len(self.fo.keys)} senden", flush=True)
new = self.fo.step(time.time(), live, gone=True)
if new is None:
return False
self.switch(new, env)
except Exception as e:
print(f"send: Sofort-Umschaltung nicht möglich ({type(e).__name__})", flush=True)
return False
with self.lock:
self.state = "running"
return True
def run(self):
env = dict(os.environ, GST_PLUGIN_PATH=PLUGIN_DIR)
env.setdefault("SB_CTL", "D")
env["BELACODER_STATS_FILE"] = BC_STATS
if self.live:
env["GST_MPP_NO_RGA"] = "1"
self.boost.off()
self.boost.on()
senv = None
if self.plan.get("spread") == "all":
senv = dict(os.environ, SRTLA_LAT_MARGIN_MS="300", SRTLA_MIN_SHARE_PCT="10")
self.senv = senv
if os.environ.get("PIPBOX_NO_SRTLA") == "1":
self.write_status()
else:
self.spawn("srtla_send", self.args("srtla_send"), env=senv)
self.write_status()
self.wait_links_ready()
if self.always is not None:
self.always.start()
self.sync_swap_base()
self.spawn("belacoder", self.args("belacoder"), env=env)
with self.lock:
self.state = "running"
print("send: Sendekette gestartet" + (" (alle Kameras immer bereit)" if self.always is not None else ""), flush=True)
while not self.stop_ev.is_set():
if self.fo is not None:
try:
new = self.fo.step(time.time(), live_keys())
if new is not None:
self.switch(new, env)
except Exception as e:
print(f"send: automatische Umschaltung abgeschaltet ({type(e).__name__})", flush=True)
self.fo = None
if self.procs.get("belacoder") is None and not self.waiting:
try:
self.spawn("belacoder", self.args("belacoder"), env=env)
except OSError:
pass
for name in ("srtla_send", "belacoder"):
p = self.procs.get(name)
if p is not None and p.poll() is not None:
with self.lock:
self.restarts[name] += 1
n = self.restarts[name]
self.state = "restarting"
why = ""
with self.lock:
if name == "belacoder" and self.last_at and time.time() - self.last_at < 15:
why = f", zuletzt: {self.last}" + (f" ({self.last_src})" if self.last_src else "")
print(f"send: {name} beendet ({exit_text(p.returncode)}), Neustart {n}{why}", flush=True)
if name == "belacoder" and self.encoder_died(env):
continue
if self.stop_ev.wait(2):
break
self.spawn(name, self.args(name), env=env if name == "belacoder" else self.senv)
with self.lock:
self.state = "running"
try:
self.sync_cfg()
self.refresh_inactive()
except Exception as e:
print(f"send: Einstellung konnte nicht nachgeführt werden ({type(e).__name__})", flush=True)
try:
if time.monotonic() - self._ups_t >= 6:
self._ups_t = time.monotonic()
self.refresh_uplinks()
except Exception as e:
print(f"send: Netze konnten nicht neu gelesen werden ({type(e).__name__})", flush=True)
if self.always is None:
self.write_status()
self.flush_stats()
self.stop_ev.wait(2)
else:
try:
self.always_state = self.always.tick()
self.sync_swap_base()
except Exception as e:
print(f"send: Kameraauswahl: Fehler {type(e).__name__}", flush=True)
try:
self.hang_check(env)
except Exception as e:
print(f"send: Wächter: Fehler {type(e).__name__}", flush=True)
if time.monotonic() - getattr(self, "_slow_t", 0) >= 2:
self._slow_t = time.monotonic()
self.write_status()
self.flush_stats()
self.stop_ev.wait(0.5)
self.shutdown()
def hang_check(self, env):
if self.watch is None or not (self.live and self.always is not None) or time.monotonic() - self._hang_t < 1.0:
return
self._hang_t = time.monotonic()
p = self.procs.get("belacoder")
if p is None or p.poll() is not None:
return
try:
mt = os.stat(pipbox_live.LIVE_STATS).st_mtime
except OSError:
mt = None
try:
with open(pipbox_live.LIVE_STATS, errors="replace") as f:
text = f.read(12000)
except OSError:
text = ""
a = self.always
expected = {}
for slot, key in enumerate(a.binder.slot_key):
info = a.pub.get(key) if key else None
if info and info.get("w", 0) > 0 and a.states.get(slot) == 3 and not (key in a.inactive and key != a.desired[0]):
expected[slot] = True
hit = self.watch.check(mt, text.split("\n", 1)[0], expected, pipbox_watch.d_state_threads(p.pid))
if hit is None:
return
soft = self.watch.soft_ok(hit)
if not soft and not self.watch.allow():
return
slots = []
for slot, key in enumerate(a.binder.slot_key):
info = a.pub.get(key) if key else None
slots.append((slot, ("Bild %dx%d, Ton %s" % (info.get("w", 0), info.get("h", 0), "ja" if info.get("audio") else "nein")) if info else "sendet nicht", a.states.get(slot, "?")))
report = pipbox_watch.build_report(hit, p.pid, time.monotonic() - (self.watch.started or time.monotonic()), text,
[f"{t} {e}" for t, e in list(self.events)], slots)
saved = pipbox_watch.write_report(STATE, report)
if soft:
self.watch.soft_done(hit["slot"])
print(f"send: Hänger erkannt ({hit['text']}); Diagnosepaket {'gesichert (hang-diagnose.txt)' if saved else 'nicht speicherbar'}, nur dieser Zweig wird zurückgesetzt", flush=True)
try:
a.control([" ".join(("feed", "sbf%d", "off")) % hit["slot"]])
except OSError as e:
print(f"send: Zweig konnte nicht zurückgesetzt werden ({type(e).__name__})", flush=True)
return
self.watch.acted_now()
self.hang_n += 1
print(f"send: Hänger erkannt ({hit['text']}); Diagnosepaket {'gesichert (hang-diagnose.txt)' if saved else 'nicht speicherbar'}, belacoder wird neu gestartet", flush=True)
try:
if hit["kind"] != "feed":
p.kill()
self.stop_proc("belacoder")
self.spawn("belacoder", self.args("belacoder"), env=env)
with self.lock:
self.state = "running"
except OSError as e:
print(f"send: Neustart nach Hänger nicht möglich ({type(e).__name__})", flush=True)
def sync_swap_base(self):
global SWAP_BASE
if self.always is not None and SWAP_BASE is not None:
SWAP_BASE = {"cams": list(self.always.binder.slot_key), "group": SWAP_BASE["group"]}
def refresh_inactive(self):
if self.fo is None:
return False
try:
mt = os.stat(f"{STATE}/pipeline.json").st_mtime_ns
except OSError:
return False
if mt == self._inact_mt:
return False
self._inact_mt = mt
new = [k for k in (load_json(f"{STATE}/pipeline.json").get("inactive") or []) if isinstance(k, str) and server.KEY_RE.match(k)][:3]
if sorted(new) == sorted(self.plan["cfg"].get("inactive") or []):
return False
if new:
self.plan["cfg"]["inactive"] = new
else:
self.plan["cfg"].pop("inactive", None)
self.fo.cfg = self.plan["cfg"]
print("send: deaktivierte Kameras geändert: " + (", ".join(new) or "keine"), flush=True)
return True
def sync_cfg(self):
if self.fo is None or SWAP_BASE is None:
return False
try:
mt = os.stat(f"{STATE}/pipeline.json").st_mtime_ns
except OSError:
return False
if mt == self._cfg_mt:
return False
self._cfg_mt = mt
new = {**server.PipelineStore.DEFAULT, **load_json(f"{STATE}/pipeline.json")}
if new.get("type") != "pip" or set(configured_keys(new)) != set(self.fo.keys):
return False
self.plan["cfg"] = new
self.fo.cfg, self.fo.keys = new, configured_keys(new)
def ordered(layout):
return tuple(k for k in self.fo.keys if k in set(layout))[:4]
self.fo.layout = ordered(self.fo.layout)
self.layout = ordered(self.layout)
return True
def wait_links_ready(self, timeout=20):
if not (shutil.which("srtla_send") or "").startswith("/usr/local/"):
time.sleep(1)
return False
end = time.time() + timeout
while time.time() < end and not self.stop_ev.is_set():
if any("srtt=" in l and "srtt=-1ms" not in l for _, l in list(self.links)):
return True
p = self.procs.get("srtla_send")
if p is not None and p.poll() is not None:
return False
self.stop_ev.wait(0.3)
return False
def refresh_uplinks(self):
st = load_json(f"{STATE}/srtla.json").get("settings", {})
ups = [u for u in st.get("uplinks", []) if isinstance(u, str)]
have = {o["iface"]: o["ip"] for o in server.iface_ips()}
ips = [have[u] for u in ups if u in have]
if not ips or ips == self.ips:
return False
self.ips = ips
with open(f"{WORK}/ips", "w") as f:
f.write("\n".join(ips) + "\n")
p = self.procs.get("srtla_send")
if p is not None and p.poll() is None:
p.send_signal(signal.SIGHUP)
print(f"send: Netze zum Senden geändert, {len(ips)} Netz(e) neu gelesen", flush=True)
return True
def shutdown(self):
with self.lock:
self.state = "stopping"
self.write_status()
if self.always is not None:
self.always.stop()
for name in ("belacoder", "srtla_send"):
p = self.procs.get(name)
if p and p.poll() is None:
p.terminate()
deadline = time.time() + 8
for p in self.procs.values():
if p is None:
continue
while p.poll() is None and time.time() < deadline:
time.sleep(0.2)
if p.poll() is None:
p.kill()
if self.live:
self.boost.off()
for f in (STATUS, STATS, BC_STATS, f"{RUN}/srtla-links.txt", server.VIEW_STATE, server.CAM_LIVE):
try:
os.remove(f)
except OSError:
pass
print("send: Sendekette beendet", flush=True)
def main():
try:
if belacoder_running():
raise Refuse("Es läuft schon ein belacoder (z. B. über die BELABOX-Oberfläche). Bitte dort zuerst beenden.")
sv, lat, mn, mx, ips, plan = prepare()
except Refuse as e:
os.makedirs(RUN, exist_ok=True)
with open(STATUS, "w") as f:
json.dump({"state": "refused", "message": str(e), "time": int(time.time())}, f)
os.chmod(STATUS, 0o644)
print(f"send: abgelehnt: {e}", flush=True)
return 1
s = Sender(sv, lat, ips, plan)
signal.signal(signal.SIGTERM, lambda *_: s.stop_ev.set())
signal.signal(signal.SIGINT, lambda *_: s.stop_ev.set())
print(f"send: Server '{sv['name']}', Bitrate {mn}-{mx} kbit/s, Latenz {lat} ms, Wege {len(ips)}", flush=True)
s.run()
return 0
if __name__ == "__main__":
sys.exit(main())
0