# This file is @generated by PDM.

# It is not intended for manual editing.

[metadata]

groups = ["default", "dev"]

strategy = ["cross_platform", "inherit_metadata"]

lock_version = "4.4.1"

[[package]]

name = "aiohttp"

version = "3.9.5"

requires_python = ">=3.8"

summary = "Async http client/server framework (asyncio)"

groups = ["default", "dev"]

dependencies = [

    "aiosignal>=1.1.2",

    "attrs>=17.3.0",

    "frozenlist>=1.1.1",

    "multidict<7.0,>=4.5",

[[package]]

name = "zipp"

version = "3.18.1"

requires_python = ">=3.8"

summary = "Backport of pathlib-compatible object wrapper for zip files"

groups = ["dev"]

files = [

    {file = "zipp-3.18.1-py3-none-any.whl", hash = "sha256:206f5a15f2af3dbaee80769fb7dc6f249695e940acca08dfb2a4769fe61e538b"},

    {file = "zipp-3.18.1.tar.gz", hash = "sha256:2884ed22e7d8961de1c9a05142eb69a247f120291bc0206a00a7642f09b5b715"},

]

[[package]]

name = "zope-interface"

version = "6.3"

requires_python = ">=3.7"

summary = "Interfaces for Python"

groups = ["default"]

dependencies = [

    "setuptools",

]

files = [

    {file = "zope.interface-6.3-cp311-cp311-macosx_10_9_x86_64.whl", hash = "sha256:600101f43a7582d5b9504a7c629a1185a849ce65e60fca0f6968dfc4b76b6d39"},

    {file = "zope.interface-6.3-cp311-cp311-macosx_11_0_arm64.whl", hash = "sha256:4d6b229f5e1a6375f206455cc0a63a8e502ed190fe7eb15e94a312dc69d40299"},

    {file = "zope.interface-6.3-cp311-cp311-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:10cde8dc6b2fd6a1d0b5ca4be820063e46ddba417ab82bcf55afe2227337b130"},

    "sse-starlette>=0.10.3",

    "starlette-exporter>=0.12.0",

    "starlette>=0.27.0",

    "statprof>=0.1.2",

    "toposort>=1.10",

    "udmx-pyusb>=2.0.0",

    "uvicorn[standard]>=0.17.6",

    "watchdog>=2.1.7",

    "webcolors>=1.11.1",

    "scipy>=1.9.3",

    "braillegraph>=0.6",

    "tenacity>=8.2.2",

    "mido>=1.2.10",

    "alsa-midi>=1.0.1",

    "treq>=22.2.0",

    "light9 @ file:///${PROJECT_ROOT}/",

    "python-debouncer>=0.1.4",

    "pytest>=8.2.0",

    "avro>=1.11.3",

    "fastavro>=1.9.4",

    "yappi>=1.6.0",

]

requires-python = ">=3.11"

import logging

import time

from typing import Dict, List, Set, Tuple, cast

from prometheus_client import Summary

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from light9.collector.device import resolve, toOutputAttrs

from light9.collector.output import Output as OutputInstance

from light9.collector.weblisteners import WebListeners

from light9.effect.settings import DeviceSettings

from light9.namespaces import L9, RDF

log = logging.getLogger('collector')

STAT_SETATTR = Summary('set_attr', 'setAttr calls')

def makeDmxMessageIndex(base: DmxIndex, offset: DmxIndex) -> DmxMessageIndex:

    return DmxMessageIndex(base + offset - 1)

def _outputMap(graph: SyncedGraph, outputs: Set[OutputUri]) -> Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]]:

    """From rdf config graph, compute a map of

       (device, outputattr) : (output, index)

    that explains which output index to set for any device update.

    """

    ret = cast(Dict[Tuple[DeviceUri, OutputAttr], Tuple[OutputUri, DmxMessageIndex]], {})

    for dc in graph.subjects(RDF.type, L9['DeviceClass']):

import json

import logging

import time

import zmq.asyncio

from prometheus_client import Summary

log = logging.getLogger('coll_client')

ZMQ_SEND = Summary('zmq_send', 'calls')

