Repository navigation
8000
Expand file tree
/
Copy pathpipbox_preview.py
More file actions
735 lines (640 loc) · 31.9 KB
/
Copy pathpipbox_preview.py
File metadata and controls
735 lines (640 loc) · 31.9 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
3
8000
58
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
#!/usr/bin/env python3
"""Vorschau des Sendebilds (Issue #52).
Zeigt das Bild, das wirklich gesendet wird: das fertig gemischte und kodierte Bild, nicht einzelne Kameras. Dazu liest ein kurzlebiger Prozess die
SRT-Datenpakete mit, die belacoder lokal an srtla_send schickt (127.0.0.1:9100, unverschlüsselt), ordnet sie, entpackt den MPEG-TS-Strom, dekodiert ihn
mit dem Hardware-Dekoder, kodiert es klein (640x360) wieder mit dem Hardware-Kodierer als H.264 und schickt es als fragmentiertes MP4 (fMP4) an den Browser, der es per
Media Source Extensions abspielt: flüssig mit 30 Bildern pro Sekunde bei rund 1,5 Mbit/s. Für Browser ohne diese Technik gibt es als Ausweichlösung Motion-JPEG
(multipart/x-mixed-replace).
Die Sendekette wird dabei nicht angefasst (kein Abzweig in der Pipeline, kein Eingriff in belacoder): Ein Seitenzweig vor dem Encoder ließ den
Mischer abbrechen, hier wird nur mitgelesen. Ohne Zuschauer läuft nichts. Eine Vorschau endet, wenn der Browser die Verbindung schließt, nach
MAX_SECONDS oder wenn der Sender stoppt.
Gemessen auf der Box (Orange Pi 5 Plus, 1080p30 HEVC, 640x360, 30 Bilder/s): H.264 rund 1,7 Mbit/s, Motion-JPEG 3 bis 7 Mbit/s; die Sendung bleibt unberührt.
Rechte: Der Webserver läuft als Benutzer pipbox und darf weder rohe Pakete lesen noch den Hardware-Dekoder benutzen (/dev/mpp_service gehört root). Die Arbeit macht darum ein
eigener kleiner Dienst (pipbox-preview.service, Aufruf: pipbox_preview.py --serve) nur mit dem Recht CAP_NET_RAW. Er wird von systemd erst gestartet, wenn jemand die Vorschau
öffnet (pipbox-preview.socket), und beendet sich nach IDLE_EXIT Sekunden ohne Zuschauer. Der Webserver spricht über einen Unix-Socket (Gruppe pipbox, Modus 0660) mit ihm:
eine JSON-Zeile hin, eine JSON-Zeile zurück, danach (nur bei "stream") die Motion-JPEG-Daten, bis eine Seite schließt."""
import ctypes
import json
import os
import select
import shutil
import socket
import struct
import subprocess
import sys
import threading
import time
LISTEN_PORT = 9100 # belacoder -> srtla_send (wie in pipbox_send.py)
SOCKET = "/run/pipbox-preview.sock"
SEND_STATUS = "/run/pipbox-send/status.json"
IDLE_EXIT = 60 # der Dienst beendet sich nach so vielen Sekunden ohne Zuschauer (systemd startet ihn bei der nächsten Anfrage neu)
MAX_REQUEST = 512 # längste Anfragezeile in Byte
MAX_SECONDS = 600 # eine Vorschau endet nach 10 Minuten, die Seite fragt dann "Weiter ansehen?"
LONG_SECONDS = 24 * 3600 # mit dem Häkchen "dauerhaft" in der Oberfläche (je Browser): so lange, danach startet die Seite von selbst neu
MAX_VIEWERS = 2 # gleichzeitige Vorschauen (jede kostet rund 18 % eines Kerns)
CHECK_EVERY = 3.0 # so oft wird geprüft, ob der Sender noch läuft
FIRST_FRAME_WAIT = 12.0 # bis zum ersten Bild: Schlüsselbild abwarten (alle 2 s), dann dekodieren
REORDER_WINDOW = 48 # so viele Pakete warten auf ein verspätetes (SRT wiederholt verlorene Pakete); danach wird die Lücke übersprungen
BOUNDARY = "pbframe"
FORMATS = ("mp4", "mjpeg")
H264_BITRATE = 1_500_000 # Bit/s des Videos zum Browser (640x360, 30 Bilder/s)
FRAGMENT_FRAMES = 6 # so viele Bilder je Häppchen (6 = 0,2 s): kleine Verzögerung, die Seite spielt am aktuellen Ende ab
FPS_RANGE = (1, 30)
WIDTH_RANGE = (160, 1280)
SO_ATTACH_FILTER = 26
SKF_AD_PKTTYPE = 0xfffff004 # BPF: Paketart (auf lo gibt es jedes Paket als ausgehende und als eingehende Kopie)
class Unavailable(Exception):
"""Die Vorschau lässt sich gerade nicht starten (Text für die Oberfläche)."""
class Busy(Unavailable):
pass
def clamp(v, lo, hi, default):
try:
v = int(v)
except (TypeError, ValueError):
return default
return max(lo, min(hi, v))
def bpf_program(port):
"""Klassischer BPF-Filter: nur UDP-Pakete an den Zielport, keine Fragmente, nicht die ausgehende Kopie auf lo.
Der Kernel wirft alles andere weg (RTMP-Ströme der Kameras laufen auch über lo und würden sonst Rechenzeit kosten)."""
ins = [
(0x30, 0, 0, 9), # ldb [9] Protokoll
(0x15, 0, 8, 17), # jeq 17 UDP, sonst verwerfen
(0x28, 0, 0, 6), # ldh [6] Fragment-Feld
(0x45, 6, 0, 0x1fff), # jset 0x1fff Fragment: verwerfen
(0xb1, 0, 0, 0), # ldxb 4*([0]&0xf) Länge des IP-Kopfs
(0x48, 0, 0, 2), # ldh [x+2] Zielport
(0x15, 0, 3, port), # jeq port sonst verwerfen
(0x20, 0, 0, SKF_AD_PKTTYPE), # ld pkt_type
(0x15, 1, 0, 4), # jeq OUTGOING ausgehende Kopie verwerfen
(0x06, 0, 0, 0x40000), # annehmen
(0x06, 0, 0, 0), # verwerfen
]
return b"".join(struct.pack("HBBI", c, jt, jf, k) for c, jt, jf, k in ins), len(ins)
def attach_filter(sock, port):
prog, n = bpf_program(port)
buf = ctypes.create_string_buffer(prog)
sock.setsockopt(socket.SOL_SOCKET, SO_ATTACH_FILTER, struct.pack("HL", n, ctypes.addressof(buf)))
return buf
def srt_data(pkt):
"""(Folgenummer, TS-Nutzdaten) eines SRT-Datenpakets aus einem IPv4/UDP-Paket, sonst None (Steuerpaket, verschlüsselt, zu kurz)."""
if len(pkt) < 36 or pkt[0] >> 4 != 4 or pkt[9] != 17:
return None
pl = pkt[(pkt[0] & 15) * 4 + 8:]
if len(pl) < 16 + 188:
return None
w0, w1 = struct.unpack("!II", pl[:8])
if w0 & 0x80000000: # Steuerpaket (Handshake, ACK, NAK, Keepalive)
return None
if (w1 >> 27) & 3: # verschlüsselt: nicht lesbar
return None
return w0, pl[16:]
class Reorder:
"""Bringt SRT-Datenpakete in die richtige Reihenfolge. SRT sendet verlorene Pakete später noch einmal; solche Wiederholungen füllen Lücken.
Ein Paket, das länger als REORDER_WINDOW Pakete fehlt, wird aufgegeben (der Dekoder fängt sich am nächsten Schlüsselbild)."""
def __init__(self, window=REORDER_WINDOW):
self.window = window
self.buf = {}
self.nxt = None
self.skipped = 0
def push(self, seq, payload):
out = []
if self.nxt is None:
self.nxt = seq
if ((seq - self.nxt) & 0x7fffffff) > 0x3fffffff: # zu alt (schon ausgegeben oder aufgegeben, oder die zweite Kopie)
return out
self.buf[seq] = payload
while self.nxt in self.buf:
out.append(self.buf.pop(self.nxt))
self.nxt = (self.nxt + 1) & 0x7fffffff
if len(self.buf) > self.window:
self.nxt = min(self.buf, key=lambda x: (x - self.nxt) & 0x7fffffff)
self.skipped += 1
while self.nxt in self.buf:
out.append(self.buf.pop(self.nxt))
self.nxt = (self.nxt + 1) & 0x7fffffff
return out
def detect_codec(pipeline_text):
"""h265 oder h264, je nachdem, welchen Encoder die Sendekette benutzt (steht in der Pipeline-Datei)."""
t = (pipeline_text or "").lower()
return "h265" if ("265" in t or "hevc" in t) else "h264"
def big_cpus(cpu_dir="/sys/devices/system/cpu"):
"""Nummern der schnellen Kerne (big.LITTLE über cpu_capacity), [] wenn alle gleich schnell sind (wie pipbox_live.big_cpus, ohne dessen Abhängigkeiten)."""
cap = {}
try:
for d in os.listdir(cpu_dir):
if d.startswith("cpu") and d[3:].isdigit():
try:
with open("%s/%s/cpu_capacity" % (cpu_dir, d)) as f:
cap[int(d[3:])] = int(f.read().strip())
except (OSError, ValueError):
pass
except OSError:
return []
if not cap or max(cap.values()) == min(cap.values()):
return []
top = max(cap.values())
return sorted(c for c, v in cap.items() if v >= top * 0.9)
def sending_now(path=SEND_STATUS, now=time.time):
"""Läuft der Sender? Der Sende-Dienst schreibt seinen Zustand alle 2 s nach /run/pipbox-send/status.json (jünger als 15 s, nicht im Beenden)."""
try:
with open(path) as f:
st = json.load(f)
return now() - float(st.get("time", 0)) < 15 and st.get("state") not in ("stopping", "refused")
except (OSError, ValueError, TypeError, AttributeError):
return False
def little_cpus(big):
"""Alle Kerne außer den schnellen: dort läuft die Vorschau, damit die schnellen für die Sendung frei bleiben."""
try:
n = os.cpu_count() or 0
except Exception:
n = 0
return [c for c in range(n) if c not in set(big)] if big and n else []
def gst_argv(codec, fps, width, cpus=None, fmt="mjpeg"):
"""Kommandozeile des Dekoders: TS von stdin, nach stdout entweder H.264 als Byte-Strom (fmt "mp4", daraus macht Fmp4 das fragmentierte MP4) oder Motion-JPEG
(fmt "mjpeg"). Der Tonzweig läuft ohne Warten, sonst blockiert er tsdemux, solange das Video noch auf das erste Schlüsselbild wartet. Kleiner machen mit
der einfachsten Methode (nearest-neighbour): das ist rund fünfmal billiger als bilinear."""
width = width - width % 2
height = (width * 9 // 16) - ((width * 9 // 16) % 2)
pre = ["taskset", "-c", ",".join(map(str, cpus))] if cpus and shutil.which("taskset") else []
if fmt == "mp4":
out = ["!", "mpph264enc", f"bitrate={H264_BITRATE}", f"gop={fps}", "profile=main", "level=31", "!", "h264parse", "config-interval=-1",
"!", "video/x-h264,stream-format=byte-stream,alignment=au", "!", "fdsink", "fd=1"]
else:
out = ["!", "mppjpegenc", "q-factor=60", "!", "multipartmux", f"boundary={BOUNDARY}", "!", "fdsink", "fd=1"]
return pre + ["gst-launch-1.0", "-q", "fdsrc", "fd=0", "!", "tsdemux", "name=t",
"t.", "!", f"video/x-{codec}", "!", "queue", "!", f"{codec}parse", "!", "mppvideodec",
"!", "videorate", "drop- "!", f"video/x-raw,framerate={fps}/1",
"!", "videoscale", "method=nearest-neighbour", "!", f"video/x-raw,width={width},height={height}"] + out + \
["t.", "!", "audio/x-opus", "!", "queue", "leaky=downstream", "!", "fakesink", "sync=false", "async=false"]
# ---- fragmentiertes MP4 aus einem H.264-Byte-Strom (Annex B), ohne GStreamer-Muxer: mp4mux lieferte im Live-Betrieb keine Häppchen
def _box(typ, *parts):
body = b"".join(parts)
return struct.pack(">I4s", 8 + len(body), typ) + body
def _full(typ, version, flags, *parts):
return _box(typ, struct.pack(">I", (version << 24) | flags), *parts)
_MATRIX = struct.pack(">9I", 0x00010000, 0, 0, 0, 0x00010000, 0, 0, 0, 0x40000000)
def init_segment(sps, pps, width, height, timescale=90000):
"""ftyp + moov für einen H.264-Videostrom (sps und pps ohne Startcode)."""
ftyp = _box(b"ftyp", b"isom", struct.pack(">I", 512), b"isomiso6avc1mp41")
avcc = _box(b"avcC", bytes([1, sps[1], sps[2], sps[3], 0xFF, 0xE1]), struct.pack(">H", len(sps)), sps, bytes([1]), struct.pack(">H", len(pps)), pps)
avc1 = _box(b"avc1", bytes(6), struct.pack(">H", 1), bytes(16), struct.pack(">HH", width, height), struct.pack(">II", 0x00480000, 0x00480000), bytes(4),
struct.pack(">H", 1), bytes(32), struct.pack(">Hh", 0x18, -1), avcc)
stbl = _box(b"stbl", _full(b"stsd", 0, 0, struct.pack(">I", 1), avc1), _full(b"stts", 0, 0, struct.pack(">I", 0)), _full(b"stsc", 0, 0, struct.pack(">I", 0)),
_full(b"stsz", 0, 0, struct.pack(">II", 0, 0)), _full(b"stco", 0, 0, struct.pack(">I", 0)))
dinf = _box(b"dinf", _full(b"dref", 0, 0, struct.pack(">I", 1), _full(b"url ", 0, 1)))
minf = _box(b"minf", _full(b"vmhd", 0, 1, bytes(8)), dinf, stbl)
mdhd = _full(b"mdhd", 0, 0, struct.pack(">IIIIHH", 0, 0, timescale, 0, 0x55C4, 0))
hdlr = _full(b"hdlr", 0, 0, struct.pack(">I", 0), b"vide", bytes(12), b"VideoHandler\0")
tkhd = _full(b"tkhd", 0, 3, struct.pack(">IIIII", 0, 0, 1, 0, 0), bytes(8), struct.pack(">hhhH", 0, 0, 0, 0), _MATRIX, struct.pack(">II", width << 16, height << 16))
trak = _box(b"trak", tkhd, _box(b"mdia", mdhd, hdlr, minf))
mvhd = _full(b"mvhd", 0, 0, struct.pack(">IIII", 0, 0, 1000, 0), struct.pack(">IhH", 0x00010000, 0x0100, 0), bytes(8), _MATRIX, bytes(24), struct.pack(">I", 2))
mvex = _box(b"mvex", _full(b"trex", 0, 0, struct.pack(">IIIII", 1, 1, 0, 0, 0)))
return ftyp + _box(b"moov", mvhd, trak, mvex)
def media_segment(seq, decode_time, samples):
"""moof + mdat; samples: [(Bytes im AVCC-Format, Dauer, Schlüsselbild)]."""
trun_rows = b"".join(struct.pack(">III", dur, len(data), 0x02000000 if key else 0x01010000) for data, dur, key in samples)
mfhd = _full(b"mfhd", 0, 0, struct.pack(">I", seq))
tfhd = _full(b"tfhd", 0, 0x020000, struct.pack(">I", 1))
tfdt = _full(b"tfdt", 1, 0, struct.pack(">Q", decode_time))
def moof(offset):
trun = _full(b"trun", 0, 0x701, struct.pack(">Ii", len(samples), offset), trun_rows)
return _box(b"moof", mfhd, _box(b"traf", tfhd, tfdt, trun))
size = len(moof(0))
return moof(size + 8) + _box(b"mdat", b"".join(d for d, _, _ in samples))
class Fmp4:
"""Macht aus dem H.264-Byte-Strom (Annex B, wie ihn der Kodierer liefert) ein fragmentiertes MP4: zuerst Init-Segment, dann alle FRAGMENT_FRAMES Bilder ein
Häppchen. Beginnt am ersten Schlüsselbild mit SPS und PPS, die Bilder haben feste Dauer (fps). feed(bytes) gibt die fertigen Stücke zurück."""
def __init__(self, fps=30, width=640, height=360, frames=FRAGMENT_FRAMES):
self.fps, self.width, self.height, self.frames = fps, width, height, frames
self.buf = b""
self.au, self.au_slice = [], False
self.sps = self.pps = None
self.started = False
self.samples = []
self.seq = 0
self.t = 0
self.dur = 90000 // fps
def feed(self, data):
"""Neue Bytes des Byte-Stroms; Rückgabe: Liste fertiger Stücke (Init-Segment, Häppchen). Die letzte NAL-Einheit bleibt liegen, bis die nächste beginnt."""
self.buf += data
out = []
starts = self._starts()
if len(starts) >= 2:
for a, b in zip(starts, starts[1:]):
nal = self.buf[a:b - 3].rstrip(b"\x00") # b - 3: vor dem Startcode; ein vierter Nullbyte des Startcodes gehört nicht zur Einheit
out += self._nal(nal)
self.buf = self.buf[starts[-1] - 3:] # die unfertige Einheit behält ihren Startcode
return out
def _starts(self):
"""Stellen direkt hinter den Startcodes (00 00 01)."""
i, res = 0, []
while True:
j = self.buf.find(b"\x00\x00\x01", i)
if j < 0:
return res
res.append(j + 3)
i = j + 3
def _nal(self, nal):
if not nal:
return []
t = nal[0] & 0x1F
out = []
if t in (6, 7, 8, 9) and self.au_slice:
out += self._emit()
elif t in (1, 5) and self.au_slice and len(nal) > 1 and nal[1] & 0x80: # first_mb_in_slice == 0: ein neues Bild beginnt
out += self._emit()
if t == 7:
self.sps = nal
elif t == 8:
self.pps = nal
self.au.append(nal)
if t in (1, 5):
self.au_slice = True
return out
def _emit(self):
nals, self.au, self.au_slice = self.au, [], False
key = any((n[0] & 0x1F) == 5 for n in nals)
out = []
if not self.started:
if not (key and self.sps and self.pps):
return out
self.started = True
out.append(init_segment(self.sps, self.pps, self.width, self.height))
data = b"".join(struct.pack(">I", len(n)) + n for n in nals if (n[0] & 0x1F) in (1, 5))
self.samples.append((data, self.dur, key))
if len(self.samples) >= self.frames:
self.seq += 1
out.append(media_segment(self.seq, self.t, self.samples))
self.t += self.dur * len(self.samples)
self.samples = []
return out
class Preview:
def __init__(self, port=LISTEN_PORT, pipeline_path=None, big_cpus=None):
self.port = port
self.pipeline_path = pipeline_path
self.big_cpus = big_cpus or globals()["big_cpus"]
self.lock = threading.Lock()
self.viewers = 0
def codec(self):
try:
with open(self.pipeline_path) as f:
return detect_codec(f.read())
except (OSError, TypeError):
return "h264"
@staticmethod
def limit(long=False):
"""Höchstdauer einer Vorschau: 10 Minuten, mit "long" (Häkchen in der Oberfläche) 24 Stunden."""
return LONG_SECONDS if long else MAX_SECONDS
def status(self, sending):
"""Für die Oberfläche: kann eine Vorschau starten, und wenn nicht, warum nicht."""
why = None
if not shutil.which("gst-launch-1.0"):
why = "gst"
elif os.geteuid() != 0:
why = "root"
elif not sending:
why = "off"
with self.lock:
busy = self.viewers >= MAX_VIEWERS
if why is None and busy:
why = "busy"
return {"available": why is None, "why": why, "codec": self.codec(), "max_seconds": MAX_SECONDS, "viewers": self.viewers}
def _acquire(self):
with self.lock:
if self.viewers >= MAX_VIEWERS:
raise Busy("busy")
self.viewers += 1
def _release(self):
with self.lock:
self.viewers = max(0, self.viewers - 1)
def open_capture(self):
"""Raw-Socket auf lo, nur UDP an den Zielport (Filter im Kernel)."""
try:
sock = socket.socket(socket.AF_PACKET, socket.SOCK_DGRAM, socket.htons(0x0003))
sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 8 * 1024 * 1024)
attach_filter(sock, self.port)
sock.bind(("lo", 0))
sock.settimeout(0.5)
return sock
except (OSError, AttributeError) as e:
raise Unavailable("capture: %s" % e)
def spawn(self, fps, width, fmt="mjpeg"):
try:
return subprocess.Popen(gst_argv(self.codec(), fps, width, little_cpus(self.big_cpus()), fmt), stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.DEVNULL, env=dict(os.environ, GST_MPP_NO_RGA="1"), bufsize=0)
except OSError as e:
raise Unavailable("gst: %s" % e)
def stream(self, write, fps, width, still_sending, max_seconds=MAX_SECONDS, ready=None, gone=None, fmt="mjpeg"):
"""Läuft, bis der Browser geht (write löst OSError aus), die Zeit um ist oder der Sender stoppt. write(bytes) gibt Daten an den Browser,
still_sending() sagt, ob der Sender noch läuft, ready() wird einmal aufgerufen, sobald alles gestartet ist, gone() sagt, ob die Gegenseite schon weg ist (auch ohne dass gerade Daten fließen). Rückgabe: Grund des Endes ("client", "zeit", "sender", "kein-bild", "fehler")."""
fps = clamp(fps, *FPS_RANGE, 30)
width = clamp(width, *WIDTH_RANGE, 640)
self._acquire()
sock = proc = None
stop = threading.Event()
try:
sock = self.open_capture()
proc = self.spawn(fps, width, fmt)
mux = Fmp4(fps, width - width % 2, (width * 9 // 16) - ((width * 9 // 16) % 2)) if fmt == "mp4" else None
if ready is not None:
ready() # Mitlesen und Dekoder laufen: ab hier kommen Daten (oder "kein-bild")
def feed():
ro = Reorder()
try:
while not stop.is_set():
try:
pkt = sock.recv(65535)
except socket.timeout:
continue
got = srt_data(pkt)
if got is None:
continue
for chunk in ro.push(*got):
proc.stdin.write(chunk)
except (OSError, ValueError):
pass
finally:
stop.set()
try:
proc.stdin.close()
except OSError:
pass
t = threading.Thread(target=feed, name="preview-feed", daemon=True)
t.start()
t0 = time.monotonic()
first = False
last_check = t0
reason = "client"
fd = proc.stdout.fileno()
while True:
now = time.monotonic()
if now - t0 > max_seconds:
reason = "zeit"
break
if gone is not None and gone():
reason = "client"
break
if stop.is_set() and not select.select([fd], [], [], 0)[0]:
reason = "fehler"
break
if not first and now - t0 > FIRST_FRAME_WAIT:
reason = "kein-bild"
break
if now - last_check > CHECK_EVERY:
last_check = now
if not still_sending():
reason = "sender"
break
if not select.select([fd], [], [], 1.0)[0]:
continue
data = os.read(fd, 65536)
if not data:
reason = "fehler" if not first else "client"
break
try:
for chunk in (mux.feed(data) if mux else [data]):
first = True # erst wenn etwas für den Browser fertig ist (beim MP4 nach dem ersten Schlüsselbild)
write(chunk)
except OSError:
reason = "client"
break
return reason
finally:
stop.set()
if proc is not None:
for fn in (proc.terminate,):
try:
fn()
except OSError:
pass
try:
proc.wait(timeout=3)
except subprocess.TimeoutExpired:
try:
proc.kill()
proc.wait(timeout=3)
except (OSError, subprocess.TimeoutExpired):
pass
for f in (proc.stdout, proc.stdin):
try:
f.close()
except (OSError, AttributeError):
pass
if sock is not None:
try:
sock.close()
except OSError:
pass
self._release()
def iface_addrs(names):
"""IPv4-Adresse je Netzwerkschnittstelle ({Name: Adresse}); was keine hat, fehlt."""
out = {}
try:
import fcntl
except ImportError:
return out
s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
try:
for n in names:
try:
out[n] = socket.inet_ntoa(fcntl.ioctl(s.fileno(), 0x8915, struct.pack("256s", n[:15].encode()))[20:24]) # SIOCGIFADDR
except OSError:
pass
finally:
s.close()
return out
def default_route_iface(path="/proc/net/route"):
"""Schnittstelle der Standardroute (kleinste Metrik), sonst None."""
best = None
try:
with open(path) as f:
for line in f.read().splitlines()[1:]:
c = line.split()
if len(c) >= 7 and c[1] == "00000000" and int(c[3], 16) & 2:
m = int(c[6])
if best is None or m < best[0]:
best = (m, c[0])
except (OSError, ValueError):
return None
return best[1] if best else None
class Traffic:
"""Verkehr der Vorschau zum Browser. Er steht sonst im Upload-Zähler der Netzwerkkarte (/proc/net/dev) und verfälscht die Anzeige der Sendung
(3 Mbit/s zusätzlich bei voller Bildrate). Der Webserver meldet, was er je Verbindung (lokale Adresse) an den Browser schickt; adjust() zieht das
samt Aufschlag für Netzwerkköpfe von der Schnittstelle ab, die diese Adresse hat. Kommt der Browser über einen Proxy auf der Box (Tailscale, Adresse
127.0.0.1) oder ist die Adresse unbekannt, gilt die Schnittstelle der Standardroute."""
TX_FACTOR = 1.046 # Ethernet 14 + IP 20 + TCP 32 Byte je 1448 Byte Nutzdaten
RX_FACTOR = 0.01 # Bestätigungen des Browsers (gemessen 0,9 %)
def __init__(self, addrs=None, default=None):
self.lock = threading.Lock()
self.by_ip = {}
self.addrs, self.default = addrs or iface_addrs, default or default_route_iface
def add(self, ip, nbytes):
with self.lock:
self.by_ip[ip or ""] = self.by_ip.get(ip or "", 0) + nbytes
def adjust(self, net):
"""net: {Schnittstelle: (rx, tx)} aus /proc/net/dev; zurück dasselbe ohne den Verkehr der Vorschau."""
with self.lock:
snap = dict(self.by_ip)
if not snap:
return net
addr = self.addrs(list(net))
per = {}
for ip, n in snap.items():
ifc = next((i for i, a in addr.items() if a == ip), None) or self.default()
if ifc in net:
per[ifc] = per.get(ifc, 0) + n
out = dict(net)
for ifc, n in per.items():
rx, tx = net[ifc]
out[ifc] = (max(0, rx - int(n * self.RX_FACTOR)), max(0, tx - int(n * self.TX_FACTOR)))
return out
TRAFFIC = Traffic()
def read_line(sock, limit=MAX_REQUEST):
"""Eine Zeile (bis \\n) byteweise lesen, damit danach die Nutzdaten unberührt im Socket bleiben. Zu lang oder zu früh geschlossen: ValueError."""
out = b""
while True:
c = sock.recv(1)
if not c:
raise ValueError("Verbindung zu früh geschlossen")
if c == b"\n":
return out
out += c
if len(out) > limit:
raise ValueError("Zeile zu lang")
def peer_gone(conn):
"""Hat die Gegenseite die Verbindung geschlossen? Ohne zu warten (select), denn recv() mit MSG_DONTWAIT wartet bei einem Socket mit Zeitgrenze trotzdem bis zu ihr."""
try:
if not select.select([conn], [], [], 0)[0]:
return False
retur
CA5D
n conn.recv(1, socket.MSG_PEEK) == b""
except (ValueError, OSError):
return True
class Service:
"""Der Dienst hinter dem Unix-Socket: nimmt Anfragen vom Webserver an. Anfragen: {"cmd": "status"} und {"cmd": "stream", "fps", "w", "long"}.
Antwort immer zuerst eine JSON-Zeile: Status, {"ok": true} oder {"error": Grund}; bei "stream" folgen danach die Bilddaten."""
def __init__(self, preview, sending=sending_now, idle=IDLE_EXIT):
self.pv, self.sending, self.idle = preview, sending, idle
self.lock = threading.Lock()
self.active = 0
self.last = time.monotonic()
def serve(self, lsock):
"""Annehmen, bis IDLE_EXIT Sekunden lang niemand mehr da ist; dann Rückkehr (der Dienst endet, systemd startet ihn bei der nächsten Anfrage)."""
lsock.settimeout(1.0)
while True:
try:
conn, _ = lsock.accept()
except socket.timeout:
with self.lock:
busy = self.active
if not busy and time.monotonic() - self.last >= self.idle:
return
continue
except OSError:
return
with self.lock:
self.active += 1
self.last = time.monotonic()
threading.Thread(target=self._run, args=(conn,), name="preview-conn", daemon=True).start()
def _run(self, conn):
try:
self.handle(conn)
except (OSError, ValueError):
pass
finally:
try:
conn.close()
except OSError:
pass
with self.lock:
self.active -= 1
self.last = time.monotonic()
def _send(self, conn, obj):
conn.sendall(json.dumps(obj).encode() + b"\n")
def handle(self, conn):
conn.settimeout(5)
try:
req = json.loads(read_line(conn))
if not isinstance(req, dict):
raise ValueError("keine Anfrage")
except ValueError:
return self._send(conn, {"error": "anfrage"})
cmd = req.get("cmd")
if cmd == "status":
return self._send(conn, self.pv.status(self.sending()))
if cmd != "stream":
return self._send(conn, {"error": "anfrage"})
st = self.pv.status(self.sending())
if not st["available"]:
return self._send(conn, {"error": st["why"]})
try:
fmt = req.get("fmt") if req.get("fmt") in FORMATS else "mp4"
self.pv.stream(conn.sendall, req.get("fps"), req.get("w"), self.sending, max_seconds=self.pv.limit(req.get("long") is True), gone=lambda: peer_gone(conn), fmt=fmt,
ready=lambda: (self._send(conn, {"ok": True}), conn.settimeout(15))) # Schreibzeit je Stück: ein Browser, der nicht liest, hält den Platz höchstens 15 s
except Busy:
self._send(conn, {"error": "busy"})
except Unavailable:
self._send(conn, {"error": "capture"})
class Client:
"""Der Webserver: fragt den Dienst über den Unix-Socket. Ist er nicht erreichbar (nicht installiert, Socket fehlt), gilt "nicht verfügbar"."""
def __init__(self, path=SOCKET, timeout=5.0):
self.path, self.timeout = path, timeout
def _call(self, req):
s = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
s.settimeout(self.timeout)
try:
s.connect(self.path)
s.sendall(json.dumps(req).encode() + b"\n")
except OSError:
s.close()
raise
return s
def status(self):
try:
s = self._call({"cmd": "status"})
try:
st = json.loads(read_line(s, 4096))
finally:
s.close()
return st if isinstance(st, dict) and "available" in st else {"available": False, "why": "missing"}
except (OSError, ValueError):
return {"available": False, "why": "missing"}
def open(self, fps, width, long=False, fmt="mp4"):
"""(Socket, Kopfzeile) einer Vorschau; bei einem Fehler des Dienstes (Socket, Kopfzeile mit "error") ist der Socket None. OSError, wenn der Dienst fehlt."""
s = self._call({"cmd": "stream", "fps": clamp(fps, *FPS_RANGE, 30), "w": clamp(width, *WIDTH_RANGE, 640), "long": bool(long), "fmt": fmt if fmt in FORMATS else "mp4"})
try:
head = json.loads(read_line(s, 4096))
except ValueError:
s.close()
return None, {"error": "capture"}
if not isinstance(head, dict) or head.get("error") or not head.get("ok"):
s.close()
return None, {"error": (head or {}).get("error", "capture") if isinstance(head, dict) else "capture"}
return s, head
def listen_socket(path=SOCKET):
"""Der Socket von systemd (Socket-Aktivierung, Dateinummer 3), sonst (Handstart zum Testen) selbst angelegt mit Gruppe pipbox und Modus 0660."""
if os.environ.get("LISTEN_PID") == str(os.getpid()) and os.environ.get("LISTEN_FDS") == "1":
return socket.socket(fileno=3)
try:
os.unlink(path)
except OSError:
pass
ls = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
ls.bind(path)
try:
import grp
os.chown(path, 0, grp.getgrnam("pipbox").gr_gid)
except (ImportError, KeyError, OSError):
pass
os.chmod(path, 0o660)
ls.listen(8)
return ls
def main(argv=None):
argv = sys.argv[1:] if argv is None else argv
if argv[:1] != ["--serve"]:
print("Aufruf: pipbox_preview.py --serve (Dienst der Vorschau, Anfragen über %s)" % SOCKET)
return 2
pv = Preview(pipeline_path="/var/tmp/pipbox/pipeline")
Service(pv).serve(listen_socket())
return 0
if __name__ == "__main__":
sys.exit(main())