light9 Changeset - af38643cffd4

Changeset - af38643cffd4

Parent rev.

Child rev.

[Not reviewed]

default

0 1 0

drewp@bigasterisk.com - 20 months ago 2023-06-03 20:10:24
drewp@bigasterisk.com

jam in fader page control, fix _lastSet feedback

1 file changed with 91 insertions and 35 deletions:

light9/midifade/midifade.py

0 comments (0 inline, 0 general)

light9/midifade/midifade.py

➞

Show inline comments

 #!bin/python
 """
 Read midi events, write fade levels to graph
 """
 import asyncio
 import logging
 import traceback
 from typing import Dict, List
 from typing import Dict, List, cast
 from light9.effect.edit import clamp
 import mido
 from rdfdb.syncedgraph.syncedgraph import SyncedGraph
 from rdflib import Literal, URIRef
 from rdflib import RDF, ConjunctiveGraph, Literal, URIRef
 from rdfdb.syncedgraph.readonly_graph import ReadOnlyConjunctiveGraph
 from light9 import networking
 from light9.namespaces import L9
 from light9.newtypes import decimalLiteral
 from light9.run_local import log
 from light9.showconfig import showUri
 mido.set_backend('alsa_midi.mido_backend')
 MAX_SEND_RATE = 20
 _lastSet = {}  #Fader:value
 _lastSet = {}  #midictlchannel:value7bit
 currentFaders = {}  # midi control channel num : FaderUri
 ctx = URIRef(showUri() + '/fade')
 def compileCurrents(graph):
     currentFaders.clear()
     try:
         new = getChansToFaders(graph)
     except ValueError:
         return  # e.g. empty-graph startup
     currentFaders.update(new)
 def getGraphMappingNode(g: ReadOnlyConjunctiveGraph|SyncedGraph) -> URIRef:
     mapping = g.value(L9['midiControl'], L9['map'])
     if mapping is None:
         raise ValueError('no :midiControl :map ?mapping')
     midiDev = g.value(mapping, L9['midiDev'])
     ourDev = 'bcf2000'
     if midiDev != Literal(ourDev):
         raise NotImplementedError(f'need {mapping} to have :midiDev {ourDev!r}')
     return mapping
 def setFader(graph: SyncedGraph, ctx, fader: URIRef, strength: float):
 def getCurMappedPage(g: SyncedGraph):
     mapping = getGraphMappingNode(g)
     return g.value(mapping, L9['outputs'])
 def setCurMappedPage(g: SyncedGraph, mapping: URIRef, newPage:URIRef):
     g.patchObject(ctx, mapping, L9.outputs, newPage)
 def getChansToFaders(g: SyncedGraph) -> Dict[int, URIRef]:
     fadePage = getCurMappedPage(g)
     ret = []
     for f in g.objects(fadePage, L9.fader):
         columnLit = cast(Literal, g.value(f, L9['column']))
         col = int(columnLit.toPython())
         ret.append((col, f))
     ret.sort()
     ctl_channels = list(range(81, 88 + 1))
     out = {}
     for chan, (col, f) in zip(ctl_channels, ret):
         out[chan] = f
     return out
 def changePage(g: SyncedGraph, dp: int):
     """dp==-1, make the previous page active, etc. Writes to graph"""
     with g.currentState() as current:
         allPages = sorted(current.subjects(RDF.type, L9.FadePage), key=lambda fp: str(fp))
         mapping = getGraphMappingNode(current)
         curPage = current.value(mapping, L9.outputs)
     if curPage is None:
         curPage = allPages[0]
     idx = allPages.index(curPage)
     newIdx = clamp(idx + dp, 0, len(allPages)-1)
     print('change from ', idx, newIdx)
     newPage =allPages[newIdx]
     setCurMappedPage(g, mapping, newPage)
 def writeHwValueToGraph(graph: SyncedGraph, ctx, fader: URIRef, strength: float):
     log.info(f'setFader(fader={fader}, strength={strength:.03f}')
     valueLit = decimalLiteral(round(strength, 3))
     with graph.currentState() as g:
         fadeSet = g.value(fader, L9['setting'])
     if fadeSet is None:
         raise ValueError(f'fader {fader} has no :setting')
     graph.patchObject(ctx, fadeSet, L9['value'], valueLit)
     _lastSet[fader] = strength
 def onMessage(graph: SyncedGraph, ctx: URIRef, m: Dict):
     if m['type'] == 'active_sensing':
         return
     if m['type'] == 'control_change':
         if m['dev'] == 'bcf2000' and m['control'] == 91:
             changePage(graph, -1)
             return
         if m['dev'] == 'bcf2000' and m['control'] == 92:
             changePage(graph, 1)
             return
         try:
             fader = {
                 'quneo': {
 : L9['show/dance2023/fadePage1f0'],
 : L9['show/dance2023/fadePage1f0'],
 : L9['show/dance2023/fadePage1f0'],
                 },
                 'bcf2000': {
 : L9['show/dance2023/fader0'],
 : L9['show/dance2023/fader1'],
 : L9['show/dance2023/fader2'],
 : L9['show/dance2023/fader3'],
 : L9['show/dance2023/fader4'],
 : L9['show/dance2023/fader5'],
 : L9['show/dance2023/fader6'],
 : L9['show/dance2023/fader7'],
+                }
                 'bcf2000': currentFaders,
             }[m['dev']][m['control']]
         except KeyError:
             log.info(f'unknown control {m}')
             return
         try:
             setFader(graph, ctx, fader, m['value'] / 127)
             writeHwValueToGraph(graph, ctx, fader, m['value'] / 127)
             _lastSet[m['control']] = m['value']
         except ValueError as e:
             log.warning(f'{e!r} - ignoring')
     else:
         log.info(f'unhandled message {m}')
 def reduceToLatestValue(ms: List[Dict]) -> List[Dict]:
     merge = {}
     for m in ms:
         normal_key = tuple(sorted(dict((k, v) for k, v in m.items() if k != 'value')))
         merge[normal_key] = m
     return merge.values()
 class WriteBackFaders:
     def __init__(self, graph: SyncedGraph, bcf_out, getCurrentValue):
         self.graph = graph
         self.bcf_out = bcf_out
         self.getCurrentValue = getCurrentValue
     def update(self):
         try:
             self._update()
         except ValueError as e:
             log.warning(repr(e))
     def _update(self):
         g = self.graph
         mapping = g.value(L9['midiControl'], L9['map'])
         if mapping is None:
             raise ValueError('no :midiControl :map ?mapping')
         midiDev = g.value(mapping, L9['midiDev'])
         ourDev = 'bcf2000'
         if midiDev != Literal(ourDev):
             raise NotImplementedError(f'need {mapping} to have :midiDev {ourDev!r}')
         fadePage = g.value(mapping, L9['outputs'])
         nupdated=0
         faders = list(g.objects(fadePage, L9.fader))
         for f in faders:
             column = int(g.value(f, L9.column).toPython())
         m = getChansToFaders(g)
         for midi_ctl_addr, f in m.items():
             fset = g.value(f, L9.setting)
             # could split this to a separate handler per fader
             value = g.value(fset, L9.value).toPython()
