Repository navigation
Expand file tree
/
Copy pathpipbox_send.py
More file actions
936 lines (872 loc) · 38 KB
/
Copy pathpipbox_send.py
File metadata and controls
936 lines (872 loc) · 38 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
#!/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())