to send the new outputs.

        client is a string naming the type of client.

        (client, clientSession) is a unique client instance.

        clientSession is deprecated.

        Each client session's last settings will be forgotten

        after clientTimeoutSec.

        """

        # todo: cleanup session code if we really don't want to be able to run multiple sessions of one client

        clientSession = ClientSessionType("no_longer_used")

        now = UnixTime(time.time())

        self._warnOnLateRequests(client, now, sendTime)

        self._forgetStaleClients(now)

        self.lastRequest[(client, clientSession)] = (now, self._resolvedSettingsDict(settings))

        deviceAttrs = self._merge(iter(self.lastRequest.values()))

        outputAttrsByDevice = self._convertToOutputAttrsPerDevice(deviceAttrs)

        pendingOut = self._flattenDmxOutput(outputAttrsByDevice)

        dt1 = time.time() - now

        self._updateOutputs(pendingOut)

        dt2 = time.time() - dt1

        if dt1 > .030 or dt2 > .030:

            log.warning("slow setAttrs: prepare %.1fms -> updateOutputs %.1fms" % (dt1 * 1000, dt2 * 1000))

    def _warnOnLateRequests(self, client, now, sendTime):

        requestLag = now - sendTime

        if requestLag > .1 and now > self._initTime + 10 and getattr(self, '_lastWarnTime', 0) < now - 3:

            self._lastWarnTime = now

            log.warning('collector.setAttrs from %s is running %.1fms after the request was made', client, requestLag * 1000)

    def _forgetStaleClients(self, now):

        staleClientSessions = []

        for clientSession, (reqTime, _) in self.lastRequest.items():

            if reqTime < now - self.clientTimeoutSec:

                staleClientSessions.append(clientSession)

        for clientSession in staleClientSessions:

            log.info('forgetting stale client %r', clientSession)

            del self.lastRequest[clientSession]

    # todo: move to settings.py

        out: Dict[Tuple[DeviceUri, DeviceAttr], VTUnion] = {}

            if (devUri, devAttr) in out:

                existingVal = out[(devUri, devAttr)]

                out[(devUri, devAttr)] = resolve(self._deviceType[devUri], devAttr, [existingVal, val])

            else:

                out[(devUri, devAttr)] = val

        return out

    def _merge(self, lastRequests):

        deviceAttrs: Dict[DeviceUri, Dict[DeviceAttr, VTUnion]] = {}  # device: {deviceAttr: value}

        for _, lastSettings in lastRequests:

            for (device, deviceAttr), value in lastSettings.items():

                if (device, deviceAttr) in self.remapOut:

                    start, end = self.remapOut[(device, deviceAttr)]

                    value = start + float(value) * (end - start)

                attrs = deviceAttrs.setdefault(device, {})

                if deviceAttr in attrs:

                    value = resolve(device, deviceAttr, [attrs[deviceAttr], value])

                attrs[deviceAttr] = value

                # list should come from the graph. these are attrs

                # that should default to holding the last position,

                # not going to 0.

                if deviceAttr in [L9['rx'], L9['ry'], L9['zoom'], L9['focus']]:

                    self.stickyAttrs[(device, deviceAttr)] = cast(float, value)

        # e.g. don't let an unspecified rotation go to 0

        for (d, da), v in self.stickyAttrs.items():

            daDict = deviceAttrs.setdefault(d, {})

            if da not in daDict:

                daDict[da] = v

        return deviceAttrs

    def _convertToOutputAttrsPerDevice(self, deviceAttrs):

        ret: Dict[DeviceUri, Dict[OutputAttr, OutputValue]] = {}

        for d, devType in self._deviceType.items():

            try:

                ret[d] = toOutputAttrs(devType, deviceAttrs.get(d, {}))

                self.listeners.outputAttrsSet(d, ret[d], self._outputMap)

            except Exception as e:

                log.error('failing toOutputAttrs on %s: %r', d, e)

        return ret

    def _flattenDmxOutput(self, outputAttrs: Dict[DeviceUri, Dict[OutputAttr, OutputValue]]) -> Dict[OutputUri, bytearray]:

        pendingOut = cast(Dict[OutputUri, bytearray], {})

        for outUri in self._outputByUri.keys():

            pendingOut[outUri] = bytearray(512)

from typing import List

from light9 import networking

from light9.collector.collector import Collector

from light9.collector.output import ArtnetDmx, DummyOutput, Output, Udmx  # noqa

from light9.collector.weblisteners import UiListener, WebListeners

from light9.namespaces import L9

from light9.run_local import log

from light9.zmqtransport import parseJsonMessage

from prometheus_client import Summary

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from starlette.applications import Starlette

from starlette.endpoints import WebSocketEndpoint

from starlette.responses import Response

from starlette.routing import Route, WebSocketRoute

from starlette.types import Receive, Scope, Send

from starlette.websockets import WebSocket

from starlette_exporter import PrometheusMiddleware, handle_metrics

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

class Updates(WebSocketEndpoint, UiListener):

    def __init__(self, listeners, scope: Scope, receive: Receive, send: Send) -> None:

        super().__init__(scope, receive, send)

        self.listeners = listeners

    async def on_connect(self, websocket: WebSocket):

        await websocket.accept()

        log.info('socket connect %s', self.scope['client'])

        self.websocket = websocket

        self.listeners.addClient(self)

    async def sendMessage(self, msgText):

        await self.websocket.send_text(msgText)

    # async def on_receive(self, websocket, data):

    #     json.loads(data)

    async def on_disconnect(self, websocket: WebSocket, close_code: int):

        self.listeners.delClient(self)

    pass

async def PutAttrs(collector: Collector, request):

    with STAT_SETATTR.time():

        collector.setAttrs(client, clientSession, settings, sendTime)

        return Response('', status_code=202)

def main():

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

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

    logging.getLogger('output.allDmx').setLevel(logging.WARNING)

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

    try:

        # todo: drive outputs with config files

        rate = 30

        outputs: List[Output] = [

            # ArtnetDmx(L9['output/dmxA/'],

            #           host='127.0.0.1',

            #           port=6445,

            #           rate=rate),

            #sudo chmod a+rw /dev/bus/usb/003/021

            Udmx(L9['output/dmxA/'], bus=3, address=21, lastDmxChannel=100),

        ]

    except Exception:

        log.error("setting up outputs:")

        traceback.print_exc()

        raise

    listeners = WebListeners()

    c = Collector(graph, outputs, listeners)

    app = Starlette(

        debug=True,

        routes=[

            # Route('/recentRequests', lambda req: get_recentRequests(req, db)),

            WebSocketRoute('/updates', endpoint=functools.partial(Updates, listeners)),

            Route('/attrs', functools.partial(PutAttrs, c), methods=['PUT']),

        ],

    )

    app.add_middleware(PrometheusMiddleware)

    app.add_route("/metrics", handle_metrics)

    # loadtest = os.environ.get('LOADTEST', False)  # call myself with some synthetic load then exit

    # if loadtest:

    #     # in a subprocess since we don't want this client to be

    #     # cooperating with the main event loop and only sending

    #     # requests when there's free time

    #     def afterWarmup():

    #         log.info('running collector_loadtest')

import logging

import math

import random

from colorsys import hsv_to_rgb

from dataclasses import dataclass

from typing import Dict, Tuple

from noise import pnoise1

from PIL import Image

from rdfdb.syncedgraph.syncedgraph import SyncedGraph

from rdflib import Literal, Namespace

from webcolors import hex_to_rgb, rgb_to_hex

from light9.effect.scale import scale

from light9.effect.simple_outputs import SimpleOutputs

from light9.namespaces import DEV, L9

from light9.newtypes import (DeviceAttr, DeviceUri, EffectAttr, EffectClass, VTUnion)

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(gnorm * 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):

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

def nsquare(t, on=.5):

    return (t % 1.0) < on

def lerp(a, b, t):

    return a + (b - a) * t

def noise(t):

    return pnoise1(t % 1000.0, 2)

def clamp(lo, hi, x):

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

def clamp255(x):

    return min(255, max(0, x))

def _8bit(f):

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

        raise TypeError(repr(f))

    return clamp255(int(f * 255))

@dataclass

class EffectEval:

    """

    runs one effect's code to turn effect attr settings into output

    device settings. No state; suitable for reload().

    """

    graph: SyncedGraph

    effect: EffectClass

    simpleOutputs: SimpleOutputs

        """

        From effect attr settings, like strength=0.75, to output device

        settings like light1/bright=0.72;light2/bright=0.78. This runs

        the effect code.

        """

        # both callers need to apply note overrides

        strength = float(effectSettings.s[EffectAttr(L9['strength'])])

        if strength <= 0:

        report = {}

        out.update(self.simpleOutputs.values(self.effect, strength, effectSettings.s.get(EffectAttr(L9['colorScale']), None)))

        if self.effect.startswith(L9['effect/']):

            tail = 'effect_' + self.effect[len(L9['effect/']):]

            try:

                func = globals()[tail]

            except KeyError:

                report['error'] = 'effect code not found for %s' % self.effect

            else:

                out.update(func(effectSettings, strength, songTime, noteTime))

        outList = [(d, a, v) for (d, a), v in out.items()]

def effect_Curtain(effectSettings, strength, songTime, noteTime):

    return {(L9['device/lowPattern%s' % n], L9['color']): literalColor(strength, strength, strength) for n in range(301, 308 + 1)}

def effect_animRainbow(effectSettings, strength, songTime, noteTime):

    out = {}

    tint = effectSettings.get(L9['tint'], '#ffffff')

    tintStrength = float(effectSettings.get(L9['tintStrength'], 0))

    tr, tg, tb = hex_to_rgb(tint)

    for n in range(1, 5 + 1):

        scl = strength * nsin(songTime + n * .3)**3

        col = literalColor(scl * lerp(nsin(songTime + n * .2), tr / 255, tintStrength), scl * lerp(nsin(songTime + n * .2 + .3), tg / 255, tintStrength),

                           scl * lerp(nsin(songTime + n * .3 + .6), tb / 255, tintStrength))

        dev = L9['device/aura%s' % n]

        out.update({

            (dev, L9['color']): col,

            (dev, L9['zoom']): .9,

        })

        ang = songTime * 4

        out.update({

            (dev, L9['rx']): lerp(.27, .7, (n - 1) / 4) + .2 * math.sin(ang + n),

            (dev, L9['ry']): lerp(.46, .52, (n - 1) / 4) + .5 * math.cos(ang + n),

        })

    return out

def effect_auraSparkles(effectSettings, strength, songTime, noteTime):

    out = {}

    tint = effectSettings.get(L9['tint'], '#ffffff')

    print(effectSettings)

    tr, tg, tb = hex_to_rgb(tint)

    for n in range(1, 5 + 1):

        scl = strength * ((int(songTime * 10) % n) < 1)

        col = literalColorHsv((songTime + (n / 5)) % 1, 1, scl)

        dev = L9['device/aura%s' % n]

        out.update({

            (dev, L9['color']): col,

            (dev, L9['zoom']): .95,

        })

        ang = songTime * 4

        out.update({

            (dev, L9['rx']): lerp(.27, .8, (n - 1) / 4) + .2 * math.sin(ang + n),

            (dev, L9['ry']): lerp(.46, .52, (n - 1) / 4) + .4 * math.cos(ang + n),

        })