Source code for jvconnected.interfaces.tslumd.mapper

from loguru import logger
import asyncio
from typing import Dict, Tuple, Set, Optional
import dataclasses
from dataclasses import dataclass, field
import enum

import jsonfactory

from tslumd import TallyColor, TallyType, TallyState, TallyKey


[docs]@dataclass class TallyMap: """Map to a single :class:`tally type <tslumd.common.TallyType>` within a specific :class:`tslumd.tallyobj.Tally` by its index """ screen_index: int = 0 #: The :attr:`tslumd.tallyobj.Screen.index` tally_index: int = 0 #: The :attr:`~.tslumd.tallyobj.Tally.index` tally_type: TallyType = TallyType.no_tally #: The :class:`~tslumd.common.TallyType` @property def tally_key(self) -> TallyKey: return (self.screen_index, self.tally_index) def to_dict(self) -> Dict: attrs = ['screen_index', 'tally_index', 'tally_type'] return {attr:getattr(self, attr) for attr in attrs}
[docs]@dataclass class DeviceMapping: """Map the preview and program tallies from UMD to a :class:`jvconnected.device.Device` This only defines the mapping, the functionality itself is carried out by :class:`MappedDevice`. """ device_index: int """The :attr:`~jvconnected.device.Device.device_index` to associate with this mapping """ program: TallyMap = field(default_factory=TallyMap) """Definition for program tally """ preview: TallyMap = field(default_factory=TallyMap) """Definition for preview tally """ def to_dict(self) -> Dict: attrs = ['device_index', 'program', 'preview'] return {attr:getattr(self, attr) for attr in attrs}
[docs]class MappedDevice: """Link between :class:`~tslumd.Tally` objects and a :class:`jvconnected.device.Device` """ umd_io: 'jvconnected.interfaces.tslumd.umd_io.UmdIo' """:class:`.UmdIo` instance""" map: DeviceMapping """Mapping definitions for the device""" device: Optional['jvconnected.device.Device'] """The device instance""" program_tally: Optional[TallyMap] """The :class:`~tslumd.tallyobj.Tally` mapped to :attr:`jvconnected.device.TallyParams.program` """ preview_tally: Optional[TallyMap] """The :class:`~tslumd.tallyobj.Tally` mapped to :attr:`jvconnected.device.TallyParams.preview` """ tally_state: TallyState #: The current state have_tallies: bool def __init__(self, umd_io: 'jvconnected.interfaces.tslumd.umd_io.UmdIo', map: DeviceMapping): self.umd_io = umd_io self.map = map self.device = None self.program_tally = None self.preview_tally = None self.tally_state = TallyState.OFF self.have_tallies = False self.get_tallies()
[docs] @logger.catch async def set_device(self, device: Optional['jvconnected.device.Device']): """Set the :attr:`device` and update its tally state """ old = self.device if old is not None and old is not device: await old.tally.set_tally_light('Off') self.device = device if device is not None: await self.update_device_tally()
[docs] def get_tallies(self) -> bool: """Attempt to find the :class:`~tslumd.tallyobj.Tally` objects in the :attr:`umd_io` Returns: bool: ``True`` if an update is needed (a tally object either changed or was found) """ if self.have_tallies: return False loop = asyncio.get_event_loop() need_update = False have_tallies = True if self.map.program.tally_type == TallyType.no_tally: pgm = None else: pgm = self.umd_io.tallies.get(self.map.program.tally_key) if pgm is None: have_tallies = False if pgm is not self.program_tally: if self.program_tally is not None: self.program_tally.unbind(self) self.program_tally = pgm if pgm is not None: pgm.bind_async(loop, on_update=self.update_device_tally) need_update = True if self.map.preview.tally_type == TallyType.no_tally: pvw = None else: pvw = self.umd_io.tallies.get(self.map.preview.tally_key) if pvw is None: have_tallies = False if pvw is not self.preview_tally: if self.preview_tally is not None: self.preview_tally.unbind(self) self.preview_tally = pvw if pvw is not None and pvw is not pgm: pvw.bind_async(loop, on_update=self.update_device_tally) need_update = True self.have_tallies = have_tallies if have_tallies: logger.debug(f'{self.map}: program={pgm}, preview={pvw}') return need_update
[docs] def update_tally_state(self, *args, **kwargs): """Update the :attr:`tally_state` using both :attr:`program_tally` and :attr:`preview_tally`. Since they are mutually exclusive in the device, priority is given to :attr:`program_tally`. Returns: bool: ``True`` if the state changed """ if not self.have_tallies: self.get_tallies() if not self.have_tallies: return False pgm = self.program_tally pvw = self.preview_tally state = TallyState.OFF if self.map.program.tally_type != TallyType.no_tally and pgm is not None: value = getattr(pgm, self.map.program.tally_type.name) if value != TallyColor.OFF: state |= TallyState.PROGRAM if self.map.preview.tally_type != TallyType.no_tally and pvw is not None: value = getattr(pgm, self.map.preview.tally_type.name) if value != TallyColor.OFF: state |= TallyState.PREVIEW if state == self.tally_state: return False self.tally_state = state return True
[docs] @logger.catch async def update_device_tally(self, *args, **kwargs): """Update the tally state (using :meth:`update_tally_state`) and send changes to the :attr:`device` """ if self.device is None: return changed = self.update_tally_state() if not changed: return if TallyState.PROGRAM in self.tally_state: value = 'Program' elif TallyState.PREVIEW in self.tally_state: value = 'Preview' else: value = 'Off' await self.device.tally.set_tally_light(value)
def __repr__(self): return f'<{self.__class__.__name__}: {self}>' def __str__(self): return f'device_index: {self.map.device_index}'
@jsonfactory.register class JsonHandler(object): def cls_to_str(self, cls): if type(cls) is not type: cls = cls.__class__ modname = '.'.join(cls.__module__.split('.')[:-1]) return f'{modname}.{cls.__name__}' def str_to_cls(self, s): prefix = '.'.join(JsonHandler.__module__.split('.')[:-1]) if s.endswith('TallyType'): return TallyType if not s.startswith(prefix): return None for cls in [TallyMap, DeviceMapping, TallyType]: if s.endswith(cls.__name__): return cls def encode(self, o): if isinstance(o, (TallyMap, DeviceMapping)): d = o.to_dict() d['__class__'] = self.cls_to_str(o) return d elif isinstance(o, TallyType): d = { '__class__':self.cls_to_str(o), 'name':o.name, 'value':o.value, } return d def decode(self, d): if '__class__' in d: cls = self.str_to_cls(d['__class__']) if cls is not None: if cls is DeviceMapping: for key in ['program', 'preview']: if not isinstance(d[key], TallyMap): d[key] = self.decode(d[key]) elif cls is TallyMap: if not isinstance(d['tally_type'], TallyType): if isinstance(d['tally_type'], int): d['tally_type'] = TallyType(d['tally_type']) else: d['tally_type'] = self.decode(d['tally_type']) elif cls is TallyType: return getattr(TallyType, d['name']) del d['__class__'] return cls(**d) return d