Changeset - 03d2f8c50a34
[Not reviewed]
default
0 1 0
drewp@bigasterisk.com - 20 months ago 2023-05-22 08:00:47
drewp@bigasterisk.com
type fixes
1 file changed with 13 insertions and 10 deletions:
0 comments (0 inline, 0 general)
light9/effect/simple_outputs.py
Show inline comments
 
import logging
 
import traceback
 
from typing import Any, Dict, List, Tuple
 
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[URIRef, List[Tuple[URIRef, URIRef, Any, bool]]] = {}
 
        self.effectOutputs: Dict[EffectUri, List[Tuple[DeviceUri, DeviceAttr, VTUnion, bool]]] = {}
 

	
 
        self.graph.addHandler(self.updateEffectsFromGraph)
 

	
 
    def updateEffectsFromGraph(self):
 
        self.effectOutputs = {}
 
        self.effectOutputs.clear()
 
        for effect in self.graph.subjects(RDF.type, L9['Effect']):
 
            log.debug(f' {effect=}')
 
            settings = []
 
            settings:List[Tuple[DeviceUri, DeviceAttr, VTUnion, bool]] = []
 
            for setting in self.graph.objects(effect, L9['setting']):
 
                settingValues = dict(self.graph.predicate_objects(setting))
 
                try:
 
                    d = settingValues.get(L9['device'], None)
 
                    a = settingValues.get(L9['deviceAttr'], None)
 
                    v = settingValues.get(L9['value'], None)
 
                    sv = settingValues.get(L9['scaledValue'], None)
 
                    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
 

	
 
                settings.append((d, a, v if v is not None else sv, bool(sv)))
 
                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
0 comments (0 inline, 0 general)