gatherProcessStats()

class App(object):

    def __init__(self, graph, show):

gatherProcessStats()

class EffectEdit(PrettyErrorHandler, cyclone.web.RequestHandler):

    def get(self):

                index = DmxMessageIndex(_index)

                _, outArray = pendingOut[outputUri]

                if outArray[index] != 0:

                    raise ValueError(f"someone already wrote to index {index}")

                outArray[index] = value

from pyftdi import ftdi

#FTDI device info

#####################

# DMX USB controller

#####################

class OpenDmxUsb():

    def __init__(self):

        self.baud_rate = 250000

        self.data_bits = 8

    #Initialize the controller

    def _init_dmx(self):

        self.ftdi.set_baudrate(self.baud_rate)

        self.ftdi.set_flowctrl(self.flow_ctrl)

        self.ftdi.purge_rx_buffer()

        self.ftdi.purge_tx_buffer()

        self.ftdi.set_rts(self.rts_state)

    #Send DMX data

        assert self.ftdi.write_data(channelVals) == 513

        # Need to generate two bits for break

            channelVals[x] = 255

        dmxUsb.send_dmx(channelVals)

        scales.init(self, self.statPath)

        self._writeStats = scales.collection(

            scales.RecentFpsStat('fps'))

        self._currentBuffer = b''

    def reconnect(self):

        pass

    def shortId(self) -> str:

        """short string to distinguish outputs"""

        return self.uri.rstrip('/').rsplit('/')[-1]

        if msg != self._lastLoggedMsg:

            log.debug(msg)

            self._lastLoggedMsg = msg

    def _write(self, buf: bytes) -> None:

        """

        write buffer to output hardware (may be throttled if updates are

        log.error('unrecoverable- exiting')

        reactor.crash()

class DummyOutput(Output):

    def __init__(self, uri, **kw):

        d = threads.deferToThread(self._write, sendingBuffer)

        d.addCallbacks(done, err)

class FtdiDmx(BackgroundLoopOutput):

    def __init__(self, uri, lastDmxChannel, rate=22):

        super().__init__(uri)

        self.lastDmxChannel = lastDmxChannel

        from .dmx_controller_output import OpenDmxUsb

        self.dmx = OpenDmxUsb()

    def _write(self, buf):

        self._writeStats.fps.mark()

        with self._writeStats.call.time():

            if not buf:

                return

            # ok to truncate the last channels if they just went

            if logAllDmx.isEnabledFor(logging.DEBUG):

                # for testing fps, smooth fades, etc

            self.dmx.send_dmx(buf)

        super().__init__(uri, rate=rate)

        self._errStats = scales.collection(self.statPath + '/write',

        self.reconnect()

    def shortId(self) -> str:

        return super().shortId() + f'_bus={self.bus}'

    def reconnect(self):

        self._connected = 0

        from pyudmx import pyudmx

        with self._writeStats.call.time():

            try:

                if not buf:

                    return

                # ok to truncate the last channels if they just went

                    self._errStats.overflow += 1

                    return

                    self._errStats.ioError += 1

                    return

                    self._errStats.pipeError += 1

                    return

                msg = 'usb: sending %s bytes to %r; error %r' % (len(buf),

                                                                 self.uri, e)

                log.warn(msg)

            out[(dev, L9['color'])] = col

    return out

def sample(img, x, y, repeat=False):

    if 0 <= x < img.width:

        return img.getpixel((x, y))

    else:

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

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

    out = {}

    imgPath = f'cur/anim/{effectSettings[L9["image"]]}'

            'effectClass': self.effectEval.effect,

        }

        effectSettings: Dict[DeviceAttr, Union[float, str]] = dict(

        effectSettings[L9['strength']] = self.evalCurve(t)

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

        log.debug('seq.onCodeChange')

        self.graph.addHandler(self.compileGraph)

        #self.updateLoop()

    @compileStats.graph.time()

    def compileGraph(self) -> None:

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

        self.notes[song] = []

        for note in self.graph.objects(song, L9['note']):

            try:

            except Exception:

                log.warn(f"failed to build Note {note} - skipping")

                anyErrors = True

            if not self.lastLoopSucceeded:

                log.info('Sequencer.update is working')

                self.lastLoopSucceeded = True

            delay = max(0, 1 / self.fps - took)

            reactor.callLater(delay, self.updateLoop)

    @updateStats.updateFps.rate()

    @inlineCallbacks

    def update(self) -> Deferred:

        with updateStats.s0_getMusic.time():

            musicState = self.music.getLatest()

            if not musicState.get('song') or not isinstance(

        with updateStats.s1_eval.time():

            settings = []

            noteReports = []

            for note in songNotes:

                try:

        dispatcher.send('state', update={'songNotes': noteReports})

            sendSecs = yield self.sendToCollector(devSettings)

        # sendToCollector's own measurement.

        #if isinstance(sendSecs, float):

        #    updateStats.s3_send_client = sendSecs

class Updates(cyclone.sse.SSEHandler):

    def __init__(self, application, request, **kwargs) -> None:

class SubClient:

    graph: SyncedGraph

    session: URIRef

    def __init__(self):

        """assumed that your init saves self.graph"""

        pass  # we may later need init code for network setup