129 lines
4.7 KiB
Python
129 lines
4.7 KiB
Python
"""Which of a device's ports someone wants right now.
|
|
|
|
A camera is sent nothing while nobody watches it. A port the device reports as
|
|
``on_demand`` has demand while a viewer socket wants bytes for the message it
|
|
is wired to — a dashboard tile, the Devices tab's preview, a screen through the
|
|
portal — or a node of a running flow reads that message.
|
|
|
|
Every socket says what it wants here, in one registry, and the result is
|
|
published on the bus as a ``device_demand`` event whenever a device's set
|
|
changes. Each device socket picks out its own and tells the device, so nothing
|
|
here has to know which sockets belong to which device.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import threading
|
|
import time
|
|
from collections.abc import Hashable
|
|
from typing import Any
|
|
|
|
from fluksio.flow.events import event_bus
|
|
|
|
|
|
class Demand:
|
|
"""The media names every live socket wants, and what that asks of devices.
|
|
|
|
One per process, as the sockets are. ``bind`` hands it the controller,
|
|
whose device store says what is wired and whose graph says what a flow
|
|
reads.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._lock = threading.Lock()
|
|
self._wanted: dict[Hashable, frozenset[str]] = {}
|
|
#: The last set published per device, so only a change is.
|
|
self._sent: dict[str, tuple[str, ...]] = {}
|
|
self._refresh_lock = threading.Lock()
|
|
self._controller: Any = None
|
|
|
|
def bind(self, controller: Any) -> None:
|
|
self._controller = controller
|
|
self.refresh()
|
|
|
|
def want(self, viewer: Hashable, names: set[str]) -> None:
|
|
"""What one socket wants bytes for now, replacing what it said before."""
|
|
with self._lock:
|
|
if names:
|
|
self._wanted[viewer] = frozenset(names)
|
|
else:
|
|
self._wanted.pop(viewer, None)
|
|
self.refresh()
|
|
|
|
def drop(self, viewer: Hashable) -> None:
|
|
"""A socket closed, and wants nothing any more."""
|
|
self.want(viewer, set())
|
|
|
|
def viewed(self) -> set[str]:
|
|
with self._lock:
|
|
return set().union(*self._wanted.values())
|
|
|
|
def wants(self, names: set[str]) -> bool:
|
|
"""Whether a socket views, or a running flow reads, any of ``names``.
|
|
|
|
What a connector's camera asks before it fetches a frame. True while
|
|
nothing is bound, which is a worker process: the sockets are in the
|
|
engine's, out of its sight.
|
|
"""
|
|
controller = self._controller
|
|
if controller is None:
|
|
# ponytail: a remote worker always sends; forward demand to workers
|
|
# when a camera placed on one costs too much.
|
|
return True
|
|
return bool(names & (self.viewed() | controller.consumed_messages()))
|
|
|
|
def ports(self, device_id: str) -> list[str]:
|
|
"""The device's ports that have demand, sorted."""
|
|
return list(self._compute().get(device_id, ()))
|
|
|
|
def refresh(self) -> None:
|
|
"""Work demand out again and publish what changed.
|
|
|
|
Called whenever its inputs move: a socket's wanted set, and a rebuild,
|
|
a started or stopped flow or a new report, which change what is wired
|
|
and what reads it.
|
|
"""
|
|
with self._refresh_lock:
|
|
found = self._compute()
|
|
for device_id in sorted(set(found) | set(self._sent)):
|
|
ports = found.get(device_id, ())
|
|
if ports == self._sent.get(device_id, ()):
|
|
continue
|
|
if ports:
|
|
self._sent[device_id] = ports
|
|
else:
|
|
self._sent.pop(device_id, None)
|
|
event_bus.publish(
|
|
{
|
|
"type": "device_demand",
|
|
"device": device_id,
|
|
"ports": list(ports),
|
|
"ts": time.time(),
|
|
}
|
|
)
|
|
|
|
def _compute(self) -> dict[str, tuple[str, ...]]:
|
|
controller = self._controller
|
|
devices = getattr(controller, "devices", None)
|
|
if devices is None:
|
|
return {}
|
|
wanted = self.viewed() | controller.consumed_messages()
|
|
found: dict[str, tuple[str, ...]] = {}
|
|
for device_id in devices.wiring():
|
|
reported = devices.reported_ports(device_id) or {}
|
|
ports = tuple(
|
|
sorted(
|
|
binding.port
|
|
for binding in devices.bound(device_id)
|
|
if binding.direction == "in"
|
|
and binding.message in wanted
|
|
and reported.get(binding.port, {}).get("on_demand") is True
|
|
)
|
|
)
|
|
if ports:
|
|
found[device_id] = ports
|
|
return found
|
|
|
|
|
|
demand = Demand()
|