DCCONSTRUCTIONS
a3c15e11e9
Discover independent camera instances and prepare their versioned runtime from Node or remote Core. Add isolated SDK workers, camera controls, raw dual-fisheye WebRTC preview, and shared action/region loading states. Recover existing Node bindings over known Tailscale addresses after a Core LAN address change. Preserve identities and trust, pin both peers, migrate endpoints with revision checks, and require real heartbeats for online status. Fix the Python client certificate profile for Go X509 verification. Pin Design Guideline 8c53f73 and retain installer/build/acceptance history. Node 0.8.19 is installed; X4 0.1.3-3 is bundled but hardware activation is pending. Validation: qualified DG/Node builds and Go race tests; 31 fleet tests; Python-to-Go certificate interoperability and live tailnet recovery with five fresh heartbeats; prior 38 X4 tests and bounded remote WebRTC acceptance. Clean-OS, replug/power autonomy, local X4 video and long-run stability remain open.
395 lines
16 KiB
Python
395 lines
16 KiB
Python
"""Private WebRTC operator view outside the vendor SDK network namespace.
|
|
|
|
Opening/closing a peer never sends a camera command. Each source has its own
|
|
worker socket, session, decoder and bounded latest-frame slot. No stitching.
|
|
"""
|
|
|
|
import asyncio
|
|
import ipaddress
|
|
import json
|
|
import logging
|
|
import threading
|
|
import time
|
|
import uuid
|
|
from concurrent.futures import TimeoutError as FutureTimeout
|
|
from fractions import Fraction
|
|
|
|
import aioice.ice
|
|
import av
|
|
from aiortc import (
|
|
RTCConfiguration,
|
|
RTCPeerConnection,
|
|
RTCRtpSender,
|
|
RTCSessionDescription,
|
|
VideoStreamTrack,
|
|
)
|
|
from aiortc.mediastreams import MediaStreamError
|
|
|
|
from .errors import failure
|
|
from .frames import decode
|
|
from .http import request
|
|
|
|
NETWORKS = tuple(
|
|
ipaddress.ip_network(value)
|
|
for value in (
|
|
"127.0.0.0/8",
|
|
"10.0.0.0/8",
|
|
"172.16.0.0/12",
|
|
"192.168.0.0/16",
|
|
"100.64.0.0/10",
|
|
)
|
|
)
|
|
|
|
|
|
def private(value):
|
|
try:
|
|
address = ipaddress.ip_address(value)
|
|
return address.version == 4 and any(address in network for network in NETWORKS)
|
|
except ValueError:
|
|
return False
|
|
|
|
|
|
_host_addresses = aioice.ice.get_host_addresses
|
|
aioice.ice.get_host_addresses = lambda use_ipv4, use_ipv6: [
|
|
value for value in _host_addresses(use_ipv4=True, use_ipv6=False) if private(value)
|
|
]
|
|
|
|
|
|
def admit_sdp(sdp):
|
|
if not isinstance(sdp, str) or not 0 < len(sdp) <= 32768:
|
|
raise ValueError("Invalid camera offer")
|
|
for line in sdp.splitlines():
|
|
if line.startswith("a=candidate:"):
|
|
fields = line.split()
|
|
if len(fields) < 8 or not (private(fields[4]) or fields[4].endswith(".local")):
|
|
raise ValueError("Camera preview requires a private network")
|
|
|
|
|
|
class Source:
|
|
def __init__(self, path, session):
|
|
self.path, self.session = path, session
|
|
self.lock = threading.Lock()
|
|
self.frame = None
|
|
self.sequence = 0
|
|
self.observed = 0.0
|
|
self.started = time.monotonic()
|
|
self.closed = threading.Event()
|
|
self.thread = threading.Thread(target=self.pump, daemon=True)
|
|
|
|
def current(self):
|
|
with self.lock:
|
|
return self.sequence, self.frame, self.observed
|
|
|
|
def pump(self):
|
|
cursor, key, decoder, resized_at = 0, None, None, 0.0
|
|
try:
|
|
while not self.closed.is_set():
|
|
raw = request(
|
|
self.path,
|
|
"/video",
|
|
{"session_id": self.session, "cursor": cursor},
|
|
timeout=2,
|
|
binary=True,
|
|
)
|
|
value = decode(raw)
|
|
if value is None:
|
|
continue
|
|
header, data = value
|
|
cursor = header["cursor"]
|
|
if header["stream_index"] != 0:
|
|
continue
|
|
current = (header["generation"], header["codec"])
|
|
if decoder is None or key != current or header["gap"]:
|
|
decoder = av.CodecContext.create(
|
|
"h264" if header["codec"] == 0 else "hevc", "r"
|
|
)
|
|
decoder.thread_count = 1
|
|
key = current
|
|
try:
|
|
for packet in decoder.parse(data):
|
|
for frame in decoder.decode(packet):
|
|
if not (0 < frame.width <= 4096 and 0 < frame.height <= 2160):
|
|
raise RuntimeError("Camera frame exceeds the admitted view profile")
|
|
now = time.monotonic()
|
|
if now - resized_at < 1 / 15:
|
|
continue
|
|
width = min(1280, frame.width) // 2 * 2
|
|
height = max(2, round(frame.height * width / frame.width) // 2 * 2)
|
|
output = frame.reformat(width=width, height=height, format="yuv420p")
|
|
output.pts = round((now - self.started) * 90000)
|
|
output.time_base = Fraction(1, 90000)
|
|
with self.lock:
|
|
self.frame, self.observed = output, now
|
|
self.sequence += 1
|
|
resized_at = now
|
|
except av.error.FFmpegError:
|
|
decoder = None
|
|
except (OSError, ValueError, RuntimeError) as error:
|
|
logging.getLogger(__name__).warning(
|
|
"X4 preview source stopped (%s)", type(error).__name__
|
|
)
|
|
finally:
|
|
self.closed.set()
|
|
with self.lock:
|
|
self.frame = None
|
|
|
|
def close(self):
|
|
self.closed.set()
|
|
|
|
|
|
class Track(VideoStreamTrack):
|
|
def __init__(self, source):
|
|
super().__init__()
|
|
self.source, self.sequence, self.next_at = source, 0, 0.0
|
|
|
|
async def recv(self):
|
|
while self.readyState == "live" and not self.source.closed.is_set():
|
|
await asyncio.sleep(max(0, self.next_at - time.monotonic()))
|
|
sequence, frame, observed = self.source.current()
|
|
if frame is not None and sequence > self.sequence and time.monotonic() - observed < 2:
|
|
self.sequence = sequence
|
|
self.next_at = time.monotonic() + 1 / 15
|
|
# Distinct AV frames per encoder: force-keyframe decisions in
|
|
# one browser must not mutate another browser's frame object.
|
|
copy = av.VideoFrame(frame.width, frame.height, "yuv420p")
|
|
for target, source in zip(copy.planes, frame.planes, strict=True):
|
|
if target.line_size == source.line_size:
|
|
target.update(source)
|
|
else:
|
|
raw = bytes(source)
|
|
padded = bytearray(target.buffer_size)
|
|
for row in range(target.height):
|
|
padded[
|
|
row * target.line_size : row * target.line_size + target.width
|
|
] = raw[row * source.line_size : row * source.line_size + target.width]
|
|
target.update(padded)
|
|
copy.pts, copy.time_base = frame.pts, frame.time_base
|
|
return copy
|
|
await asyncio.sleep(0.01)
|
|
raise MediaStreamError
|
|
|
|
|
|
class Peers:
|
|
def __init__(self, source_factory=Source):
|
|
self.items, self.sources = {}, {}
|
|
self.primed = {}
|
|
self.source_factory = source_factory
|
|
|
|
def source(self, key, path):
|
|
source = self.sources.get(key)
|
|
if source is not None and source.closed.is_set():
|
|
raise RuntimeError("Camera video session ended")
|
|
if source is None:
|
|
if len(self.sources) >= 4:
|
|
raise ValueError("Остановите лишний просмотр камеры.")
|
|
source = self.source_factory(path, key[1])
|
|
self.sources[key] = source
|
|
source.thread.start()
|
|
return source
|
|
|
|
async def prime(self, key, path):
|
|
# Start reading before the explicit camera START. Some previews expose
|
|
# an independently decodable beginning only once per capture session.
|
|
source = self.source(key, path)
|
|
if key in self.primed:
|
|
return {"ok": True}
|
|
|
|
async def watch():
|
|
started, active, fresh = time.monotonic(), False, time.monotonic()
|
|
while key in self.primed:
|
|
try:
|
|
status = await asyncio.to_thread(request, path, "/snapshot", timeout=2)
|
|
preview = status.get("status", {}).get("preview")
|
|
ended = status.get("session_id") != key[1] or source.closed.is_set()
|
|
if status.get("online"):
|
|
fresh = time.monotonic()
|
|
ended |= active and preview == 0
|
|
active |= preview == 1
|
|
ended |= active and time.monotonic() - fresh > 15
|
|
except (OSError, RuntimeError, ValueError):
|
|
ended = True
|
|
if ended or (not active and time.monotonic() - started > 45):
|
|
await self.release(key)
|
|
return
|
|
await asyncio.sleep(1)
|
|
|
|
self.primed[key] = asyncio.create_task(watch())
|
|
return {"ok": True}
|
|
|
|
async def release(self, key):
|
|
task = self.primed.pop(key, None)
|
|
if task is not None and task is not asyncio.current_task():
|
|
task.cancel()
|
|
for identifier, entry in tuple(self.items.items()):
|
|
if entry["key"] == key:
|
|
await self.close(key, identifier)
|
|
source = self.sources.pop(key, None)
|
|
if source is not None:
|
|
source.close()
|
|
await asyncio.to_thread(source.thread.join, 3)
|
|
return {"ok": True}
|
|
|
|
async def verify(self, key, timeout=8):
|
|
# An active preview may have emitted its only keyframe minutes ago.
|
|
# Verify two NEW decoded frames from its existing source, never open a
|
|
# competing decoder or accept a retained frame as fresh evidence.
|
|
source = self.sources.get(key)
|
|
if source is None or source.closed.is_set():
|
|
return failure("preview_restart_required")
|
|
initial = source.current()[0]
|
|
started = time.monotonic()
|
|
while time.monotonic() - started < timeout:
|
|
if self.sources.get(key) is not source or source.closed.is_set():
|
|
break
|
|
sequence, frame, observed = source.current()
|
|
if frame is not None and sequence >= initial + 2 and time.monotonic() - observed < 2:
|
|
return {
|
|
"state": "complete",
|
|
"result": {
|
|
"verified": True,
|
|
"streams": [
|
|
{
|
|
"stream_index": 0,
|
|
"width": frame.width,
|
|
"height": frame.height,
|
|
"frames": sequence - initial,
|
|
}
|
|
],
|
|
"duration_ms": round((time.monotonic() - started) * 1000),
|
|
},
|
|
}
|
|
await asyncio.sleep(0.02)
|
|
return failure("preview_restart_required")
|
|
|
|
async def offer(self, key, path, params):
|
|
admit_sdp(params["sdp"])
|
|
# Inventory remains independent; these limits bound active media only.
|
|
if len(self.items) >= 4 or sum(item["key"] == key for item in self.items.values()) >= 2:
|
|
raise ValueError("Закройте лишний просмотр камеры.")
|
|
source = self.source(key, path)
|
|
pc = RTCPeerConnection(RTCConfiguration(iceServers=[]))
|
|
identifier = "peer_" + uuid.uuid4().hex
|
|
entry = {"key": key, "pc": pc, "source": source, "seen": time.monotonic(), "tasks": set()}
|
|
self.items[identifier] = entry
|
|
|
|
async def telemetry(channel):
|
|
while identifier in self.items:
|
|
sequence, frame, observed = source.current()
|
|
if channel.readyState == "open" and channel.bufferedAmount < 65536:
|
|
channel.send(
|
|
json.dumps(
|
|
{
|
|
"frame_sequence": sequence,
|
|
"frame_age_ms": round((time.monotonic() - observed) * 1000)
|
|
if frame
|
|
else None,
|
|
"fresh": frame is not None and time.monotonic() - observed < 2,
|
|
}
|
|
)
|
|
)
|
|
await asyncio.sleep(0.5)
|
|
|
|
def task(coroutine):
|
|
value = asyncio.create_task(coroutine)
|
|
entry["tasks"].add(value)
|
|
value.add_done_callback(entry["tasks"].discard)
|
|
|
|
@pc.on("datachannel")
|
|
def datachannel(channel):
|
|
if channel.label != "sensor":
|
|
channel.close()
|
|
return
|
|
|
|
@channel.on("message")
|
|
def message(value):
|
|
if value == "keepalive":
|
|
entry["seen"] = time.monotonic()
|
|
|
|
task(telemetry(channel))
|
|
|
|
@pc.on("connectionstatechange")
|
|
async def changed():
|
|
if pc.connectionState in ("failed", "closed"):
|
|
await self.close(key, identifier)
|
|
|
|
async def watchdog():
|
|
while identifier in self.items:
|
|
if source.closed.is_set() or time.monotonic() - entry["seen"] > 30:
|
|
await self.close(key, identifier)
|
|
return
|
|
await asyncio.sleep(1)
|
|
|
|
try:
|
|
sender = pc.addTrack(Track(source))
|
|
transceiver = next(item for item in pc.getTransceivers() if item.sender == sender)
|
|
transceiver.setCodecPreferences(
|
|
[
|
|
codec
|
|
for codec in RTCRtpSender.getCapabilities("video").codecs
|
|
if codec.mimeType.lower() == "video/h264"
|
|
]
|
|
)
|
|
await pc.setRemoteDescription(RTCSessionDescription(sdp=params["sdp"], type="offer"))
|
|
await pc.setLocalDescription(await pc.createAnswer())
|
|
task(watchdog())
|
|
return {"peer_id": identifier, "sdp": pc.localDescription.sdp, "type": "answer"}
|
|
except BaseException:
|
|
await self.close(key, identifier)
|
|
raise
|
|
|
|
async def close(self, key, identifier):
|
|
entry = self.items.get(identifier)
|
|
if entry is None:
|
|
return {"ok": True}
|
|
if entry["key"] != key:
|
|
raise ValueError("Preview belongs to a different camera session")
|
|
self.items.pop(identifier)
|
|
current = asyncio.current_task()
|
|
for task in tuple(entry["tasks"]):
|
|
if task is not current:
|
|
task.cancel()
|
|
await entry["pc"].close()
|
|
if key not in self.primed and not any(item["key"] == key for item in self.items.values()):
|
|
source = self.sources.pop(key)
|
|
source.close()
|
|
await asyncio.to_thread(source.thread.join, 3)
|
|
return {"ok": True}
|
|
|
|
|
|
class Engine:
|
|
def __init__(self):
|
|
self.loop = asyncio.new_event_loop()
|
|
self.peers = Peers()
|
|
self.thread = threading.Thread(target=self.loop.run_forever, daemon=True)
|
|
self.thread.start()
|
|
|
|
def capture(self, key, path, start):
|
|
work = self.peers.prime(key, path) if start else self.peers.release(key)
|
|
future = asyncio.run_coroutine_threadsafe(work, self.loop)
|
|
try:
|
|
return future.result(timeout=10)
|
|
except FutureTimeout as error:
|
|
future.cancel()
|
|
raise RuntimeError("Camera decoder setup timed out") from error
|
|
|
|
def call(self, key, path, action, params):
|
|
work = (
|
|
self.peers.offer(key, path, params)
|
|
if action == "offer"
|
|
else self.peers.close(key, params["peer_id"])
|
|
)
|
|
future = asyncio.run_coroutine_threadsafe(work, self.loop)
|
|
try:
|
|
return {"state": "complete", "result": future.result(timeout=20)}
|
|
except FutureTimeout as error:
|
|
future.cancel()
|
|
raise RuntimeError("Camera preview timed out") from error
|
|
|
|
def verify(self, key):
|
|
future = asyncio.run_coroutine_threadsafe(self.peers.verify(key), self.loop)
|
|
try:
|
|
return future.result(timeout=10)
|
|
except FutureTimeout as error:
|
|
future.cancel()
|
|
raise RuntimeError("Camera image verification timed out") from error
|