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

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

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

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

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:

import logging

from rdflib import Literal, URIRef

from webcolors import hex_to_rgb, rgb_to_hex

log = logging.getLogger('device')

class Device:

    pass

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

    """

        return sum(floatVals) / len(floatVals)

    return max(values)

def toOutputAttrs(

        deviceType: DeviceClass,

                                ]) -> Dict[OutputAttr, OutputValue]:

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

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

    """

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

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

        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)

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

    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)

    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

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

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

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