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

def nsin(x):

    return (math.sin(x * (2 * math.pi)) + 1) / 2

def ncos(x):

from rdflib import Literal

def scale(value, strength):

    if isinstance(value, Literal):

        value = value.toPython()

    if isinstance(value, Decimal):

        value = float(value)

    if isinstance(value, str):

        if value[0] == '#':

            if strength == '#ffffff':

                return value

            r, g, b = hex_to_rgb(value)

            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.effect import effecteval

from light9.effect.sequencer import Note

from light9.effect.settings import DeviceSettings

from light9.effect.simple_outputs import SimpleOutputs

from light9.metrics import metrics

from light9.namespaces import L9, RDF

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.graph = graph

        self.sendToCollector = sendToCollector

        # Notes without times- always on

        self.notes: List[Note] = []

    #     await self.graph.addAsyncHandler(self.update)

    #     log.info('startupdating task done')

    def onCodeChange(self):

        log.debug('seq.onCodeChange')

        self.graph.addHandler(self.compileGraph)

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

    # @metrics('update_call_fader').time()

    # async def update(self):

    #     log.info(f'update {len(self.notes)=}')

    #     devSettings = self.computeOutput()

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

    if isinstance(ret, Decimal):

        return float(ret)

    return ret

def prettyFormat(x: Union[float, str]):

    if isinstance(x, float):

        return round(x, 4)

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

        if self.timed:

            originTime = typedValue(float, self.graph, self.uri, L9['originTime'])

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

            out[ea] = pyType(settingValues[L9['value']])

        return BareEffectSettings(s=out)

    def getCurvePoints(self, curve: Curve, attr, originTime: float) -> List[Tuple[float, float]]:

        points = []

        po = list(self.graph.predicate_objects(curve))

            return self.points[0][1]

        if self.points[i][0] > t:

            return self.points[i][1]

        if i >= len(self.points) - 1:

            return self.points[i][1]

        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] = {

            'note': str(self.uri),

            'effectClass': str(self.effectClass),

        }

        s = self.evalCurve(t) if strength is None else strength

        out = self.baseEffectSettings.withStrength(s)

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

'''

FADER_GRAPH = PREFIXES + '''

        :value 0.6 .

'''

class TestUntimedFaderNote:

    def test_returnsEffectSettings(self):

        g = MockSyncedGraph(FADER_GRAPH)

        n = Note(g, NoteUri(L9['fade1']), timed=False)

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

    def test_scalesStrengthWithCurve(self):

        pass

    lastSend = 0

    while True:

        await q.get()

        now = time.time()

        if now > lastSend + .2:

            lastSend = now

            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)

    async def send(settings: DeviceSettings):

        await sendToCollector('effectSequencer', session, settings)

    # seq = Sequencer(graph, send)  # per-song timed notes

    faders = FaderEval(graph, send)  # bin/fade's untimed notes

from light9.run_local import log

def test_import():

    async def go():

        # this sets up some watcher tasks

        from light9.effect.sequencer.service import app

        print(app)

    asyncio.run(go(), debug=True)

"""

Data structure and convertors for a table of (device,attr,value)

rows. These might be effect attrs ('strength'), device attrs ('rx'),

or output attrs (dmx channel).

BareSettings means (attr,value), no device.

"""

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] != '#':

        raise ValueError(h)

    return [int(h[i:i + 2], 16) for i in (1, 3, 5)]

def parseHexNorm(h):

    return [x / 255 for x in parseHex(h)]

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

        s1 = DeviceSettings(self.graph, [

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

            (L9['light1'], L9['attr2'], 0.3),

        ])

        s2 = DeviceSettings(self.graph, [

            (L9['light1'], L9['attr2'], 0.3),

            (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'], L9['setting'], L9['foo_set0'], ctx),

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

                (L9['foo_set0'], L9['deviceAttr'], L9['brightness'], ctx),

                (L9['foo_set0'], L9['value'], Literal(0.1), ctx),

                (L9['foo'], L9['setting'], L9['foo_set1'], ctx),

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

            (DEV['aura1'], L9['rx'], 0),

            (DEV['aura2'], L9['rx'], 0.1),

        ])

        # aura1 is all defaults (zeros), so it doesn't get listed

        self.assertCountEqual([DEV['aura2']], s.devices())

            (DEV['aura1'], L9['ry'], 0.3),

        ])

        self.assertEqual(0.36055512754639896, s1.distanceTo(s2))

L1 = L9['light1']

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)])),

                (.3, DeviceSettings(self.graph, [(L1, ZOOM, 1.0)])),

            ]))

    def testMixColors(self):

        self.assertEqual(

            DeviceSettings(self.graph, [(L1, ZOOM, HexColor('#503000'))]),

from typing import Any, Dict, List, Tuple

from rdflib import URIRef

from light9.effect.scale import scale

from light9.namespaces import L9, RDF

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:

                    d = settingValues.get(L9['device'], None)

                    a = settingValues.get(L9['deviceAttr'], None)

                    v = settingValues.get(L9['value'], None)

                    sv = settingValues.get(L9['scaledValue'], None)

                    if not (bool(v) ^ bool(sv)):

                        raise NotImplementedError('no value for %s' % setting)

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)

VT = TypeVar('VT', float, int, str)  # remove

HexColor = NewType('HexColor', str)