def toCollectorJson(client, session, settings: DeviceSettings) -> str:

    assert isinstance(settings, DeviceSettings)

    return json.dumps({

        'settings': settings.asList(),

        'client': client,

        'clientSession': session,

        'sendTime': time.time(),

import logging

from rdflib import Literal, URIRef

from webcolors import hex_to_rgb, rgb_to_hex

log = logging.getLogger('device')

class Device:

    pass

class ChauvetColorStrip(Device):

    """

     device attrs:

       color

def _8bit(f):

    if not isinstance(f, (int, float)):

        raise TypeError(repr(f))

    return clamp255(int(f * 255))

def _maxColor(values: List[HexColor]) -> HexColor:

    rgbs = [hex_to_rgb(v) for v in values]

    maxes = [max(component) for component in zip(*rgbs)]

    return cast(HexColor, rgb_to_hex(tuple(maxes)))

    """

    return one value to use for this attr, given a set of them that

    have come in simultaneously. len(values) >= 1.

    bug: some callers are passing a device instance for 1st arg

    """

    if len(values) == 1:

        return values[0]

    if deviceAttr == DeviceAttr(L9['color']):

        return _maxColor(cast(List[HexColor], values))

    # incomplete. how-to-resolve should be on the DeviceAttr defs in the graph.

    if deviceAttr in map(DeviceAttr, [L9['rx'], L9['ry'], L9['zoom'], L9['focus'], L9['iris']]):

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

                floatVals.append(float(v))

            else:

                raise TypeError(repr(v))

        # averaging with zeros? not so good

        return sum(floatVals) / len(floatVals)

    return max(values)

def toOutputAttrs(

        deviceType: DeviceClass,

    return dict((OutputAttr(u), OutputValue(v)) for u, v in untype_toOutputAttrs(deviceType, deviceAttrSettings).items())

def untype_toOutputAttrs(deviceType, deviceAttrSettings) -> Dict[URIRef, int]:

    """

    Given device attr settings like {L9['color']: Literal('#ff0000')},

    return a similar dict where the keys are output attrs (like

    L9['red']) and the values are suitable for Collector.setAttr

    :outputAttrRange happens before we get here.

    """

import asyncio

import io

import logging

import time

from typing import Any, Awaitable, Dict, List, Protocol, Tuple

import fastavro

from fastavro.schema import load_schema

from light9.newtypes import (DeviceUri, DmxIndex, DmxMessageIndex, OutputAttr, OutputUri, OutputValue)

import starlette.websockets

import websockets

log = logging.getLogger('weblisteners')

def shortenOutput(out: OutputUri) -> str:

    return str(out).rstrip('/').rsplit('/', 1)[-1]

class UiListener(Protocol):

        attrRows = []

        for attr, val in attrs.items():

            outputUri, bufIndex = outputMap[(dev, attr)]

            dmxIndex = DmxIndex(bufIndex + 1)

            attrRows.append({'attr': attr.rsplit('/')[-1], 'val': val, 'chan': (shortenOutput(outputUri), dmxIndex)})

        attrRows.sort(key=lambda r: r['chan'])

        for row in attrRows:

            row['chan'] = '%s %s' % (row['chan'][0], row['chan'][1])

        out = io.BytesIO()

        fastavro.schemaless_writer(out, self.CollectorUpdateSchema, {'OutputAttrsSet': {'dev': dev, 'attrs': attrRows}})

        msg = out.getvalue()

        return msg

from twisted.internet.defer import inlineCallbacks, returnValue

from light9 import networking

from light9.curvecalc.curve import CurveResource

from light9.namespaces import L9, RDF, RDFS

def clamp(x, lo, hi):

    return max(lo, min(hi, x))

@inlineCallbacks

def getMusicStatus():

    resp = yield treq.get(networking.musicPlayer.path('time'), timeout=.5)

    body = yield resp.json_content()

    returnValue(body)

        quads.extend([

            (song, L9['curve'], curve, ctx),

            (effect, RDFS.label, droppedLabel, ctx),

            (effect, L9['code'], Literal('env = %s' % curve.n3()), ctx),

        ])

        if L9['EffectClass'] in droppedTypes:

            quads.extend([

                (effect, RDF.type, dropped, ctx),

            ] + [(effect, L9['code'], c, ctx) for c in droppedCodes])

        elif L9['Submaster'] in droppedTypes:

            quads.extend([

            ])

        else:

        _maybeAddMusicLine(quads, effect, song, ctx)

    print("adding")

    for qq in quads:

        print(qq)

    returnValue(Patch(addQuads=quads))

@inlineCallbacks

def songNotePatch(graph, dropped, song, event, ctx, note=None):

    """

    quads = []

    fade = 2 if event == 'default' else 0.1

    if note:

        musicStatus = yield getMusicStatus()

        songTime = musicStatus['t']

        _finishCurve(graph, note, quads, ctx, songTime)

    else:

        if L9['Effect'] in droppedTypes:

            musicStatus = yield getMusicStatus()

            songTime = musicStatus['t']

        else:

            raise NotImplementedError

    returnValue((note, Patch(addQuads=quads)))

def _point(ctx, uri, t, v):

def _finishCurve(graph, note, quads, ctx, songTime):

    with graph.currentState() as g:

        origin = g.value(note, L9['originTime']).toPython()

        curve = g.value(note, L9['curve'])

    pt2 = graph.sequentialUri(curve + 'p')

    pt3 = graph.sequentialUri(curve + 'p')

                 _point(ctx, pt3, songTime - origin + .5, 0))

def _makeNote(graph, song, note, quads, ctx, dropped, songTime, event, fade):

    note = graph.sequentialUri(song + '/n')

    curve = graph.sequentialUri(note + 'c')

    quads.extend([

        (song, L9['note'], note, ctx),

        (note, RDF.type, L9['Note'], ctx),

        (note, L9['curve'], curve, ctx),

        (note, L9['effectClass'], dropped, ctx),

        (note, L9['originTime'], Literal(songTime), ctx),

    curve.insert_pt((t2, 1))

    curve.insert_pt((t2 + fade, 0))

def _maybeAddMusicLine(quads, effect, song, ctx):

    """

    add a line getting the current music into 'music' if any code might

    be mentioning that var

    """

    for spoc in quads:

        if spoc[1] == L9['code'] and 'music' in spoc[2]:

            break

import logging

import random

from PIL import Image

from webcolors import rgb_to_hex

from light9.effect.scale import scale

from light9.effect.settings import DeviceSettings

from light9.namespaces import L9

random.seed(0)

log = logging.getLogger('effectfunc')

    if not (0 <= y < img.height):

        return (0, 0, 0)

    if 0 <= x < img.width:

        return img.getpixel((x, y))

    elif not repeat:

        return (0, 0, 0)

    else:

        return img.getpixel((x % img.width, y))

def effect_scale(strength: float, devs: DeviceSettings) -> DeviceSettings:

    out = []

    return effect_scale(scl, devs)

def effect_image(

    songTime: float,  #

    strength: float,

    period: float,

    image: Image.Image,

    devs: DeviceSettings,

) -> DeviceSettings:

    x = int((songTime / period) * image.width)

    out = []

        if da != L9['color']:

            continue

        color8 = sample8(image, x, y, repeat=True)

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

        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

                    log.debug("%s asked for devs but we have none in config", self.uri)

                    return DeviceSettings(self.graph, [])

            elif arg.name == 'songTime':

                v = songTime

            else:

                eaForName = EffectAttr(L9[arg.name])

                v = self._getEffectAttrValue(eaForName, inputs)

            kw[arg.name] = v

        if False and log.isEnabledFor(logging.DEBUG):

            log.debug('calling %s with %s', c.func, kw)

from decimal import Decimal

from webcolors import hex_to_rgb, rgb_to_hex

from light9.newtypes import VTUnion

def scale(value: VTUnion, strength: float):

    if isinstance(value, Decimal):

        raise TypeError()

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

import logging

import time

from dataclasses import dataclass

from typing import List, Optional, cast

from prometheus_client import Summary

from rdfdb import SyncedGraph

from rdflib import URIRef

from rdflib.term import Node

from light9.effect.effect_function_library import EffectFunctionLibrary

from light9.effect.effecteval2 import EffectEval2

from light9.effect.settings import DeviceSettings, EffectSettings

from light9.namespaces import L9, RDF

import asyncio

import json

import logging

import time

from louie import dispatcher

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from sse_starlette.sse import EventSourceResponse

from starlette.applications import Starlette

from starlette.routing import Route

from starlette_exporter import PrometheusMiddleware, handle_metrics

from light9.background_loop import loop_forever

from light9.collector.collector_client_asyncio import sendToCollector

from light9.effect.effect_function_library import EffectFunctionLibrary

from light9.effect.sequencer.eval_faders import FaderEval

from light9.effect.sequencer.sequencer import Sequencer, StateUpdate

from light9.run_local import log

RATE = 20

async def changes():

    state = {}

    q = asyncio.Queue()

        if now > lastSend + .2:

            lastSend = now

            yield json.dumps(state)

async def send_page_updates(request):

    return EventSourceResponse(changes())

def main():

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

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

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

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

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

    logging.getLogger('seq.fader').setLevel(logging.INFO)

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

    lib = EffectFunctionLibrary(graph)

    faders = FaderEval(graph, lib)  # bin/fade's untimed effects

    #@metrics('computeAndSend').time() # needs rework with async

    async def update(first_run):

        ds = faders.computeOutput()

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

import decimal

import logging

from dataclasses import dataclass

from typing import Any, Dict, Iterable, List, Sequence, Set, Tuple, cast

import numpy

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from rdflib import Literal, URIRef

from light9.collector.device import resolve

from light9.localsyncedgraph import LocalSyncedGraph

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