######################################################
## MX Remote Python Interface ##
## ##
## author: Lars Op den Kamp (lars@opdenkamp-it.nl) ##
## copyright (c) 2021-2026 Op den Kamp IT Solutions ##
######################################################
'''Remote device discovery and management over UDP multicast/broadcast.'''
from functools import cached_property
import logging
import asyncio
from typing import Any
from ..compat import override
import aiofiles
import aiohttp
import os
from pathlib import Path
import random
import time
from datetime import datetime, timezone
from .ConnectionAsync import ConnectionAsync
from ..const import MXR_HELLO_INTERVAL_MIN, MXR_HELLO_INTERVAL_RAND, __version__
from .Device import Device
from ..Interface import ConnectionCallbacks, DeviceRegistry, MxrDeviceUid, BayLinks, BayBase, DeviceBase, MxrCallbacks, AudioEndpoint, mxr_broadcast_address, OneIPDeviceSettings, OneIPPowerSaveSchedule, OneIPPtpMessages, OsdSyncContent, OsdSyncRequest
from ..proto.Constants import MXR_PROTOCOL_VERSION, DeviceFeature, OsdSyncWhat
from ..proto.FrameDiscover import FrameDiscover, constructFrameDiscover
from ..proto.Factory import process_mxr_frame
from ..proto.FrameHello import FrameHello, constructFrameHello
from ..proto.FramePing import FramePing
from ..proto.FrameBase import FrameBase
from ..proto.FrameTime import FrameTime
from ..proto.FrameTimeZone import FrameTimeZone
from ..proto.FrameV2IPPrivacy import FrameV2IPPrivacy
from ..proto.FrameV2IPSettingsAll import FrameV2IPSettingsAll
from ..proto.FrameV2IPOsdSync import FrameV2IPOsdSync
from .OsdSync import LogoServer, OsdSender, PASS_GAP, osd_content_refusal
from ..proto.Constants import (ONEIP_DEVICE_SETTINGS_REPORTED_ONLY, ONEIP_DSCP_MAX, ONEIP_IR_PROFILE_MAX, ONEIP_PTP_DOMAIN_MAX, ONEIP_SINK_ROTATION_MAX,
OneIPPowerSaveLevel, OneIPPtpMode,
ONEIP_IR_PROFILE_NOT_SET, OneIPDeviceSetting)
from ..proto.Svd import SvdMap
from ..Uid import MxrDeviceUid
from .State import State
import traceback
from ..const import MX_BCAST_UDP_PORT, MX_MCAST_UDP_IP, MX_MCAST_UDP_PORT
_LOGGER = logging.getLogger(__name__)
# The opcodes a device accepts from a sender it has no record of: the
# announcement itself, and the requests for one - discover asks everyone, ping
# asks one device.
_ACCEPT_UNANNOUNCED = (0x00, 0x01, 0x4A)
_BROKEN_PROTOCOL = 2
'''The header protocol a broken revision of one layout was stamped with, which
a unit drops.'''
# How long a discover to every device stands in for asking any one of them, in
# seconds. Each device answers it with everything it holds, after a jitter of up
# to a few seconds, so asking one again inside this window repeats an answer
# already on its way.
_DISCOVER_COVERS = 60
# How often a device that still owes part of its description is asked again,
# and how many times. Every unit resends everything every five minutes, so
# giving up costs at most that wait, where asking on would cost every unit a
# full resend each time.
_REASK_INTERVAL = 60
_REASK_TRIES = 3
# What this client had heard of a device before a frame from it: whether it had
# gone offline, and its reboot toggle as last announced.
_Heard = tuple[bool, bool]
def _heard(device:DeviceBase) -> _Heard:
features = device.features
return (not device.online,
(features is not None) and (DeviceFeature.BOOT_BIT in features))
def _missed_state(before:_Heard|None, device:DeviceBase) -> bool:
'''Whether a device just heard may hold state this client never received.
A device repeats a frame only for a short while after it changes, so
whatever it sent before this client first heard it, or while it was out of
earshot, may not come again. before is None for a device this frame
registered.
Not asked: a device whose reboot toggle flipped, or that is about to reboot,
because a booting device sends everything it holds; and a management client,
which holds nothing to send.
'''
features = device.features
if (features is None) or (DeviceFeature.MANAGER in features) \
or (DeviceFeature.STATUS_REBOOTING in features):
return False
if (before is None):
return True
away, boot_bit = before
return away and (boot_bit == (DeviceFeature.BOOT_BIT in features))
[docs]
class Remote(DeviceRegistry, ConnectionCallbacks):
''' Main component that handles the network connections and registration of remote devices '''
[docs]
def __init__(self, target_ip:str|None=None, port:int|None=None, http_session:aiohttp.ClientSession|None=None, open_connection:bool=True, callbacks:MxrCallbacks|None=None, name:str="MXR Python", local_ip:str|None=None, broadcast:bool|None=None, addr_filter:str|None=None, uid_path:str|None=None, logo_port:int|None=None) -> None:
'''Initialise the remote controller.
:param target_ip: multicast/broadcast IP to use, or None for the default
:param port: UDP port to use, or None for the default
:param http_session: shared aiohttp session for API calls, or None to create one
:param open_connection: open the UDP connection immediately when True
:param callbacks: event callbacks for device state changes
:param name: human-readable name advertised on the network
:param local_ip: local interface IP to bind to, or None for any
:param broadcast: use broadcast instead of multicast when True
:param addr_filter: only log frames from this IP address when set
:param uid_path: file holding this client's uid, or None for ~/.mxr-uid.
Give concurrent clients on one host separate paths - see _load_uid
:param logo_port: TCP port to serve overlay sync logo images on, or None
for a free one
'''
DeviceRegistry.__init__(self)
ConnectionCallbacks.__init__(self)
self._name = name
self._closing = False
self._own_session = (http_session is None)
self._callbacks = State(callbacks, http_session)
self.remotes:dict[MxrDeviceUid,Device] = {}
self._links = BayLinks(self)
self._hello_due = 0.0
self._probe_failures = 0
self._tasks:set[asyncio.Task[None]] = set()
self._uid:bytes|None = None
self._uid_path = uid_path
self._local_ip = local_ip
self._broadcast = broadcast
self._target_ip = target_ip
self._port = port
self._discover_timeout = 0
# Per device still owing part of its description: how often it was
# asked again, and when last.
self._reasked:dict[MxrDeviceUid, tuple[int, float]] = {}
self._state_request:MxrDeviceUid|None = None
self._addr_filter = addr_filter
self._svd = SvdMap()
self._osd = OsdSender()
self._osd_queued = asyncio.Event()
self._logos = LogoServer()
self._logo_port = logo_port
if open_connection:
self.conn = ConnectionAsync(callbacks=self, target_ip=self.target_ip, port=self.port, local_ip=self._local_ip)
else:
self.conn = None
@property
def library_version(self) -> str:
''' version of the mx_remote library '''
return __version__
@property
def protocol_version(self) -> int:
''' protocol version used by this library '''
return MXR_PROTOCOL_VERSION
@property
def net_protocol_version_max(self) -> int:
''' highest protocol version used by devices on the network '''
proto = 0
for _, device in self.remotes.items():
if (device.protocol > proto):
proto = device.protocol
return proto
@property
def net_protocol_version_min(self) -> int:
''' lowest protocol version used by devices on the network '''
proto = MXR_PROTOCOL_VERSION
for _, device in self.remotes.items():
if (device.protocol < proto):
proto = device.protocol
return proto
@cached_property
def target_ip(self) -> str:
if self._target_ip is None:
if self._broadcast is None or not self._broadcast:
return MX_MCAST_UDP_IP
bcast = mxr_broadcast_address(self._local_ip)
if bcast is not None:
return bcast
_LOGGER.warning("failed to determine broadcast address, using multicast")
return MX_MCAST_UDP_IP
return self._target_ip
@property
def local_ip(self) -> str|None:
return self._local_ip
@property
def broadcast(self) -> bool:
return ((self._broadcast is not None) and self._broadcast)
@property
def port(self) -> int:
if self._port is None:
return MX_MCAST_UDP_PORT if (self._broadcast is None or not self._broadcast) else MX_BCAST_UDP_PORT
return self._port
async def _load_uid(self) -> None:
'''Load this client's uid, generating and persisting one if absent.
The uid is this client's identity on the mesh: addressing is by uid in
the payload, and process_frame() drops any frame whose remote_id is our
own. Two clients sharing a uid are therefore one peer to every device,
and each silently discards everything the other sends. Pass separate
uid_path values to run concurrent clients from one account.
'''
if self._uid is not None:
return
uid_path = Path(self._uid_path) if (self._uid_path is not None) else Path.home().joinpath(".mxr-uid")
try:
async with aiofiles.open(uid_path, "rb") as f:
self._uid = await f.read()
except Exception:
_LOGGER.info(f"failed to read {uid_path}. creating new file")
self._uid = os.urandom(16)
async with aiofiles.open(uid_path, "wb") as f:
await f.write(self._uid)
# clear cached uid so it gets recomputed with the loaded value
vars(self).pop('uid', None)
@property
def uid_raw(self) -> bytes|None:
return self._uid
@cached_property
def uid(self) -> MxrDeviceUid:
return MxrDeviceUid(self.uid_raw)
@property
def name(self) -> str:
return self._name
@property
def callbacks(self) -> MxrCallbacks:
return self._callbacks.callbacks
@property
def http_session(self) -> aiohttp.ClientSession:
''' Active HTTP client session for API commands '''
return self._callbacks.http_session
@property
def svd_map(self) -> SvdMap:
return self._svd
[docs]
async def update_config(self, callbacks:MxrCallbacks|None=None, name:str|None=None, target_ip:str|None=None, port:int|None=None, local_ip:str|None=None, broadcast:bool|None=None) -> None:
'''Update runtime configuration, reconnecting if network parameters changed.'''
if (callbacks is not None):
self._callbacks = State(callbacks, self.http_session)
if (name is not None):
self._name = name
if (target_ip is not None) or (port is not None) or (local_ip is not None) or (broadcast is not None):
changed = False
if (target_ip is not None) and (self._target_ip != target_ip):
_LOGGER.debug(f"updating target ip to {target_ip}")
self._target_ip = target_ip
del self.target_ip
changed = True
if (port is not None) and (self._port != port):
_LOGGER.debug(f"updating target port to {port}")
self._port = port
changed = True
if (local_ip is not None) and (self._local_ip != local_ip):
_LOGGER.debug(f"updating target ip to {local_ip}")
self._local_ip = local_ip
changed = True
if (broadcast is not None) and (self._broadcast != broadcast):
_LOGGER.debug(f"updating target ip to {broadcast}")
self._broadcast = broadcast
changed = True
if changed:
if (self.conn is not None):
_LOGGER.debug(f"closing connection")
self.conn.close()
_LOGGER.debug(f"opening new mx_remote listener on target={self.target_ip} listener={self._local_ip}:{self.port}")
self.conn = ConnectionAsync(callbacks=self, target_ip=self.target_ip, port=self.port, local_ip=self._local_ip)
await self.conn.start_srv()
[docs]
def has_completed_devices(self) -> bool:
'''Return True if at least one registered device has completed configuration.'''
for _, device in self.remotes.items():
if device.configuration_complete:
return True
return False
def _on_task_done(self, task:'asyncio.Task[None]') -> None:
'''Drop a finished background task, and report it if something killed it.
asyncio only surfaces an unretrieved task exception when the task object is
garbage collected, which is a different moment and may never come, so a task
that dies otherwise does so in silence. Retrieving it here is what makes any
future background task fail loudly, including one whose own guard misses.
'''
self._tasks.discard(task)
if task.cancelled():
return
exc = task.exception()
if (exc is not None):
_LOGGER.error(f"background task {task.get_name()} died: {exc!r}")
async def _background_probe(self) -> None:
while not self._closing:
await asyncio.sleep(1)
try:
self._probe_once()
self._probe_failures = 0
except Exception as e:
# A send raises on a closed or unreachable socket. Letting that
# escape ends the task, and this loop is the only thing that
# announces this client and drives discovery, so one transient
# failure would silence it permanently.
#
# A disconnected interface fails every tick, so only the first
# carries a traceback: a full one per second buries everything
# else in the log, and the repeats say nothing the first did not.
self._probe_failures += 1
if (self._probe_failures == 1):
_LOGGER.warning(f"probe failed: {traceback.format_exc()}")
else:
_LOGGER.debug(f"probe still failing ({self._probe_failures}): {e!r}")
def _probe_once(self) -> None:
'''One pass of the probe loop: announce if due, ask a device again for
what it still owes, and discover every device while none has answered.
A discover to every device costs every unit a full resend, so it goes
out at most once a minute, and only until a device answers: every unit
heard that discover, and one still owing something is asked by name.'''
if (time.time() >= self._hello_due):
self.tx_hello()
# Every registered device, whatever any of them has finished. A device
# stops pinging whether or not it ever described itself, and this is the
# only thing that notices: nothing arrives to mark a device gone, so a
# pass skipped is a device that stays online in the registry for as long
# as the client runs.
for uid, device in self.remotes.items():
device.check_online()
if device.check_configuration_complete_timeout():
if device.configuration_complete:
self._reasked.pop(uid, None)
elif device.online:
self._reask(uid)
if (len(self.remotes) == 0) and ((time.time() - self._discover_timeout) >= _DISCOVER_COVERS):
self.tx_discover()
def _reask(self, device:MxrDeviceUid) -> None:
'''Ask a device again for its state, at most once a minute and
_REASK_TRIES times until it has described itself.'''
tries, last = self._reasked.get(device, (0, 0.0))
if (tries >= _REASK_TRIES) or ((time.time() - last) < _REASK_INTERVAL):
return
if self._request_state(device):
self._reasked[device] = (tries + 1, time.time())
[docs]
async def start_async(self) -> None:
'''Start the server that listens for mx_remote frames from other devices.'''
if (self.conn is None):
raise Exception("connection not open")
await self._load_uid()
await self.conn.start_srv()
# Announce before anything else can be sent. A device drops every frame
# from a uid it has no record of, except hello and discover, so a
# command issued between here and the probe loop's first pass would be
# dropped by every peer that has not seen us - silently, since nothing
# answers a frame it discarded.
#
# The discover behind it makes each peer hello back rather than waiting
# out its own announce interval. It goes second so a peer has our
# announce before it answers.
self.tx_hello()
self.tx_discover()
for loop in (self._background_probe(), self._osd_loop()):
task = asyncio.create_task(loop)
self._tasks.add(task)
task.add_done_callback(self._on_task_done)
[docs]
async def stop_async(self) -> None:
'''Stop the server and close all connections.'''
await self.close()
[docs]
async def close(self) -> None:
'''Close all open connections and cancel background tasks.'''
_LOGGER.debug("closing mx_remote listener")
self._closing = True
for task in self._tasks:
task.cancel()
if self._tasks:
await asyncio.gather(*self._tasks, return_exceptions=True)
self._tasks.clear()
self._osd.drop()
await self._logos.stop()
if self.conn is not None:
self.conn.close()
if self._own_session:
await self.http_session.close()
[docs]
def get_by_serial(self, serial:str) -> DeviceBase|None:
'''Get the cached device matching the given serial number, or None.'''
for _, remote in self.remotes.items():
if serial == remote.serial:
return remote
return None
[docs]
def get_by_uid(self, remote_id:str|MxrDeviceUid|None) -> DeviceBase|None:
'''Get the cached device matching the given unique id, or None.'''
if (remote_id is None):
return None
uid = remote_id if isinstance(remote_id, MxrDeviceUid) else MxrDeviceUid(remote_id)
if ((dev := self.remotes.get(uid)) is not None):
return dev
if isinstance(remote_id, str):
return self.get_by_serial(serial=remote_id)
return None
[docs]
def uid_to_user_string(self, remote_id:str|MxrDeviceUid|bytes|None) -> str:
if (remote_id is None):
return '<none>'
if not isinstance(remote_id, MxrDeviceUid):
remote_id = MxrDeviceUid(remote_id)
if remote_id in self.remotes.keys():
return self.remotes[remote_id].serial
return str(remote_id)
[docs]
def get_by_stream_ip(self, ip:str, audio:bool=False) -> BayBase|None:
for _, dev in self.remotes.items():
if not dev.is_oneip or dev.oneip_sources is None or dev.first_input is None:
continue
if not audio and (dev.first_input.oneip_source is not None) and (dev.first_input.oneip_source.video.ip == ip):
return dev.first_input
if audio and (dev.first_input.oneip_source is not None) and (dev.first_input.oneip_source.audio.ip == ip):
return dev.first_input
return None
[docs]
def get_bay_by_portnum(self, remote_id:str|MxrDeviceUid, portnum:int) -> BayBase|None:
'''Get the cached bay for a device, given the device id and port number.'''
device = self.get_by_uid(remote_id=remote_id) if isinstance(remote_id, MxrDeviceUid) else self.get_by_serial(serial=remote_id)
if device is None:
return None
return device.get_by_portnum(portnum)
[docs]
def get_bay_by_portname(self, remote_id:str|MxrDeviceUid, portname:str) -> BayBase|None:
device = self.get_by_uid(remote_id=remote_id) if isinstance(remote_id, MxrDeviceUid) else self.get_by_serial(serial=remote_id)
if device is None:
return None
return device.get_by_portname(portname=portname)
def get_audio_endpoint(self, device:MxrDeviceUid, id:int) -> AudioEndpoint|None:
dev = self.get_by_uid(remote_id=device)
if (dev is not None):
return dev.audio_endpoint_by_id(id=id)
return None
@property
def links(self) -> BayLinks:
return self._links
[docs]
def transmit(self, data: bytes) -> int:
'''Transmit raw data over the UDP connection. Returns bytes sent.'''
if (self.conn is None):
raise Exception("connection closed")
return self.conn.transmit(data=data)
[docs]
def tx_discover(self) -> int:
'''Transmit a discover frame. All remotes will respond with a hello frame.'''
pkt = constructFrameDiscover(self)
if (pkt is None):
return 0
self._discover_timeout = time.time()
_LOGGER.debug("discovering devices")
return self.transmit(pkt.frame)
def _request_state(self, device:MxrDeviceUid) -> bool:
'''Ask one device for everything it holds, with a discover naming it.
True when it was asked.
Skipped while a discover to every device is recent enough to have
covered it. A receiver that predates the named form answers it as a
discover to every device, so each one asked costs the whole mesh's state
on such a network.
'''
if ((time.time() - self._discover_timeout) < _DISCOVER_COVERS):
return False
pkt = FrameDiscover.construct(mxr=self, target=device)
if (pkt is None):
return False
_LOGGER.debug(f"asking {self.uid_to_user_string(device)} for its state")
self.transmit(pkt.frame)
return True
def _send_to_all(self, frame:FrameBase|None) -> bool:
'''Send a frame to every device, True when all of it was written.'''
if (frame is None):
return False
return (self.transmit(frame.frame) == len(frame.frame))
[docs]
async def set_all_oneip_device_settings(self, settings:OneIPDeviceSettings) -> bool:
'''Change settings on every OneIP device of the mesh with one frame.
settings carries the settings behind their bits in valid, as a device
reports them. Each device applies those it has and ignores the rest,
and none below protocol 0x2A applies any. A setting only a device
reports about itself, a profile, PTP mode, PTP domain or power save
level out of range, a PTP DSCP above ONEIP_DSCP_MAX, a stream TTL outside
1 to 255, a rotation above ONEIP_SINK_ROTATION_MAX and a power save time
that is not a time of day are refused here, since every device would
ignore them. Each device takes a rotation only by its own step.
Nothing is cached: each device that applies a change reports its
settings, and its oneip_settings reads that.
'''
valid = settings.valid
why = None
if (int(valid) == 0):
why = 'no setting is carried'
elif (valid & ONEIP_DEVICE_SETTINGS_REPORTED_ONLY):
why = 'a setting only a device reports is carried'
elif (OneIPDeviceSetting.IR_PROFILE in valid) and not (0 <= (settings.ir_profile or 0) < ONEIP_IR_PROFILE_MAX):
why = 'no such infrared profile'
elif (OneIPDeviceSetting.IR_PROFILE_SINK in valid) \
and not (ONEIP_IR_PROFILE_NOT_SET <= (settings.ir_profile_sink or 0) < ONEIP_IR_PROFILE_MAX):
why = 'no such output infrared profile'
elif (OneIPDeviceSetting.AUTO_POWER_SAVE in valid) and not (0 <= (settings.auto_power_save or 0) <= 0xFFFF):
why = 'more idle minutes than the field holds'
elif (OneIPDeviceSetting.POWER_SAVE_SCHEDULE in valid) \
and not (settings.power_save_schedule or OneIPPowerSaveSchedule()).is_valid():
why = 'a power save time is not a time of day'
elif (OneIPDeviceSetting.PTP_MODE in valid) \
and not (OneIPPtpMode.OFF <= (settings.ptp_mode or 0) <= OneIPPtpMode.AUTO):
why = 'no such PTP mode'
elif (OneIPDeviceSetting.PTP_DOMAIN in valid) \
and not (0 <= (settings.ptp_domain or 0) <= ONEIP_PTP_DOMAIN_MAX):
why = 'no such PTP domain'
elif (OneIPDeviceSetting.PTP_PRIORITY1 in valid) and not (0 <= (settings.ptp_priority1 or 0) <= 0xFF):
why = 'a PTP priority1 the field cannot hold'
elif (OneIPDeviceSetting.POWER_SAVE_LEVEL in valid) \
and not (OneIPPowerSaveLevel.NONE <= (settings.power_save_level or 0) <= OneIPPowerSaveLevel.OFF):
why = 'no such power save level'
elif (OneIPDeviceSetting.PTP_DSCP in valid) \
and not (0 <= (settings.ptp_messages or OneIPPtpMessages()).dscp <= ONEIP_DSCP_MAX):
why = 'a PTP DSCP above 63'
elif (OneIPDeviceSetting.STREAM_TTL in valid) \
and not (1 <= (settings.ptp_messages or OneIPPtpMessages()).stream_ttl <= 0xFF):
why = 'a stream TTL outside 1 to 255'
elif (OneIPDeviceSetting.SINK_ROTATION in valid) \
and not (0 <= (settings.ptp_messages or OneIPPtpMessages()).sink_rotation <= ONEIP_SINK_ROTATION_MAX):
why = f'a rotation above {ONEIP_SINK_ROTATION_MAX} degrees'
if (why is not None):
_LOGGER.warning(f"not changing the settings of every OneIP device: {why}")
return False
return self._send_to_all(FrameV2IPSettingsAll.construct(mxr=self, settings=settings))
[docs]
async def set_all_oneip_power_save(self, enabled:bool) -> bool:
'''Switch power save on or off on every OneIP device of the mesh with
one frame, written as Device.set_oneip_power_save() writes it.
Otherwise as set_all_oneip_device_settings().'''
return await self.set_all_oneip_device_settings(OneIPDeviceSettings.power_save_switch(enabled))
[docs]
async def request_all_oneip_privacy(self) -> bool:
'''Ask every OneIP source unit of the mesh how its streams are
encrypted, as Device.request_oneip_privacy() asks one. A unit below
protocol 0x30 ignores it.'''
return self._send_to_all(FrameV2IPPrivacy.construct_request(mxr=self, target=None, target_uid=None))
[docs]
async def set_mesh_time_zone(self, zone:str, rule:str) -> bool:
'''Set the time zone of every device that hears it.
zone is an IANA name such as Europe/Amsterdam and rule the POSIX TZ rule
the devices keep time by, such as CET-1CEST,M3.5.0,M10.5.0/3. A device
applies the rule and shows the name. Neither may be empty, hold a NUL,
or be longer than its field on the wire leaves room for;
clear_mesh_time_zone() sends both empty.
The mesh controller takes it too, and announces it from then on with
every periodic broadcast. A device takes it only from a management
application or the controller, and this client announces itself as a
management application.
'''
frame = FrameTimeZone.construct(mxr=self, zone=zone, rule=rule)
if (frame is None):
_LOGGER.warning(f"not setting time zone {zone!r} {rule!r}: empty, holding a NUL, or too long")
return False
return self._send_to_all(frame)
[docs]
async def clear_mesh_time_zone(self) -> bool:
'''Clear the time zone of every device that hears it, which then keeps
UTC. Sent, received and announced as set_mesh_time_zone() with an empty
name and rule.'''
return self._send_to_all(FrameTimeZone.construct_clear(mxr=self))
[docs]
async def set_mesh_time(self, when:datetime|None=None) -> bool:
'''Set the clock of every device that hears it to when, or to now.
A device keeps its own clock where that is within 2s of when. It takes
the time only from a management application or the controller; see
set_mesh_time_zone(). A time before 1970 or past 2106, which 32 bits of
seconds do not hold, is refused.
'''
frame = FrameTime.construct(mxr=self, when=(when if (when is not None) else datetime.now(timezone.utc)))
if (frame is None):
_LOGGER.warning(f"not setting the mesh time to {when}: it does not fit the frame")
return False
return self._send_to_all(frame)
def _arm_hello(self) -> None:
'''Schedule the next announcement, mirroring the firmware's hello timer.
MatrixOS re-draws min 2.5s + random(0..2.5s) after each hello it sends,
so a mesh of devices does not synchronise into bursts. The interval is
re-drawn per send rather than fixed for the same reason.
MatrixOS stretches that to 20-30s once every device it knows is on 0x2A.
This client keeps the short interval: a peer below 0x2A takes it offline
15s after its last hello.
'''
self._hello_due = time.time() + MXR_HELLO_INTERVAL_MIN \
+ (random.random() * MXR_HELLO_INTERVAL_RAND)
[docs]
def tx_hello(self) -> int:
'''Transmit a hello frame to announce this device on the network.'''
pkt = constructFrameHello(self)
if (pkt is None):
return 0
_LOGGER.debug("sending hello")
sent = self.transmit(pkt.frame)
if (sent > 0):
# Only a hello that went out buys the next interval. A send that wrote
# nothing leaves _hello_due where it was, so the next probe tick tries
# again, which is what firmware does - it re-arms its own timer only
# after a send succeeds. Arming here on failure instead
# would silence a client for a full interval every time a write failed,
# and on an interface that never writes, permanently.
self._arm_hello()
return sent
[docs]
def on_connection_made(self) -> None:
'''Callback invoked after ConnectionAsync has started the server.'''
self.tx_hello()
self.tx_discover()
[docs]
def process_frame(self, timestamp:float, data: bytes, addr: tuple[str, int]) -> None:
'''Decode and process a received mx_remote frame.
A device that may hold state this client missed is left in
_state_request rather than asked here, so replaying a capture sends
nothing; on_datagram_received does the asking.'''
proc = False
self._state_request = None
try:
frame = process_mxr_frame(mxr=self, timestamp=timestamp, data=data, addr=addr)
if (frame is not None) and (frame.header.protocol > MXR_PROTOCOL_VERSION):
# The stamp is the version at which this opcode's payload last
# changed, not the sender's version. Above ours means this
# opcode's layout is newer than we know: the frames we would
# decode into wrong state, and the ones devices drop before
# dispatch.
#
# A release raises the stamp only where a layout moved, so this
# silences those opcodes until the library is bumped, not the
# mesh.
_LOGGER.debug(f"rx {addr[0]}: opcode {frame.header.opcode:02X} needs protocol "
f"{frame.header.protocol:02X}, this build knows {MXR_PROTOCOL_VERSION:02X}")
frame = None
if (frame is not None) and ((len(data) != 24 + frame.header.payload_len)
or (frame.header.protocol == _BROKEN_PROTOCOL)):
# A unit drops a datagram whose length is not the header's
# exactly, and a frame stamped with the version a broken
# revision once stamped.
_LOGGER.debug(f"rx {addr[0]}: opcode {frame.header.opcode:02X} dropped: {len(data)} bytes "
f"declaring {frame.header.payload_len}, stamped {frame.header.protocol:02X}")
frame = None
if (frame is not None) and (frame.header.remote_id != self.uid):
# A device announces itself before it sends anything else, so a
# frame from a uid with no record is one no device would act on.
# Decode and report it - this library is also used to watch the
# bus - but keep it out of the cache and the callbacks.
#
# Hello, discover and ping are exempt: the hello is what makes
# a sender known, so gating it leaves it unknown forever, and a
# discover or a ping is how a stranger asks to be told.
proc = (frame.remote_device is not None) \
or (frame.header.opcode in _ACCEPT_UNANNOUNCED)
if (self._addr_filter is None) or (addr[0] == self._addr_filter):
ts = f'[{timestamp}] ' if (self.conn is None) else ''
skipped = '' if proc else ' (sender has not announced)'
_LOGGER.debug(f"{ts}rx {addr[0]}: {frame.header.opcode:02X}({len(frame)}) - {str(frame)}{skipped}")
except Exception:
_LOGGER.warning(f"failed to decode frame {traceback.format_exc()}")
raise
try:
if (frame is not None) and proc:
known = self.remotes.get(frame.remote_id)
before = None if (known is None) else _heard(known)
frame.process()
device = self.remotes.get(frame.remote_id)
if (device is not None) and _missed_state(before, device):
self._state_request = frame.remote_id
except Exception:
_LOGGER.warning(f"failed to process frame: {traceback.format_exc()}")
raise
[docs]
def on_datagram_received(self, data: bytes, addr: tuple[str, int]) -> None:
'''Called when a UDP frame was received.'''
timestamp = time.time()
self.process_frame(timestamp=timestamp, data=data, addr=addr)
if (self._state_request is not None):
self._request_state(self._state_request)
# ---- overlay sync ----
@property
def osd_sync_held(self) -> bool:
'''Whether this client holds an overlay sync, and so announces itself as
the mesh's overlay sync manager.'''
return self._osd.holds
[docs]
async def send_osd_sync(self, content:OsdSyncContent) -> int|None:
'''Have every unit copy content, and keep it in place, returning the
sync's number, or None when it was refused or did not go out.
Units take a sync from the mesh controller or a management application,
which this client announces itself as. Each unit fetches the images
from a server this client runs, then asks for the settings, which this
client answers until the content is replaced or clear_osd_sync(). Meanwhile
it announces itself as the mesh's overlay sync manager, so a unit waking
from power save takes this content too, rather than the controller's.
The controller can still send its own over it; the last sync a unit
took is what it shows. Each unit's osd_sync_report says how it went.
Refused for content a unit would refuse: copying nothing, more than
OSD_SYNC_RECORDS_MAX settings, a caption font over OSD_SYNC_FONT_MAX or
text over OSD_SYNC_TEXT_MAX bytes, a logo other than -1 to 4, or an image
that is not a PNG within OSD_SYNC_IMAGE_MAX bytes and
OSD_SYNC_IMAGE_WIDTH_MAX by OSD_SYNC_IMAGE_HEIGHT_MAX pixels, shown at
over OSD_SYNC_PERCENT_MAX percent. Refused content leaves the held sync as
it was; content that fails to go out leaves none held.
'''
if ((why := osd_content_refusal(content)) is not None):
_LOGGER.warning(f"not sending an overlay sync: {why}")
return None
http_port = 0
if (content.images is not None):
try:
await self._logos.start(self._logo_port or 0)
except OSError as e:
_LOGGER.warning(f"not sending an overlay sync: the logo server did not start: {e}")
await self.clear_osd_sync()
return None
self._logos.images = tuple(image.png for image in content.images)
http_port = self._logos.port
was_held = self._osd.holds
self._osd.hold(content, http_port)
# Announced first, so the controller has stood back by the time a unit
# deferring this sync wakes and asks for it.
if not was_held:
self.tx_hello()
seq = self._tx_osd_sync(MxrDeviceUid(bytes(16)), None, None)
if (seq is None):
await self.clear_osd_sync()
return seq
[docs]
async def resend_osd_sync(self, device:DeviceBase) -> int|None:
'''Have one unit copy the held sync again, returning the new sync's
number. As send_osd_sync().
Refused while no sync is held, and for a unit that does not announce
DeviceFeature.OSD_SYNC, as one that does not take overlay syncs.'''
if (device.features is None) or (DeviceFeature.OSD_SYNC not in device.features):
_LOGGER.warning(f"not resending an overlay sync to {device}: it takes none")
return None
if not self._osd.holds:
_LOGGER.warning(f"not resending an overlay sync to {device}: none is held")
return None
return self._tx_osd_sync(device.remote_id, None, device)
[docs]
async def clear_osd_sync(self) -> None:
'''Stop holding the sync this client sent, and hand the overlay sync
manager's part back to the controller. Units keep what they took; one
still taking it fails, as its requests go unanswered.'''
await self._logos.stop()
if self._osd.drop():
self.tx_hello()
def _tx_osd_sync(self, target:MxrDeviceUid, what:OsdSyncWhat|None, to:DeviceBase|None) -> int|None:
'''Number and send a new sync of the held content.'''
if ((sync := self._osd.new_sync(what)) is None):
return None
seq, what, http_port, records, slots = sync
frame = FrameV2IPOsdSync.construct_sync(mxr=self, target=target, seq=seq, what=what, http_port=http_port,
records=records, slots=slots, to=to)
if (frame is None) or (self.transmit(frame.frame) != len(frame.frame)):
return None
return seq
def _answer_osd_request(self, request:OsdSyncRequest) -> None:
'''Answer a unit's overlay sync request: queue the settings it lacks, or
send it a new sync of its own.'''
what = self._osd.answer(request)
if (self._osd.queued > 0):
self._osd_queued.set()
if (what is not None) and ((unit := self.get_by_uid(request.target)) is not None):
self._tx_osd_sync(request.target, what, unit)
def _osd_pass(self) -> int:
'''Send the next few settings units asked for, returning how many.'''
sent = 0
for unit, seq, index, records, entry in self._osd.next_pass():
frame = FrameV2IPOsdSync.construct_setting(mxr=self, target=unit, seq=seq, index=index,
records=records, entry=entry, to=self.get_by_uid(unit))
if (frame is not None):
self.transmit(frame.frame)
sent += 1
return sent
async def _osd_loop(self) -> None:
'''Send settings as units ask for them, until the client is closing.'''
while not self._closing:
try:
if (self._osd_pass() > 0):
await asyncio.sleep(PASS_GAP)
continue
except Exception:
_LOGGER.warning(f"overlay sync pass failed: {traceback.format_exc()}")
self._osd_queued.clear()
if (self._osd.queued == 0):
try:
await asyncio.wait_for(self._osd_queued.wait(), timeout=1)
except asyncio.TimeoutError:
pass
@override
def on_mxr_update(self, data:Any) -> None:
if isinstance(data, OsdSyncRequest):
self._answer_osd_request(data)
elif isinstance(data, FrameHello):
self._on_mxr_hello(data)
elif isinstance(data, (FramePing, FrameDiscover)):
# Announcing is driven by a clock, and these are the frames that add
# a hello to it. A device that pings hears nothing back within about
# 1.5s before it takes this client offline, and one that discovers
# drops this client's frames until it has had a hello; the next
# scheduled one can be seconds away.
self.tx_hello()
def _on_mxr_hello(self, hello_frame:FrameHello) -> None:
'''Hello frame received. Register or update the local device cache.'''
d = self.get_by_uid(hello_frame.remote_id)
if d is None:
d = Device(self, hello_frame)
self.remotes[hello_frame.remote_id] = d
d.on_mxr_update(hello_frame)
# The V2IP spelling of a Remote method, from before the rename to OneIP.
from ..deprecated import alias_members as _alias_members
_alias_members(Remote, 'set_all_v2ip_device_settings')