Files
@ 54cf7034bee0
Branch filter:
Location: light9/light9/effect/effecteval2.py
54cf7034bee0
4.3 KiB
text/x-python
EE don't abort compile on partial data
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 | import traceback
import inspect
import logging
from dataclasses import dataclass
from typing import Callable, List, Optional
from rdfdb.syncedgraph.syncedgraph import SyncedGraph
from rdflib import RDF
from rdflib.term import Node
from light9.effect.effect_function_library import EffectFunctionLibrary
from light9.effect.settings import DeviceSettings, EffectSettings
from light9.namespaces import L9
from light9.newtypes import (DeviceAttr, DeviceUri, EffectAttr, EffectFunction, EffectUri, VTUnion)
from light9.typedgraph import typedValue
log = logging.getLogger('effecteval')
@dataclass
class Config:
effectFunction: EffectFunction
esettings: EffectSettings
devSettings: Optional[DeviceSettings] # the EffectSettings :effectAttr :devSettings item, if there was one
func: Callable
funcArgs: List[inspect.Parameter]
@dataclass
class EffectEval2:
"""Runs one effect code to turn EffectSettings (e.g. strength) into DeviceSettings"""
graph: SyncedGraph
uri: EffectUri
lib: EffectFunctionLibrary
config: Optional[Config] = None
def __post_init__(self):
self.graph.addHandler(self._compile)
def _compile(self):
self.config = None
if not self.graph.contains((self.uri, RDF.type, L9['Effect'])):
return
try:
effectFunction = typedValue(EffectFunction, self.graph, self.uri, L9['effectFunction'])
effSets = []
devSettings = None
for s in self.graph.objects(self.uri, L9['setting']):
attr = typedValue(EffectAttr, self.graph, s, L9['effectAttr'])
if attr == L9['deviceSettings']:
value = typedValue(Node, self.graph, s, L9['value'])
rows = []
for ds in self.graph.objects(value, L9['setting']):
d = typedValue(DeviceUri, self.graph, ds, L9['device'])
da = typedValue(DeviceAttr, self.graph, ds, L9['deviceAttr'])
v = typedValue(VTUnion, self.graph, ds, L9['value'])
rows.append((d, da, v))
devSettings = DeviceSettings(self.graph, rows)
else:
value = typedValue(VTUnion, self.graph, s, L9['value'])
effSets.append((self.uri, attr, value))
esettings = EffectSettings(self.graph, effSets)
try:
effectFunction = typedValue(EffectFunction, self.graph, self.uri, L9['effectFunction'])
except ValueError:
raise ValueError(f'{self.uri} has no :effectFunction')
func = self.lib.getFunc(effectFunction)
# This should be in EffectFunctionLibrary
funcArgs = list(inspect.signature(func).parameters.values())
self.config = Config(effectFunction, esettings, devSettings, func, funcArgs)
except Exception:
log.error(f"while compiling {self.uri}")
traceback.print_exc()
def compute(self, songTime: float, inputs: EffectSettings) -> DeviceSettings:
"""
calls our function using inputs (publishedAttr attrs, e.g. :strength)
and effect-level settings including a special attr called :deviceSettings
with DeviceSettings as its value
"""
if self.config is None:
return DeviceSettings(self.graph, [])
c = self.config
kw = {}
for arg in c.funcArgs:
if arg.annotation == DeviceSettings:
v = c.devSettings
elif arg.name == 'songTime':
v = songTime
else:
eaForName = EffectAttr(L9[arg.name])
v = self._getEffectAttrValue(eaForName, inputs)
kw[arg.name] = v
log.debug('calling %s with %s', c.func, kw)
return c.func(**kw)
def _getEffectAttrValue(self, attr: EffectAttr, inputs: EffectSettings) -> VTUnion:
c = self.config
if c is None:
raise
try:
return inputs.getValue(self.uri, attr, defaultToZero=False)
except KeyError:
pass
try:
return c.esettings.getValue(self.uri, attr, defaultToZero=False)
except KeyError:
pass
return self.lib.getDefaultValue(c.effectFunction, attr)
|