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/')

def literalColor(rnorm, gnorm, bnorm):

    return Literal(rgb_to_hex((

        int(rnorm * 255),  #

        int(bnorm * 255))))

from rdflib import Literal

def scale(value, strength):

                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

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,

    def compileGraph(self) -> None:

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

        self.notes = []

            def compileFader() -> Note:

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

    @metrics('compile_fader').time()

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

                    timed=False)

    def computeOutput(self) -> DeviceSettings:

        notesSettings = []

        now = UnixTime(time.time())

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

            deviceSettings, report = ee.outputFromEffect(

                songTime=now, # probably wrong

                noteTime=now, # wrong

                )

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/> .

'''

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')

@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

    # 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.points.sort()

        else:

            self.points = []

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

        """

        out: Dict[EffectAttr, VTUnion] = {}

        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:

'''

FADER_GRAPH = PREFIXES + '''

        :value 0.6 .

'''

NOTE_GRAPH = PREFIXES + '''

        '''

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")

from light9.run_local import log

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')

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.assertFalse(s1 != s2)

    def testMissingFieldsEqZero(self):

    def testFalseIfZero(self):

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

    def testFromResource(self):

            ]))

        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, [

    def testAsList(self):

        sets = [

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

        ]

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

class TestFromBlend(unittest.TestCase):

    def setUp(self):

    def testSingle(self):

    def testScale(self):

    def testMixFloats(self):

        self.assertEqual(

class SimpleOutputs:

    """

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

    """

        self.graph.addHandler(self.updateEffectsFromGraph)

    def updateEffectsFromGraph(self):

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

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

            settings = []

import decimal

from rdflib import URIRef

from rdflib.term import Node

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

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: