import logging

import math

import random

from colorsys import hsv_to_rgb

from noise import pnoise1

from PIL import Image

from webcolors import hex_to_rgb, rgb_to_hex

from light9.effect.scale import scale

from light9.namespaces import DEV, L9

SKY = Namespace('http://light9.bigasterisk.com/theater/skyline/device/')

random.seed(0)

log = logging.getLogger('effecteval')

log.info("reload effecteval")

def literalColor(rnorm, gnorm, bnorm):

    return Literal(rgb_to_hex((

        int(rnorm * 255),  #

        int(bnorm * 255))))

def literalColorHsv(h, s, v):

    return literalColor(*hsv_to_rgb(h, s, v))

from rdflib import Literal

def scale(value, strength):

    if isinstance(value, Literal):

        value = value.toPython()

            if isinstance(strength, Literal):

                strength = strength.toPython()

            if isinstance(strength, str):

                sr, sg, sb = [v / 255 for v in hex_to_rgb(strength)]

            else:

                sr = sg = sb = strength

    elif isinstance(value, (int, float)):

        return value * strength

    raise NotImplementedError("%r,%r" % (value, strength))

from light9.newtypes import EffectAttr, NoteUri, UnixTime

log = logging.getLogger('seq.fader')

class FaderEval:

    """peer to Sequencer, but this one takes the current :Fader settings -> sendToCollector

    The current faders become Notes in here, for more code reuse.

    """

    def __init__(self,

                 graph: SyncedGraph,

                 sendToCollector: Callable[[DeviceSettings], Coroutine[None ,None,None]],

                 ):

        #self.updateLoop()

    @metrics('compile_graph_fader').time()

    def compileGraph(self) -> None:

        """rebuild our data from the graph"""

        self.notes = []

            def compileFader() -> Note:

                return self.compileFader(cast(URIRef, fader))

            self.notes.append(compileFader())

        # if self.notes:

        #     asyncio.create_task(self.startUpdating())

    @metrics('compile_fader').time()

    def compileFader(self, fader: URIRef) -> Note:

                    timed=False)

    def computeOutput(self) -> DeviceSettings:

        notesSettings = []

        now = UnixTime(time.time())

        for note in self.notes:

            effectSettings, report = note.outputCurrent()

            if effectSettings.s[EffectAttr(L9['strength'])]==0:

                continue

            ee = effecteval.EffectEval(self.graph, note.effectClass, self.simpleOutputs)

            deviceSettings, report = ee.outputFromEffect(

                songTime=now, # probably wrong

                noteTime=now, # wrong

                )

            if deviceSettings:

                notesSettings.append(deviceSettings)

        return DeviceSettings.merge(self.graph, notesSettings)

from unittest import mock

from light9.effect.sequencer.eval_faders import FaderEval

from light9.effect.settings import DeviceSettings

from light9.mock_syncedgraph import MockSyncedGraph

from light9.namespaces import L9

PREFIXES = '''

@prefix : <http://light9.bigasterisk.com/> .

@prefix effect: <http://light9.bigasterisk.com/effect/> .

@prefix rdfs: <http://www.w3.org/2000/01/rdf-schema#> .

@prefix show: <http://light9.bigasterisk.com/show/dance2023/> .

@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .

@prefix dev: <http://light9.bigasterisk.com/theater/test/device/> .

@prefix dmxA: <http://light9.bigasterisk.com/output/dmxA/> .

'''

NOTE_GRAPH = PREFIXES + '''

                :dataType :scalar .

            :strength

                a :EffectAttr;

                rdfs:label "strength" .

                rdfs:label "SimpleDimmer";

                :deviceAttr :brightness;

                :attr [ :outputAttr :level; :dmxOffset 0 ] .

                a :SimpleDimmer;

                :dmxUniverse dmxA:;

                :dmxBase 178 .

                :setting effect:effect1_set1 .

                :scaledValue 0.5 .

                :value 0.6 .

        '''

class TestFaderEval:

    def test_faderValueScalesEffectSettings(self):

        g = MockSyncedGraph(NOTE_GRAPH)

        sender = mock.MagicMock()

        f = FaderEval(g, sender)

        devSettings = f.computeOutput()

        assert devSettings == DeviceSettings(g, [(L9['light1'], L9['brightness'], 0.3)])

import bisect

import logging

import time

from decimal import Decimal