-            hwcurrent = int(self.getCurrentValue(f) * 127)
+            hwcurrent = self.getCurrentValue(midi_ctl_addr)
             hwgoal = int(value * 127)
             if abs(hwcurrent - hwgoal) > 5:
                 midi_ctl_addr = column + 80
             print(f'{f} {hwcurrent=} {hwgoal=}')
             if abs(hwcurrent - hwgoal) > 2:
                 self.sendToBcf(midi_ctl_addr, hwgoal)
                 nupdated+=1
         log.info(f'updated {nupdated} of {len(faders)} connected faders')
         log.info(f'wrote to {nupdated} of {len(m)} mapped faders')
     def sendToBcf(self, control, value):
         _lastSet[control] = value
         msg = mido.Message('control_change', control=control, value=value)
         self.bcf_out.send(msg)
 async def main():
     logging.getLogger('autodepgraphapi').setLevel(logging.INFO)
     logging.getLogger('syncedgraph').setLevel(logging.INFO)
     logging.getLogger('graphedit').setLevel(logging.INFO)
     graph = SyncedGraph(networking.rdfdb.url, "midifade")
     ctx = URIRef(showUri() + '/fade')
     msgs = asyncio.Queue()
     loop = asyncio.get_event_loop()
     def onMessageMidoThread(dev, message):
         loop.call_soon_threadsafe(msgs.put_nowait, message.dict() | {'dev': dev})
     async def reader():
         while True:
             recents = [await msgs.get()]
             while not msgs.empty():
                 recents.append(msgs.get_nowait())
             try:
                 for msg in reduceToLatestValue(recents):
                     onMessage(graph, ctx, msg)
             except Exception as e:
                 traceback.print_exc()
                 log.warning("error in onMessage- continuing anyway")
             await asyncio.sleep(1 / MAX_SEND_RATE)
     asyncio.create_task(reader())
     openPorts = []
     for inputName in mido.get_input_names():
+    for inputName in mido.get_input_names():  # type: ignore
         if inputName.startswith('Keystation'):
             dev = "keystation"
         elif inputName.startswith('BCF2000'):
             dev = 'bcf2000'
         elif inputName.startswith('QUNEO'):
             dev = 'quneo'
         else:
             continue
         log.info(f'listening on input {inputName} {dev=}')
         openPorts.append(mido.open_input(  #
+        openPorts.append(mido.open_input(  # type: ignore
             inputName,  #
             callback=lambda message, dev=dev: onMessageMidoThread(dev, message)))
     bcf_out = mido.open_output('BCF2000:BCF2000 MIDI 1 28:0')
     graph.addHandler(lambda: compileCurrents(graph))
     for outputName in mido.get_output_names():  # type: ignore
         if outputName.startswith('BCF2000'):
             bcf_out = mido.open_output(outputName)  # type: ignore
     wb = WriteBackFaders(graph, bcf_out, getCurrentValue=lambda f: _lastSet.get(f, 0))
     graph.addHandler(wb.update)
             break
     while True:
         await asyncio.sleep(1)
 if __name__ == '__main__':
     asyncio.run(main())

0 comments (0 inline, 0 general)