Source code for tallypi.outputs.umd

from loguru import logger
import asyncio
from dataclasses import dataclass
from typing import Optional, Tuple, Iterable, Set, Dict, Union

from tslumd import UmdSender, TallyType, Screen, Tally, TallyKey
from tslumd.sender import Client

from tallypi.common import MultiTallyConfig, MultiTallyOption
from tallypi.baseio import BaseOutput
from tallypi.config import Option, ListOption
from tallypi.utils import SetProperty

[docs]@dataclass(frozen=True) class ClientData: """Container for network client data """ hostaddr: str #: The network host address hostport: int #: The port number @classmethod def get_init_options(cls) -> Tuple[Option]: return ( Option(name='hostaddr', type=str, required=True, title='Host Address'), Option(name='hostport', type=int, required=True, title='Host Port'), ) @classmethod def from_tuple(cls, client: Client) -> 'ClientData': addr, port = client return ClientData(hostaddr=addr, hostport=port) @property def as_tuple(self) -> Client: """The client data as a tuple of (:attr:`hostaddr`, :attr:`hostport`) """ return (self.hostaddr, self.hostport)
ClientsOption = ListOption( name='clients', type=ClientData, required=False, sub_options=ClientData.get_init_options(), title='Clients', ) ClientOrData = Union[Client, ClientData] """:data:`~tslumd.sender.Client` or :class:`ClientData` """ INDICATOR_PROPS = { tt for tt in TallyType.all()}
[docs]class UmdOutput(BaseOutput, namespace='umd.UmdOutput', final=True): """Networked tally output using the UMDv5 protocol Arguments: config (MultiTallyConfig): The initial value for :attr:`~tallypi.baseio.BaseIO.config` clients (Iterable[ClientOrData], optional): The initial :attr:`clients` to set all_off_on_close: (bool, optional): Value to set for :attr:`all_off_on_close` Properties: clients (set): A :class:`~.utils.SetProperty` containing the remote host addresses as address/port tuples. """ sender: UmdSender """The :class:`tslumd.sender.UmdSender` instance """ clients = SetProperty(copy_on_change=True) def __init__( self, config: MultiTallyConfig, clients: Optional[Iterable[ClientOrData]] = None, all_off_on_close: Optional[bool] = False, ) -> None: super().__init__(config) self.sender = UmdSender(all_off_on_close=all_off_on_close) self.bind(clients=self._on_clients_changed) if clients is not None: for c in clients: self.add_client(c) @property def all_off_on_close(self) -> bool: """Alias for :attr:`tslumd.sender.UmdSender.all_off_on_close`. If ``True``, the sender will turn all tally indicators off before closing. Default is ``False`` """ return self.sender.all_off_on_close @all_off_on_close.setter def all_off_on_close(self, value: bool): self.sender.all_off_on_close = value
[docs] @classmethod def get_init_options(cls) -> Tuple[Option]: return ( MultiTallyOption, ClientsOption, Option( name='all_off_on_close', type=bool, required=False, default=False, ) )
[docs] async def open(self): if self.running: return await self.running = True
[docs] async def close(self): if not self.running: return await self.sender.close()
[docs] def add_client(self, client: ClientOrData): """Add an item to :attr:`clients` """ if not isinstance(client, ClientData): client = ClientData.from_tuple(client) self.clients.add(client)
[docs] def remove_client(self, client: ClientOrData): """Remove an item from :attr:`clients` """ if not isinstance(client, ClientData): client = ClientData.from_tuple(client) self.clients.discard(client)
[docs] @logger.catch async def on_receiver_tally_change(self, tally: Tally, props_changed: Set[str], **kwargs): if not self.running: return if not self.tally_matches(tally): return for prop, tally_type in INDICATOR_PROPS.items(): if prop not in props_changed: continue match = self.tally_matches(tally, tally_type, return_matched=True) if not match: continue color = self.get_merged_tally(tally, tally_type) self.sender.set_tally_color(, tally_type, color)
def _on_clients_changed(self, instance, value, **kwargs): cl_tuples = set([c.as_tuple for c in value]) self.sender.clients &= cl_tuples self.sender.clients |= cl_tuples