from typing import Any, Dict, List, Optional, Tuple, Union, cast

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from light9.namespaces import L9

log = logging.getLogger('sequencer')

def pyType(n):

    ret = n.toPython()

    return x

@dataclass

class Note:

    """A Note outputs EffectAttr settings.

    Sample graph:

     :note1 a :Note; :curve :n1c1; :effectClass effect:allcolor;

    It can animate the EffectAttr settings over time, in two ways:

       the :strength EffectAttr over time

       effect can include animation. A `Fader` is an untimed note.

    around the creation of Note objects.

    """

    graph: SyncedGraph

    uri: NoteUri

    # simpleOutputs: SimpleOutputs

    timed: bool = True

        ec = self.graph.value(self.uri, L9['effectClass'])

        if ec is None:

            raise ValueError(f'note {self.uri} has no :effectClass')

        self.effectClass = EffectClass(ec)

        self.baseEffectSettings = self.getBaseEffectSettings()

            self.points: List[Tuple[float, float]] = []

            for curve in self.graph.objects(self.uri, L9['curve']):

                self.points.extend(self.getCurvePoints(cast(Curve, curve), L9['strength'], originTime))

            self.points.sort()

        else:

            self.points = []

        e.g. that you set in the note's body in the timeline editor

        """

        out: Dict[EffectAttr, VTUnion] = {}

        for s in self.graph.objects(self.uri, L9['setting']):

            settingValues = dict(self.graph.predicate_objects(s))

            ea = cast(EffectAttr, settingValues[L9['effectAttr']])

        p1, p2 = self.points[i], self.points[i + 1]

        frac = (t - p1[0]) / (p2[0] - p1[0])

        y = p1[1] + (p2[1] - p1[1]) * frac

        return y

    def outputCurrent(self):  # no graph

        return self._outputSettings(t=None, strength=self.value)

    def _outputSettings(

            self,

            t: float | None,

            strength: Optional[float] = None  #

        if t is None:

            if self.timed:

                raise TypeError()

            t = time.time()  # so live effects will move

        report: Dict[str, Any] = {

@prefix xsd: <http://www.w3.org/2001/XMLSchema#> .

@prefix dev: <http://light9.bigasterisk.com/theater/test/device/> .

@prefix dmxA: <http://light9.bigasterisk.com/output/dmxA/> .

'''

FADER_GRAPH = PREFIXES + '''

        :value 0.6 .

'''

class TestUntimedFaderNote:

        out, report = n.outputCurrent()

        assert report['effectSettings'] == {'http://light9.bigasterisk.com/strength': 0.6}

        assert out == BareEffectSettings(s={EffectAttr(L9['strength']): 0.6})

NOTE_GRAPH = PREFIXES + '''

        '''

class TestTimedNote:

    @pytest.mark.skip()

            yield json.dumps(state)

async def send_page_updates(request):

    return EventSourceResponse(changes())

###################################################################

    ds = faders.computeOutput()

    await sendToCollector('effectSequencer', session='0', settings=ds)

async def _forever(faders):

    while True:

        await _send_one(faders)

        await asyncio.sleep(0.1)

def send_updates_forever(faders):

    asyncio.create_task(_forever(faders))

####################################################################

def main():

    session = 'effectSequencer'

    graph = SyncedGraph(networking.rdfdb.url, "effectSequencer")

    logging.getLogger('autodepgraphapi').setLevel(logging.INFO)

    logging.getLogger('syncedgraph').setLevel(logging.INFO)

    logging.getLogger('sse_starlette.sse').setLevel(logging.INFO)

from light9.run_local import log

def test_import():

    async def go():

"""

from __future__ import annotations

import decimal

import logging

from dataclasses import dataclass

import numpy

from rdflib import Literal, URIRef

from light9.collector.device import resolve

from light9.namespaces import L9, RDF

log = logging.getLogger('settings')

def parseHex(h):

    if h[0] != '#':

from typing import cast

from rdflib import Literal

from light9.localsyncedgraph import LocalSyncedGraph

