Source code for jvconnected.interfaces.paramspec

from __future__ import annotations
from typing import List, Dict, Any, ClassVar, Iterator, Optional
from dataclasses import dataclass, field

from pydispatch import Dispatcher, Property, ListProperty

[docs]@dataclass class Value: """Base class for value definitions """ py_type = str
[docs]@dataclass class BoolValue(Value): """A boolean value definition """ py_type = bool
[docs]@dataclass class IntValue(Value): """A numeric value definition """ py_type = int value_min: int = 0 """Minimum value of the parameter""" value_max: int = 255 """Maximum value of the parameter""" value_default: int = 0 """Default or natural center of value range"""
[docs]@dataclass class ChoiceValue(Value): """A string value definition with a defined set of choices """ py_type = str choices: List[str] = field(default_factory=list) """The expected string values for the parameter"""
[docs]class BaseParameterSpec(Dispatcher): """Base Parameter Spec """
[docs] def on_device_value_changed(self, param: 'ParameterSpec', value: Any): """Fired when the parameter value has changed on the device """
_doc_field_names: ClassVar[List[str]] = [] group_name: str = '' """The name of the :class:`~jvconnected.device.ParameterGroup` as accessed by the ``parameter_groups`` attribute of :class:`~jvconnected.device.Device` """ name: str = '' """The parameter name, typically the same as :attr:`prop_name`""" full_name: str = '' """Combination of :attr:`group_name` and :attr:`name`""" setter_method: str = '' """Method name on the :class:`~jvconnected.device.ParameterGroup` used to set the parameter value (if available) """ adjust_method: str = '' """Method name on the :class:`~jvconnected.device.ParameterGroup` used to increment/decrement the parameter value such as :meth:`jvconnected.device.ExposureParams.adjust_iris` """ _events_ = ['on_device_value_changed'] def __init__(self, **kwargs): self._param_group_spec = None self._device_param_group = None self.group_name = kwargs.get('group_name', '') self.name = kwargs.get('name', '') self.full_name = kwargs.get('full_name', '') self.setter_method = kwargs.get('setter_method', '') self.adjust_method = kwargs.get('adjust_method', '') @property def param_group_spec(self) -> Optional['ParameterGroupSpec']: """The parent :class:`ParameterGroupSpec` instance """ return self._param_group_spec @param_group_spec.setter def param_group_spec(self, pgs: Optional['ParameterGroupSpec']): self._param_group_spec = pgs if pgs is None: self.device_param_group = None else: self.device_param_group = pgs.device_param_group @property def device_param_group(self) -> Optional['jvconnected.device.ParameterGroup']: """The :class:`jvconnected.device.ParameterGroup` bound to this instance """ return self._device_param_group @device_param_group.setter def device_param_group(self, pg: Optional['jvconnected.device.ParameterGroup']): old = self.device_param_group if old is not None: old.unbind(self) self._device_param_group = pg if pg is not None: self._bind_to_param_group(pg) def _bind_to_param_group(self, pg: 'jvconnected.device.ParameterGroup'): pass def copy(self) -> 'BaseParameterSpec': kw = self._build_copy_kwargs() cls = self.__class__ return cls(**kw) def _build_copy_kwargs(self) -> Dict: attrs = ['group_name', 'name', 'full_name', 'setter_method', 'adjust_method'] return {attr:getattr(self, attr) for attr in attrs}
[docs] async def increment_value(self): """Increment the device value """ await self.adjust_param_value(True)
[docs] async def decrement_value(self): """Decrement the device value """ await self.adjust_param_value(False)
[docs] async def adjust_value(self, direction: bool): """Increment or decrement the device value Arguments: direction (bool): If True, increment, otherwise decrement Raises: ValueError: If there is no :attr:`adjust_method` defined """ pg = self.device_param_group if not self.adjust_method: raise ValueError(f'No adjust method for {self}') m = getattr(pg, self.adjust_method) await m(direction)
def build_docstring_lines(self, indent=0): indent_str = ' ' * indent lines = [] # attrs = ['full_name', 'prop_name', 'value_type', 'setter_method', 'adjust_method'] for attr in self._doc_field_names: val = getattr(self, attr) if val is None or val == '': continue valstr = str(val) if attr == 'value_type': clsname = str(val.__class__).rstrip("'>").split('.')[-1] valstr = f':class:`{clsname}`' if isinstance(val, ChoiceValue): valstr = f'{valstr}: ``choices={val.choices}``' elif isinstance(val, IntValue): valstr = f'{valstr}: ``value_min={val.value_min}, value_max={val.value_max}``' else: valstr = f'``"{valstr}"``' s = f'{indent_str}* **{attr}**: {valstr}' lines.append(s) return lines
[docs]class ParameterSpec(BaseParameterSpec): """Specifications for a single parameter within a :class:`jvconnected.device.ParameterGroup` """ _doc_field_names: ClassVar[List[str]] = [ 'full_name', 'prop_name', 'value_type', 'setter_method', 'adjust_method', ] prop_name: str = '' """The Property/attribute name within the :class:`jvconnected.device.ParameterGroup` containing the parameter value """ value_type: Value = field(default_factory=Value) """Specifications for the expected value of the attribute in :class:`~jvconnected.device.ParameterGroup` One of :class:`BoolValue`, :class:`IntValue`, or :class:`ChoiceValue` """ value: Any = Property() """The current device value""" def __init__(self, **kwargs): super().__init__(**kwargs) self.prop_name = kwargs.get('prop_name', '') self.value_type = kwargs['value_type'] if not len(self.prop_name): self.prop_name = self.name def _build_copy_kwargs(self) -> Dict: kw = super()._build_copy_kwargs() attrs = ['prop_name', 'value_type'] kw.update({attr:getattr(self, attr) for attr in attrs}) return kw def _bind_to_param_group(self, pg: 'jvconnected.device.ParameterGroup'): self.value = self.get_param_value() pg.bind(**{self.prop_name:self.on_device_prop_change}) def on_device_prop_change(self, instance, value, **kwargs): if instance is not self.device_param_group: return prop = kwargs['property'] assert prop.name == self.prop_name assert value == self.get_param_value() self.value = value self.emit('on_device_value_changed', self, self.value, prop_name=prop.name, value_type=self.value_type, )
[docs] def get_param_value(self) -> Any: """Get the current device value """ pg = self.device_param_group value = getattr(pg, self.prop_name) if value is not None: assert isinstance(value, self.value_type.py_type) return value
[docs] async def set_param_value(self, value: Any): """Set the device value Raises: ValueError: If no :attr:`setter_method` is defined """ pg = self.device_param_group if self.setter_method: m = getattr(pg, self.setter_method) await m(value) else: raise ValueError(f'No setter method for {self}')
[docs]class MultiParameterSpec(BaseParameterSpec): """Combines multiple :class:`ParameterSpec` definitions """ _doc_field_names: ClassVar[List[str]] = [ 'full_name', 'prop_names', 'value_types', 'setter_method', 'adjust_method', ] prop_names: List[str]# = field(default_factory=list) """The Property/attribute names within the :class:`jvconnected.device.ParameterGroup` containing the parameter values """ value_types: List[Value]# = field(default_factory=list) """Specifications for the expected values of the attribute in :class:`~jvconnected.device.ParameterGroup` """ value: List[Any] = ListProperty() """The current device value""" def __init__(self, **kwargs): super().__init__(**kwargs) self.prop_names = kwargs['prop_names'] self.value_types = kwargs['value_types'] self.value = [vt.py_type for vt in self.value_types] def _build_copy_kwargs(self) -> Dict: kw = super()._build_copy_kwargs() attrs = ['prop_name', 'value_type'] kw.update({attr:getattr(self, attr) for attr in attrs}) return kw def _bind_to_param_group(self, pg: 'jvconnected.device.ParameterGroup'): pg.bind(**{self.prop_name:self.on_device_prop_change}) def on_device_prop_change(self, instance, value, **kwargs): if instance is not self.device_param_group: return prop = kwargs['property'] assert prop.name == self.prop_name i = self.prop_names.index(prop.name) vtype = self.value_types[i] assert isinstance(value, vtype.py_type) self.value[i] = value self.emit('on_device_value_changed', self, self.value, prop_name=prop.name, value_type=vtype, )
[docs] def get_param_value(self) -> List[Any]: """Get the current device value """ pg = self.device_param_group value = [getattr(pg, key) for key in self.prop_names] for vtype, item in zip(self.value_types, value): assert isinstance(item, vtype) return value
[docs] async def set_param_value(self, value: List[Any]): """Set the device value Raises: ValueError: If no :attr:`setter_method` is defined """ pg = self.device_param_group if self.setter_method: m = getattr(pg, self.setter_method) await m(*value) else: raise ValueError(f'No setter method for {self}')
[docs]class ParameterGroupSpec(Dispatcher): """A group of :class:`ParameterSpec` definitions for a :class:`jvconnected.device.ParameterGroup` that can be attached to an existing :class:`~jvconnected.device.Device` instance. Arguments: device (jvconnected.device.Device): A :class:`~jvconnected.device.Device` instance to attach to Attributes: name (str): The name of the :class:`jvconnected.device.ParameterGroup` within its parent :attr:`device` device: Instance of :class:`jvconnected.device.Device` device_param_group: The :class:`jvconnected.device.ParameterGroup` associated with this instance """ name: ClassVar[str] parameter_list: ClassVar[List[ParameterSpec]] parameters: Dict[str, ParameterSpec]
[docs] def on_device_value_changed( self, group: 'ParameterGroupSpec', param: ParameterSpec, value: Any ): """Fired when the value of a parameter has changed on the device """
_events_ = ['on_device_value_changed'] def __init_subclass__(cls): super().__init_subclass__() doc_lines = cls.__doc__.splitlines() doc_lines.extend([ '', ':Parameter Definitions:', ]) for param in cls.parameter_list: param.group_name = cls.name param.full_name = f'{cls.name}.{param.name}' param_lines = [ f' {param.name}', ] param_lines.extend(param.build_docstring_lines(8)) param_lines.append('') doc_lines.extend(param_lines) cls.parameters = {p.name:p for p in cls.parameter_list} cls.__doc__ = '\n'.join(doc_lines) def __init__(self, device: 'jvconnected.device.Device'): self.device = device self.device_param_group = device.parameter_groups[self.name] self.parameter_list = [p.copy() for p in self.parameter_list] self.parameters = {p.name:p for p in self.parameter_list} for p in self.parameter_list: p.param_group_spec = self p.bind(on_device_value_changed=self.on_param_spec_value_changed) multi_params = [p for p in self.parameters.values() if isinstance(p, MultiParameterSpec)] self.multi_params = {} for param in multi_params: for prop in param.prop_names: self.multi_params[prop] = param
[docs] @classmethod def all_parameter_group_cls(cls) -> Iterator['ParameterGroupSpec']: """Iterate through all ParameterGroupSpec subclasses """ def iter_subcls(_cls): if _cls is not ParameterGroupSpec: yield _cls for subcls in _cls.__subclasses__(): yield from iter_subcls(subcls) # return [c for c in iter_subcls(ParameterGroupSpec)] yield from iter_subcls(ParameterGroupSpec)
[docs] @classmethod def find_parameter_group_cls(cls, name: str) -> 'ParameterGroupSpec': """Search for a ParameterGroupSpec class by its :attr:`name` """ for _cls in cls.all_parameter_group_cls(): if _cls.name == name: return _cls raise KeyError(f'No subclass found for "{name}"')
[docs] def get_param_value(self, name: str) -> Any: """Get the current device value for the given parameter Arguments: name (str): The :class:`ParameterSpec` name """ param = self.parameters[name] return param.get_param_value()
[docs] async def set_param_value(self, name: str, value: Any): """Set the device value for the given parameter Arguments: name (str): The :class:`ParameterSpec` name value: The value to set Raises: ValueError: If there is no setter method for the :class:`ParameterSpec` """ param = self.parameters[name] await param.set_param_value(value)
[docs] async def increment_param_value(self, name: str): """Increment the device value for the given parameter Arguments: name (str): The :class:`ParameterSpec` name """ await self.adjust_param_value(name, True)
[docs] async def decrement_param_value(self, name: str): """Decrement the device value for the given parameter Arguments: name (str): The :class:`ParameterSpec` name """ await self.adjust_param_value(name, False)
[docs] async def adjust_param_value(self, name: str, direction: bool): """Increment or decrement the device value for the given parameter Arguments: name (str): The :class:`ParameterSpec` name direction (bool): If True, increment, otherwise decrement Raises: ValueError: If there is no :attr:`~ParameterSpec.adjust_method` defined """ param = self.parameters[name] await param.adjust_value(direction)
def on_param_spec_value_changed(self, param: 'BaseParameterSpec', value: Any, **kwargs): self.emit('on_device_value_changed', self, param, value, **kwargs) def __getitem__(self, key): return self.parameters[key]
[docs]class ExposureParams(ParameterGroupSpec): """:class:`ParameterGroupSpec` definition for :class:`jvconnected.device.ExposureParams` """ name = 'exposure' parameter_list = [ ParameterSpec( name='mode', value_type=ChoiceValue( choices=['Auto', 'Manual', 'IrisPriority', 'ShutterPriority'], ), ), ParameterSpec( name='iris_mode', value_type=ChoiceValue( choices=['Manual', 'Auto', 'AutoAELock'], ), ), ParameterSpec( name='iris_pos', value_type=IntValue(), setter_method='set_iris_pos', ), ParameterSpec( name='gain_mode', value_type=ChoiceValue( choices=[ 'ManualL', 'ManualM', 'ManualH', 'AGC', 'AlcAELock', 'LoLux', 'Variable', ], ), ), ParameterSpec( name='gain_pos', value_type=IntValue(value_min=-6, value_max=24, value_default=0), adjust_method='adjust_gain', ), ParameterSpec( name='shutter_mode', value_type=ChoiceValue( choices=['Off', 'Manual', 'Step', 'Variable', 'Eei'], ), ), ParameterSpec( name='master_black_pos', value_type=IntValue(value_min=-50, value_max=50, value_default=0), adjust_method='adjust_master_black', ), ]
[docs]class PaintParams(ParameterGroupSpec): """:class:`ParameterGroupSpec` definition for :class:`jvconnected.device.PaintParams` """ name = 'paint' parameter_list = [ ParameterSpec( name='white_balance_mode', value_type=ChoiceValue( choices=[ 'Preset', 'A', 'B', 'Faw', 'FawAELock', 'Faw', 'Awb', 'OnePush', '3200K', '5600K', 'Manual', ], ), ), ParameterSpec( name='red_normalized', value_type=IntValue(value_min=-32, value_max=32, value_default=0), setter_method='set_red_pos', ), ParameterSpec( name='blue_normalized', value_type=IntValue(value_min=-32, value_max=32, value_default=0), setter_method='set_blue_pos', ), # MultiParameterSpec( # name='wb_pos', # prop_names = ['red_normalized', 'blue_normalized'], # value_types=[ # IntValue(value_min=-32, value_max=32, value_default=0), # IntValue(value_min=-32, value_max=32, value_default=0), # ], # setter_method='set_wb_pos', # ), ParameterSpec( name='detail_pos', value_type=IntValue(value_min=-10, value_max=10, value_default=0), adjust_method='adjust_detail', ), ]
[docs]class TallyParams(ParameterGroupSpec): """:class:`ParameterGroupSpec` definition for :class:`jvconnected.device.TallyParams` """ name = 'tally' parameter_list = [ ParameterSpec(name='program', value_type=BoolValue(), setter_method='set_program'), ParameterSpec(name='preview', value_type=BoolValue(), setter_method='set_preview'), ParameterSpec( name='tally_status', value_type=ChoiceValue( choices=['Off', 'Program', 'Preview'], ), ), ]