Files @ 74b4acd3dde0
Branch filter:

Location: light9/light9/effect/simple_outputs.py

drewp@bigasterisk.com
collector client retries a bit. not sure we want this.
import logging
import traceback
from typing import Any, Dict, List, Tuple, cast
from light9.newtypes import DeviceAttr, DeviceUri, EffectUri, VTUnion
from light9.typedgraph import typedValue

from rdflib import URIRef

from light9.effect.scale import scale
from light9.namespaces import L9, RDF

log = logging.getLogger('simple')


class SimpleOutputs:
    """
    Watches graph for effects that are just fading output attrs.
    Call `values` to get (dev,attr):value settings.

    The alternative to 'simple' is 'custom code' (see effecteval.py)
    """

    def __init__(self, graph):
        self.graph = graph

        # effect : [(dev, attr, value, isScaled)]
        self.effectOutputs: Dict[EffectUri, List[Tuple[DeviceUri, DeviceAttr, VTUnion, bool]]] = {}

        self.graph.addHandler(self.updateEffectsFromGraph)

    def updateEffectsFromGraph(self):
        self.effectOutputs.clear()
        for effect in self.graph.subjects(RDF.type, L9['Effect']):
            log.debug(f' {effect=}')
            settings:List[Tuple[DeviceUri, DeviceAttr, VTUnion, bool]] = []
            for setting in self.graph.objects(effect, L9['setting']):
                try:
                    d = typedValue(DeviceUri, self.graph, setting, L9['device'])
                    a = typedValue(DeviceAttr, self.graph, setting, L9['deviceAttr'])
                    v = typedValue(VTUnion | None, self.graph, setting, L9['value'])
                    sv = typedValue(VTUnion|None, self.graph, setting, L9['scaledValue'])
                    if not (bool(v) ^ bool(sv)):
                        raise NotImplementedError('no value for %s' % setting)
                    if d is None:
                        raise TypeError('no device on %s' % effect)
                    if a is None:
                        raise TypeError('no attr on %s' % effect)
                except Exception:
                    traceback.print_exc()
                    continue

                settingValue = cast(VTUnion, v if v is not None else sv)
                settings.append((d, a, settingValue, bool(sv)))
                log.debug(f'   effect {effect} has {settings=}')
            if settings:
                self.effectOutputs[effect] = settings
            # also have to read eff :effectAttr [ :tint x; :tintStrength y ]

    def values(self, effect, strength, colorScale):
        out = {}
        for dev, devAttr, value, isScaled in self.effectOutputs.get(effect, []):
            if isScaled:
                value = scale(value, strength)
            if colorScale is not None and devAttr == L9['color']:
                value = scale(value, colorScale)
            out[(dev, devAttr)] = value
        return out