class TestDeviceSettings(unittest.TestCase):

    def setUp(self):

    def testToVectorZero(self):

        ds = DeviceSettings(self.graph, [])

        self.assertEqual([0] * 30, ds.toVector())

    def testEq(self):

            (L9['light1'], L9['attr1'], 0.5),

        ])

        self.assertTrue(s1 == s2)

        self.assertFalse(s1 != s2)

    def testMissingFieldsEqZero(self):

    def testFalseIfZero(self):

        self.assertFalse(DeviceSettings(self.graph, []))

    def testFromResource(self):

        ctx = L9['']

        self.graph.patch(

            Patch(addQuads=[

                (L9['foo_set1'], L9['device'], L9['light1'], ctx),

                (L9['foo_set1'], L9['deviceAttr'], L9['speed'], ctx),

                (L9['foo_set1'], L9['scaledValue'], Literal(0.2), ctx),

            ]))

        s = DeviceSettings.fromResource(self.graph, DeviceUri(L9['foo']))

    def testToVector(self):

        v = DeviceSettings(self.graph, [

            (DeviceUri(DEV['aura1']), DeviceAttr(L9['rx']), 0.5),

            (DeviceUri(DEV['aura1']), DeviceAttr(L9['color']), HexColor('#00ff00')),

        ]).toVector()

    def testFromVector(self):

        self.assertEqual(

            DeviceSettings(self.graph, [

                (DeviceUri(DEV['aura1']), DeviceAttr(L9['rx']), 0.5),

                (DeviceUri(DEV['aura1']), DeviceAttr(L9['color']), HexColor('#00ff00')),

            ]), s)

    def testAsList(self):

        sets = [

            (DeviceUri(L9['light1']), DeviceAttr(L9['attr1']), 0.5),

        ]

        self.assertCountEqual(sets, DeviceSettings(self.graph, sets).asList())

    def testDevices(self):

        s = DeviceSettings(self.graph, [

ZOOM = L9['zoom']

class TestFromBlend(unittest.TestCase):

    def setUp(self):

    def testSingle(self):

    def testScale(self):

    def testMixFloats(self):

        self.assertEqual(

            DeviceSettings(self.graph, [(L1, ZOOM, 0.4)]),

            DeviceSettings.fromBlend(self.graph, [

                (.2, DeviceSettings(self.graph, [(L1, ZOOM, 0.5)])),

log = logging.getLogger('simple')

class SimpleOutputs:

    """

    Call `values` to get (dev,attr):value settings.

    """

    def __init__(self, graph):

        self.graph = graph

        # effect : [(dev, attr, value, isScaled)]

        self.effectOutputs: Dict[URIRef, List[Tuple[URIRef, URIRef, Any, bool]]] = {}

        self.graph.addHandler(self.updateEffectsFromGraph)

    def updateEffectsFromGraph(self):

        for effect in self.graph.subjects(RDF.type, L9['Effect']):

            log.debug(f' {effect=}')

            settings = []

            for setting in self.graph.objects(effect, L9['setting']):

                settingValues = dict(self.graph.predicate_objects(setting))

                try:

import decimal

from rdflib import URIRef

from rdflib.term import Node

ClientType = NewType('ClientType', str)

ClientSessionType = NewType('ClientSessionType', str)

Curve = NewType('Curve', URIRef)

OutputUri = NewType('OutputUri', URIRef)  # e.g. dmxA

DeviceUri = NewType('DeviceUri', URIRef)  # e.g. :aura2

DeviceClass = NewType('DeviceClass', URIRef)  # e.g. :Aura

DmxIndex = NewType('DmxIndex', int)  # 1..512

DmxMessageIndex = NewType('DmxMessageIndex', int)  # 0..511

DeviceAttr = NewType('DeviceAttr', URIRef)  # e.g. :rx

EffectAttr = NewType('EffectAttr', URIRef)  # e.g. :chaseSpeed

NoteUri = NewType('NoteUri', URIRef)

OutputAttr = NewType('OutputAttr', URIRef)  # e.g. :xFine

OutputValue = NewType('OutputValue', int)  # byte in dmx message

Song = NewType('Song', URIRef)

UnixTime = NewType('UnixTime', float)

    if hasattr(t1, '__supertype__'):

        t1 = t1.__supertype__

    return issubclass(t1, t2)

def typedValue(objType: Type[_ObjType], graph, subj, pred) -> _ObjType:

    If objType is not an rdflib.Node, we toPython() the value."""

    obj = graph.value(subj, pred)

    if obj is None:

        raise ValueError()

    conv = obj if _isSubclass2(objType, Node) else obj.toPython()

    if objType is float and isinstance(conv, decimal.Decimal):