1268 lines
37 KiB
Python
1268 lines
37 KiB
Python
|
|
#!/usr/bin/python3
|
|
# -*- coding: utf-8 -*-
|
|
|
|
"""
|
|
Midi3 light version with LJ2 settings support
|
|
v0.7.0
|
|
|
|
Midi Handler :
|
|
|
|
- Hook to the MIDI host
|
|
- Enumerate connected midi devices and spawn a process/device to handle incoming events
|
|
|
|
by Sam Neurohack
|
|
from /team/laser
|
|
|
|
Midi conversions from https://github.com/craffel/pretty-midi
|
|
Guide to the MIDI Software Specification : http://www.somascape.org/midi/tech/spec.html#rpns
|
|
|
|
"""
|
|
|
|
import time
|
|
from threading import Thread
|
|
|
|
import rtmidi
|
|
from rtmidi.midiutil import open_midiinput
|
|
from rtmidi.midiconstants import (CHANNEL_PRESSURE, CONTROLLER_CHANGE, NOTE_ON, NOTE_OFF, SYSTEM_EXCLUSIVE, MIDI_TIME_CODE,
|
|
PITCH_BEND, POLY_PRESSURE, PROGRAM_CHANGE, TIMING_CLOCK, SONG_CONTINUE, SONG_START, SONG_STOP)
|
|
import mido
|
|
from mido import MidiFile
|
|
|
|
import traceback
|
|
import weakref
|
|
import sys
|
|
from sys import platform
|
|
import os
|
|
import re
|
|
from collections import deque
|
|
from libs import log, tools
|
|
import json
|
|
|
|
oscIP = "127.0.0.1"
|
|
oscPORT = 8000
|
|
|
|
is_py2 = sys.version[0] == '2'
|
|
if is_py2:
|
|
from queue import Queue
|
|
from OSC import OSCServer, OSCClient, OSCMessage
|
|
else:
|
|
from queue import Queue
|
|
from OSC3 import OSCServer, OSCClient, OSCMessage
|
|
|
|
|
|
print("")
|
|
|
|
midiname = ["Name"] * 16
|
|
midiport = [rtmidi.MidiOut() for i in range(16) ]
|
|
|
|
OutDevice = []
|
|
InDevice = []
|
|
|
|
midisync = True
|
|
debug = True
|
|
|
|
# max 16 midi port array
|
|
|
|
midinputsname = ["Name"] * 16
|
|
midinputsqueue = [Queue() for i in range(16) ]
|
|
midinputs = []
|
|
|
|
# False = server / True = Client
|
|
clientmode = False
|
|
|
|
#Mser = False
|
|
|
|
MidInsNumber = 0
|
|
|
|
|
|
clock = mido.Message(type="clock")
|
|
|
|
start = mido.Message(type ="start")
|
|
stop = mido.Message(type ="stop")
|
|
ccontinue = mido.Message(type ="continue")
|
|
reset = mido.Message(type ="reset")
|
|
songpos = mido.Message(type ="songpos")
|
|
|
|
nocolor = 64
|
|
green = 16
|
|
yellow = 127
|
|
red = 3
|
|
|
|
#mode = "maxwell"
|
|
|
|
'''
|
|
print "clock",clock)
|
|
print "start",start)
|
|
print "continue", ccontinue)
|
|
print "reset",reset)
|
|
print "sonpos",songpos)
|
|
'''
|
|
|
|
try:
|
|
input = raw_input
|
|
except NameError:
|
|
# Python 3
|
|
Exception = Exception
|
|
|
|
|
|
STATUS_MAP = {
|
|
'noteon': NOTE_ON,
|
|
'noteoff': NOTE_OFF,
|
|
'programchange': PROGRAM_CHANGE,
|
|
'controllerchange': CONTROLLER_CHANGE,
|
|
'pitchbend': PITCH_BEND,
|
|
'polypressure': POLY_PRESSURE,
|
|
'channelpressure': CHANNEL_PRESSURE
|
|
}
|
|
|
|
PadLeds = [0] * 64
|
|
PadTops= [0] * 8
|
|
PadRights= [0] * 8
|
|
LaunchLedMatrix = [(0,1,2,3,4,5,6,7),(16,17,18,19,20,21,22,23),(32,33,34,35,36,37,38,39),(48,49,50,51,52,53,54,55),(64,65,66,67,68,69,70,71),(80,81,82,83,84,85,86,87),(96,97,98,99,100,101,102,103),(112,113,114,115,116,117,118,119)]
|
|
LaunchRight = (8,24,40,56,72,88,104,120)
|
|
# CC
|
|
LaunchTop = (104,105,106,107,108,109,110,111)
|
|
PadTop = [0,0,0,0,0,0,0,0]
|
|
PadRight = [0,0,0,0,0,0,0,0]
|
|
PadMatrix = [0] * 64
|
|
matrix1 = [1,1]
|
|
matrix2 = [1,1]
|
|
matrix3 = [1,1]
|
|
TopSelection = [0] *8
|
|
|
|
def SendOSC(oscaddress,oscargs=''):
|
|
|
|
oscmsg = OSCMessage()
|
|
oscmsg.setAddress(oscaddress)
|
|
oscmsg.append(oscargs)
|
|
|
|
osclientlj = OSCClient()
|
|
osclientlj.connect((oscIP, oscPORT))
|
|
|
|
|
|
try:
|
|
osclientlj.sendto(oscmsg, (oscIP, oscPORT ))
|
|
oscmsg.clearData()
|
|
except:
|
|
log.err('Connection to OSC server refused : died ?')
|
|
pass
|
|
#time.sleep(0.001
|
|
|
|
def SendUI(oscaddress,oscargs=''):
|
|
|
|
oscmsg = OSCMessage()
|
|
oscmsg.setAddress(oscaddress)
|
|
oscmsg.append(oscargs)
|
|
|
|
osclientlj = OSCClient()
|
|
osclientlj.connect((TouchOSCIP, TouchOSCPort))
|
|
|
|
#print("MIDI Aurora sending UI :", oscmsg, "to",TouchOSCIP,":",TouchOSCPort)
|
|
try:
|
|
osclientlj.sendto(oscmsg, (TouchOSCIP, TouchOSCPort))
|
|
oscmsg.clearData()
|
|
except:
|
|
log.err('Connection to UI refused : died ?')
|
|
pass
|
|
#time.sleep(0.001
|
|
|
|
|
|
# Ask redis for a given key
|
|
def fromKey(keyname):
|
|
|
|
return r.get(keyname)
|
|
|
|
# Write to redis key
|
|
def toKey(keyname,keyvalue):
|
|
#print(keyname,keyvalue)
|
|
# Store encoded data in Redis
|
|
return r.set(keyname,keyvalue)
|
|
|
|
# Publish to redis key
|
|
def toKeyevent(eventname):
|
|
|
|
print("redis midi event key :", eventname)
|
|
r.publish("/midi/last_event", eventname)
|
|
|
|
|
|
|
|
def GetTime():
|
|
return time.strftime("%a, %d %b %Y %H:%M:%S", time.localtime())
|
|
|
|
notes = ["C","C#","D","D#","E","F","F#","G","G#","A","A#","B"]
|
|
def midi2note(midinote):
|
|
|
|
print("midinote",midinote, "note", notes[midinote%12]+str(round(midinote/12)))
|
|
return notes[midinote%12]+str(round(midinote/12))
|
|
|
|
|
|
def note2midi(note_name):
|
|
"""Converts a note name in the format
|
|
``'(note)(accidental)(octave number)'`` (e.g. ``'C#4'``) to MIDI note
|
|
number.
|
|
``'(note)'`` is required, and is case-insensitive.
|
|
``'(accidental)'`` should be ``''`` for natural, ``'#'`` for sharp and
|
|
``'!'`` or ``'b'`` for flat.
|
|
If ``'(octave)'`` is ``''``, octave 0 is assumed.
|
|
Parameters
|
|
----------
|
|
note_name : str
|
|
A note name, as described above.
|
|
Returns
|
|
-------
|
|
note_number : int
|
|
MIDI note number corresponding to the provided note name.
|
|
Notes
|
|
-----
|
|
Thanks to Brian McFee.
|
|
"""
|
|
|
|
# Map note name to the semitone
|
|
pitch_map = {'C': 0, 'D': 2, 'E': 4, 'F': 5, 'G': 7, 'A': 9, 'B': 11}
|
|
# Relative change in semitone denoted by each accidental
|
|
acc_map = {'#': 1, '': 0, 'b': -1, '!': -1}
|
|
|
|
# Reg exp will raise an error when the note name is not valid
|
|
try:
|
|
# Extract pitch, octave, and accidental from the supplied note name
|
|
match = re.match(r'^(?P<n>[A-Ga-g])(?P<off>[#b!]?)(?P<oct>[+-]?\d+)$',
|
|
note_name)
|
|
|
|
pitch = match.group('n').upper()
|
|
offset = acc_map[match.group('off')]
|
|
octave = int(match.group('oct'))
|
|
|
|
except:
|
|
raise ValueError('Improper note format: {}'.format(note_name))
|
|
# Convert from the extrated ints to a full note number
|
|
return 12*(octave + 1) + pitch_map[pitch] + offset
|
|
|
|
|
|
|
|
def hz2midi(frequency):
|
|
"""Convert a frequency in Hz to a (fractional) note number.
|
|
Parameters
|
|
----------
|
|
frequency : float
|
|
Frequency of the note in Hz.
|
|
Returns
|
|
-------
|
|
note_number : float
|
|
MIDI note number, can be fractional.
|
|
"""
|
|
# MIDI note numbers are defined as the number of semitones relative to C0
|
|
# in a 440 Hz tuning
|
|
return 12*(np.log2(frequency) - np.log2(440.0)) + 69
|
|
|
|
def midi2hz(note_number):
|
|
"""Convert a (fractional) MIDI note number to its frequency in Hz.
|
|
Parameters
|
|
----------
|
|
note_number : float
|
|
MIDI note number, can be fractional.
|
|
Returns
|
|
-------
|
|
note_frequency : float
|
|
Frequency of the note in Hz.
|
|
"""
|
|
# MIDI note numbers are defined as the number of semitones relative to C0
|
|
# in a 440 Hz tuning
|
|
return 440.0*(2.0**((note_number - 69)/12.0))
|
|
|
|
# /cc cc number value
|
|
def cc(midichannel, ccnumber, value, mididest):
|
|
|
|
if debug == True:
|
|
print("Midix sending Midi channel", midichannel, "cc", ccnumber, "value", value, "to", mididest)
|
|
|
|
MidiMsg([CONTROLLER_CHANGE+midichannel-1, ccnumber, value], mididest)
|
|
|
|
|
|
|
|
#
|
|
# MIDI Startup and handling
|
|
#
|
|
|
|
mqueue = Queue()
|
|
inqueue = Queue()
|
|
bpm = 0
|
|
running = True
|
|
samples = deque()
|
|
last_clock = None
|
|
|
|
|
|
#
|
|
# Events from Generic MIDI Handling
|
|
#
|
|
|
|
def MidinProcess(inqueue, portname):
|
|
|
|
inqueue_get = inqueue.get
|
|
bpm = 0
|
|
samples = deque()
|
|
last_clock = None
|
|
|
|
while True:
|
|
|
|
time.sleep(0.001)
|
|
msg = inqueue_get()
|
|
#print("")
|
|
#print("Generic from", portname,"msg : ", msg)
|
|
|
|
# NOTE ON message on all midi channels
|
|
if NOTE_ON -1 < msg[0] < 160 and msg[2] !=0 :
|
|
|
|
MidiChannel = msg[0]-144
|
|
MidiNote = msg[1]
|
|
MidiVel = msg[2]
|
|
print()
|
|
print("NOTE ON :", "Channel", MidiChannel, "note :", MidiNote, 'velocity :', MidiVel )
|
|
|
|
# redis key : "/midi/noteon/midichannel" value : "note/velocity"
|
|
if r.set("/midi/noteon/"+str(MidiChannel), str(MidiNote)+"/"+str(MidiVel))==True:
|
|
print("redis :", "/midi/noteon/"+str(MidiChannel)+" : "+ str(MidiNote)+"/"+str(MidiVel))
|
|
|
|
NoteOn(MidiNote, MidiVel, "pads" , midichannel=MidiChannel)
|
|
|
|
for param in conf['params']:
|
|
|
|
islj2 = param["name"].find('lasernumber')
|
|
if param["type"] == 0 and param["note"] == msg[1] and islj2 != -1:
|
|
|
|
# LJ2 parameter : replace lasernumber with channel number
|
|
tolj = param["name"].replace('lasernumber',str(MidiChannel))
|
|
print("LJ2 button :", tolj, ["1"])
|
|
SendOSC(tolj, ['1'])
|
|
|
|
|
|
# OSC : /midi/noteon midichannel note velocity
|
|
SendOSC("/midi/noteon",[MidiChannel, msg[1], msg[2]])
|
|
print("osc :","/midi/noteon/",[MidiChannel, msg[1], msg[2]])
|
|
toKeyevent("/midi/noteon/"+str(MidiChannel)+"/"+str(MidiNote)+"/"+str(MidiVel))
|
|
|
|
'''
|
|
# Sampler mode : note <63 launch snare.wav / note > 62 kick.wav
|
|
if MidiNote < 63 and MidiVel >0:
|
|
|
|
if platform == 'darwin':
|
|
os.system("afplay snare.wav")
|
|
else:
|
|
os.system("aplay snare.wav")
|
|
|
|
|
|
if MidiNote > 62 and MidiVel >0:
|
|
|
|
if platform == 'darwin':
|
|
os.system("afplay kick.wav")
|
|
else:
|
|
os.system("aplay kick.wav")
|
|
'''
|
|
|
|
# NOTE OFF or Note with 0 velocity on all midi channels
|
|
if NOTE_OFF -1 < msg[0] < 145 or (NOTE_OFF -1 < msg[0] < 160 and msg[2] == 0):
|
|
|
|
|
|
if msg[0] > 143:
|
|
MidiChannel = msg[0]-144
|
|
else:
|
|
MidiChannel = msg[0]-128
|
|
MidiNote = msg[1]
|
|
print()
|
|
print("NOTE_off channel :", MidiChannel, "note :", MidiNote)
|
|
NoteOff(MidiNote, "pads" , midichannel=MidiChannel)
|
|
|
|
# redis key : "/midi/noteon/midichannel" value : "note"
|
|
if r.set("/midi/noteoff/"+str(MidiChannel), str(MidiNote)) ==True:
|
|
print("redis :", "/midi/noteoff/"+str(MidiChannel)+" : "+ str(MidiNote))
|
|
|
|
# OSC : /midi/noteoff midichannel note
|
|
SendOSC("/midi/noteoff",[MidiChannel, msg[1]])
|
|
print('osc :', "/midi/noteoff",[MidiChannel, msg[1]])
|
|
|
|
|
|
# # CC on all Midi Channels
|
|
if CONTROLLER_CHANGE -1 < msg[0] < 192:
|
|
|
|
MidiChannel = msg[0]-175
|
|
print()
|
|
cc(MidiChannel, msg[1], msg[2], "pads" )
|
|
#print("channel", MidiChannel, "CC :", msg[1], msg[2])
|
|
print("CC : channel : "+str(msg[0]-175-1)+" CC : "+str(msg[1])+" value : "+str(msg[2]))
|
|
toKeyevent("/midi/cc/"+str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
|
|
# redis key : "/midi/cc/midichannel/ccnumber" value : "ccvalue"
|
|
if r.set("/midi/cc/"+str(MidiChannel)+"/"+str(msg[1]),str(msg[2]))==True:
|
|
print("redis :", "/midi/cc/"+str(MidiChannel)+"/"+str(msg[1]), ":", str(msg[2]))
|
|
|
|
# OSC : /midi/cc midichannel ccnumber value
|
|
SendOSC("/midi/cc",[msg[0]-175-1, msg[1], msg[2]])
|
|
#print("osc :","/midi/cc",[msg[0]-175-1, msg[1], msg[2]] )
|
|
|
|
|
|
for param in conf['params']:
|
|
|
|
#print()
|
|
#print('custom action',param["name"], param)
|
|
# print('type', param["type"] )
|
|
islj2 = param["name"].find('lasernumber')
|
|
# print('lasernumber position', islj2)
|
|
|
|
|
|
# 0-127 parameter
|
|
if param["type"] == 7:
|
|
|
|
# LJ2 parameter : replace lasernumber with channel number - 1
|
|
if islj2 != -1:
|
|
|
|
if param["CC"] == msg[1]:
|
|
|
|
print('lj2 7 bits custom action')
|
|
tolj = param["name"].replace('lasernumber',str(MidiChannel-1))
|
|
SendOSC(tolj, [midifactor(ccnumber = msg[1], channel = MidiChannel, low =param["low"], high=param["high"], default =1)])
|
|
|
|
print('tolj', tolj, midifactor(ccnumber = msg[1], channel = MidiChannel, low =param["low"], high=param["high"], default =1) )
|
|
#toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
#toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
#tolj = param["name"].replace('lasernumber',str(MidiChannel))
|
|
#print("LJ2 cc :", tolj, ["1"])
|
|
#SendOSC(tolj, ['1'])
|
|
|
|
# Not LJ2
|
|
elif MidiChannel == param["chanIN"] and param["CC"] == msg[1]:
|
|
#print(param["name"]+"/"+ str(msg[1])+"/"+str(msg[2]))
|
|
SendOSC(param["name"], [msg[0]-175-1, msg[1], msg[2]])
|
|
toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
|
|
|
|
# convert 7 bits midi cc to 8 bits and send to "name"
|
|
if param["type"] == 8:
|
|
|
|
# LJ2 parameter : replace lasernumber with channel number - 1
|
|
if islj2 != -1:
|
|
|
|
if param["CC"] == msg[1]:
|
|
print('lj2 8 bits custom action')
|
|
tolj = param["name"].replace('lasernumber',str(MidiChannel-1))
|
|
SendOSC(tolj, [msg[2]*2])
|
|
print('tolj', tolj, msg[2]*2 )
|
|
#toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
#toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
|
|
# Not LJ2
|
|
elif MidiChannel == param["chanIN"] and param["CC"] == msg[1]:
|
|
SendOSC(param["name"], [msg[0]-175-1, msg[1], msg[2]*2])
|
|
toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
|
|
|
|
# reranged 2 cc -> 14 bits parameters
|
|
if param["type"] == 14:
|
|
|
|
# LJ2 parameter : replace lasernumber with channel number - 1
|
|
if islj2 != -1:
|
|
if param["lowCC"] == msg[1] or param["highCC"] == msg[1]:
|
|
|
|
print('lj2 14 bits reranged custom action')
|
|
tolj = param["name"].replace('lasernumber',str(MidiChannel-1))
|
|
print('highcc',str(param["highCC"]),r.get("/midi/cc/"+str(MidiChannel)+"/"+str(param["highCC"])))
|
|
print('lowcc',str(param["lowCC"]),r.get("/midi/cc/"+str(MidiChannel)+"/"+str(param["lowCC"])))
|
|
|
|
SendOSC(tolj, [str(int(midifactor14(highcc=param["highCC"], lowcc=param["lowCC"], channel=MidiChannel, low=param["low"], high=param["high"], default=param["default"])))])
|
|
print('tolj', tolj, int(midifactor14(highcc=param["highCC"], lowcc=param["lowCC"], channel=MidiChannel, low=param["low"], high=param["high"], default=param["default"])) )
|
|
#toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
#toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]*2))
|
|
|
|
# Not LJ2
|
|
elif MidiChannel == param["chanIN"] and (param["lowCC"] == msg[1] or param["highCC"] == msg[1]):
|
|
#print(param["name"]+"/"+ str(msg[1])+"/"+str(msg[2]))
|
|
|
|
highcc = r.get("/midi/cc/"+str(MidiChannel)+"/"+str(msg[1]))
|
|
print('highcc',highcc)
|
|
midifactor14(highcc, lowcc, ccnumber, channel, param["low"], param["high"], default)
|
|
|
|
SendOSC(param["name"], [msg[0]-175-1, msg[1], msg[2]])
|
|
toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
toKey(param["name"],str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
|
|
|
|
# 14 bits parameters with one CC
|
|
if param["type"] == 140:
|
|
|
|
# LJ2 parameter : replace lasernumber with channel number - 1
|
|
if islj2 != -1:
|
|
if param["CC"] == msg[1]:
|
|
print('lj2 140 bits custom action')
|
|
tolj = param["name"].replace('lasernumber',str(MidiChannel-1))
|
|
print('no program here')
|
|
|
|
# Not LJ2
|
|
elif MidiChannel == param["chanIN"] and param["CC"] == msg[1]:
|
|
#print(param["name"]+"/"+ str(msg[1])+"/"+str(msg[2]))
|
|
SendOSC(param["name"], [msg[0]-175-1, msg[1], msg[2]])
|
|
toKeyevent(param["name"]+"/"+ str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
toKey(param["name"], str(MidiChannel)+"/"+str(msg[1])+"/"+str(msg[2]))
|
|
|
|
midifactor140(ccnumber, channel=MidiChannel, low=param["low"], high=param["high"], default = param["default"])
|
|
#midifactor14(highcc, lowcc, channel = cchannel, low =0, high=16383, default =1)
|
|
|
|
|
|
if msg[0] == TIMING_CLOCK:
|
|
now = time.time()
|
|
|
|
if last_clock is not None:
|
|
samples.append(now - last_clock)
|
|
last_clock = now
|
|
|
|
if len(samples) > 24:
|
|
samples.popleft()
|
|
|
|
if len(samples) >= 2:
|
|
#bpm = 2.5 / (sum(samples) / len(samples))
|
|
#print("%.2f bpm" % bpm)
|
|
|
|
bpm = round(2.5 / (sum(samples) / len(samples))) # Against BPM lot very tiny change :
|
|
sync = True
|
|
# print("MIDI BPM", bpm)
|
|
|
|
#print("Midi clock : BPM", bpm)
|
|
|
|
# OSC : "/midi/clock"
|
|
SendOSC("/midi/clock",[])
|
|
print("osc : /midi/clock")
|
|
# SendAU("/aurora/bpm",[bpm])
|
|
|
|
|
|
if msg[0] in (SONG_CONTINUE, SONG_START):
|
|
running = True
|
|
|
|
#print("START/CONTINUE received.")
|
|
#print("Midi in process send /aurora/start")
|
|
|
|
# OSC : /midi/start
|
|
SendOSC("/midi/start",[])
|
|
print("osc : /midi/start")
|
|
|
|
|
|
if msg[0] == SONG_STOP:
|
|
running = False
|
|
#print("STOP received.")
|
|
#print("Midi in process send /aurora/stop")
|
|
|
|
# OSC : /midi/start
|
|
SendOSC("/midi/stop",[])
|
|
print("osc : /midi/stop")
|
|
|
|
|
|
if msg[0] == SYSTEM_EXCLUSIVE:
|
|
print('sysex', msg)
|
|
# check to see if this is a timecode frame
|
|
if len(msg) == 8 and msg[0:4] == [127,127,1,1]:
|
|
data = message.data[4:]
|
|
tc = tools.mtc_decode(data)
|
|
print('FF:',tc)
|
|
|
|
|
|
# https://en.wikipedia.org/wiki/MIDI_timecode
|
|
if msg[0] == MIDI_TIME_CODE:
|
|
frame_type = (msg[1] >> 4) & 0xF
|
|
quarter_frames[frame_type] = msg[1] & 15
|
|
if frame_type == 7:
|
|
tc = tools.mtc_decode_quarter_frames(quarter_frames)
|
|
print('QF:',tc)
|
|
SendTCview("/MIDIQF",[str(tc)])
|
|
|
|
|
|
'''
|
|
# other midi message
|
|
if msg[0] != NOTE_OFF and msg[0] != NOTE_ON and msg[0] != CONTROLLER_CHANGE:
|
|
pass
|
|
|
|
print("from", portname,"other midi message")
|
|
MidiMsg(msg[0],msg[1],msg[2],mididest)
|
|
'''
|
|
|
|
#def NoteOn(note, color, mididest):
|
|
#https://pypi.org/project/python-rtmidi/0.3a/
|
|
# NOTE_ON=#90 et NOTE_OFF=#80 on ajoute le channel (0 le premier) pour envoyer effectivement sur le channel
|
|
def NoteOn(note, color, mididest, midichannel=0):
|
|
global MidInsNumber
|
|
|
|
if debug == True:
|
|
print("Sending", note, color, "to", mididest, "on channel", midichannel)
|
|
|
|
for port in range(MidInsNumber):
|
|
|
|
# To mididest
|
|
if midiname[port].find(mididest) == 0:
|
|
midiport[port].send_message([NOTE_ON+midichannel, note, color])
|
|
|
|
# To All
|
|
elif mididest == "all" and midiname[port].find(mididest) != 0:
|
|
midiport[port].send_message([NOTE_ON+midichannel, note, color])
|
|
|
|
|
|
elif mididest == "pads" and midiname[port].find("Launchpad") > -1:
|
|
midiport[port].send_message([NOTE_ON+midichannel, note, color])
|
|
|
|
if mode == "clitools":
|
|
x,y = PadIndex(note)
|
|
print("Y :",y)
|
|
if 0< y < 3:
|
|
ClsCli1()
|
|
if 2< y < 5:
|
|
ClsCli2()
|
|
if 4< y < 7:
|
|
ClsCli3()
|
|
if 6< y < 9:
|
|
ClsCli4()
|
|
if y < 8 and x < 8:
|
|
PadNoteOnXY(x,y,red)
|
|
|
|
|
|
|
|
|
|
def NoteOff(note, mididest, midichannel=0):
|
|
global MidInsNumber
|
|
|
|
|
|
for port in range(MidInsNumber):
|
|
|
|
# To mididest
|
|
if midiname[port].find(mididest) != -1:
|
|
midiport[port].send_message([NOTE_OFF, note, 0])
|
|
|
|
# To All
|
|
elif mididest == "all" and midiname[port].find(mididest) == -1:
|
|
midiport[port].send_message([NOTE_OFF, note, 0])
|
|
|
|
|
|
#elif mididest == "pads" and midiname[port].find("Launchpad") > -1:
|
|
# midiport[port].send_message([NOTE_OFF, note, 0])
|
|
|
|
|
|
# Generic call back : new msg forwarded to queue
|
|
class AddQueue(object):
|
|
def __init__(self, portname, port):
|
|
self.portname = portname
|
|
self.port = port
|
|
#print "AddQueue", port)
|
|
self._wallclock = time.time()
|
|
|
|
def __call__(self, event, data=None):
|
|
message, deltatime = event
|
|
self._wallclock += deltatime
|
|
#print "inqueue : [%s] @%0.6f %r" % ( self.portname, self._wallclock, message))
|
|
message.append(deltatime)
|
|
midinputsqueue[self.port].put(message)
|
|
|
|
|
|
#
|
|
# MIDI OUT Handling
|
|
#
|
|
|
|
|
|
class OutObject():
|
|
|
|
_instances = set()
|
|
counter = 0
|
|
|
|
def __init__(self, name, kind, port):
|
|
|
|
self.name = name
|
|
self.kind = kind
|
|
self.port = port
|
|
|
|
self._instances.add(weakref.ref(self))
|
|
OutObject.counter += 1
|
|
|
|
print("Adding OutDevice name", self.name, "kind", self.kind, "port", self.port)
|
|
|
|
@classmethod
|
|
def getinstances(cls):
|
|
dead = set()
|
|
for ref in cls._instances:
|
|
obj = ref()
|
|
if obj is not None:
|
|
yield obj
|
|
else:
|
|
dead.add(ref)
|
|
cls._instances -= dead
|
|
|
|
def __del__(self):
|
|
OutObject.counter -= 1
|
|
|
|
|
|
|
|
def OutConfig():
|
|
global midiout, MidInsNumber
|
|
|
|
#
|
|
if len(OutDevice) == 0:
|
|
print("")
|
|
log.info("MIDIout...")
|
|
print("List and attach to available devices on host with IN port :")
|
|
|
|
# Display list of available midi IN devices on the host, create and start an OUT instance to talk to each of these Midi IN devices
|
|
midiout = rtmidi.MidiOut()
|
|
available_ports = midiout.get_ports()
|
|
|
|
for port, name in enumerate(available_ports):
|
|
|
|
midiname[port]=name
|
|
midiport[port].open_port(port)
|
|
#print )
|
|
#print "New OutDevice [%i] %s" % (port, name))
|
|
|
|
OutDevice.append(OutObject(name, "generic", port))
|
|
|
|
# Search for a LaunchPad
|
|
if name.find("Launchpad") == 0:
|
|
|
|
OutDevice.append(OutObject(name, "launchpad", port))
|
|
print("Launchpad mini start animation")
|
|
Start()
|
|
time.sleep(0.2)
|
|
|
|
#print "")
|
|
print(len(OutDevice), "Out devices")
|
|
#ListOutDevice()
|
|
MidInsNumber = len(OutDevice)+1
|
|
|
|
def ListOutDevice():
|
|
|
|
for item in OutObject.getinstances():
|
|
|
|
print(item.name)
|
|
|
|
def FindOutDevice(name):
|
|
|
|
port = -1
|
|
for item in OutObject.getinstances():
|
|
#print "searching", name, "in", item.name)
|
|
if name == item.name:
|
|
#print 'found port',item.port)
|
|
port = item.port
|
|
return port
|
|
|
|
|
|
def DelOutDevice(name):
|
|
|
|
Outnumber = Findest(name)
|
|
print('deleting OutDevice', name)
|
|
|
|
if Outnumber != -1:
|
|
print('found OutDevice', Outnumber)
|
|
delattr(OutObject, str(name))
|
|
print("OutDevice", Outnumber,"was removed")
|
|
else:
|
|
print("OutDevice was not found")
|
|
|
|
|
|
|
|
#
|
|
# MIDI IN Handling
|
|
# Create processing thread and queue for each device
|
|
#
|
|
|
|
class InObject():
|
|
|
|
_instances = set()
|
|
counter = 0
|
|
|
|
def __init__(self, name, kind, port, rtmidi):
|
|
|
|
self.name = name
|
|
self.kind = kind
|
|
self.port = port
|
|
self.rtmidi = rtmidi
|
|
self.queue = Queue()
|
|
|
|
self._instances.add(weakref.ref(self))
|
|
InObject.counter += 1
|
|
|
|
print("Adding InDevice name", self.name, "kind", self.kind, "port", self.port)
|
|
|
|
@classmethod
|
|
def getinstances(cls):
|
|
dead = set()
|
|
for ref in cls._instances:
|
|
obj = ref()
|
|
if obj is not None:
|
|
yield obj
|
|
else:
|
|
dead.add(ref)
|
|
cls._instances -= dead
|
|
|
|
def __del__(self):
|
|
InObject.counter -= 1
|
|
|
|
|
|
def InConfig():
|
|
|
|
print("")
|
|
log.info("MIDIin...")
|
|
|
|
# client mode
|
|
if debug == True:
|
|
if clientmode == True:
|
|
print("midix in client mode")
|
|
else:
|
|
print("midix in server mode")
|
|
|
|
print("List and attach to available devices on host with OUT port :")
|
|
|
|
if platform == 'darwin':
|
|
mido.set_backend('mido.backends.rtmidi/MACOSX_CORE')
|
|
|
|
genericnumber = 0
|
|
|
|
for port, name in enumerate(mido.get_input_names()):
|
|
|
|
|
|
outport = FindOutDevice(name)
|
|
midinputsname[port]=name
|
|
|
|
#print "name",name, "Port",port, "Outport", outport)
|
|
# print "midinames", midiname)
|
|
|
|
#ListInDevice()
|
|
|
|
try:
|
|
#print name, name.find("RtMidi output"))
|
|
if name.find("RtMidi output") > -1:
|
|
print("No thread started for device", name)
|
|
else:
|
|
portin = object
|
|
port_name = ""
|
|
portin, port_name = open_midiinput(outport)
|
|
|
|
if midisync == True:
|
|
portin.ignore_types(timing=False)
|
|
|
|
#midinputs.append(portin)
|
|
InDevice.append(InObject(name, "generic", outport, portin))
|
|
|
|
thread = Thread(target=MidinProcess, args=(midinputsqueue[port],port_name))
|
|
thread.setDaemon(True)
|
|
thread.start()
|
|
|
|
#print "Thread launched for midi port", port, "portname", port_name, "Inname", midiname.index(port_name)
|
|
#print "counter", InObject.counter
|
|
#midinputs[port].set_callback(AddQueue(name),midinputsqueue[port])
|
|
#midinputs[port].set_callback(AddQueue(name))
|
|
#genericnumber += 1
|
|
InDevice[InObject.counter-1].rtmidi.set_callback(AddQueue(name,port))
|
|
#if name.find("Launch") > -1:
|
|
# Cls()
|
|
|
|
except Exception:
|
|
traceback.print_exc()
|
|
|
|
#print "")
|
|
print(InObject.counter, "In devices")
|
|
#ListInDevice()
|
|
|
|
|
|
def ListInDevice():
|
|
|
|
#print "known IN devices :"
|
|
for item in InObject.getinstances():
|
|
|
|
print(item.name)
|
|
print("")
|
|
|
|
def FindInDevice(name):
|
|
|
|
port = -1
|
|
for item in InObject.getinstances():
|
|
#print "searching", name, "in", item.name)
|
|
if name in item.name:
|
|
#print 'found port',item.port)
|
|
port = item.port
|
|
return port
|
|
|
|
|
|
def DelInDevice(name):
|
|
|
|
Innumber = Findest(name)
|
|
print('deleting InDevice', name)
|
|
|
|
if Innumber != -1:
|
|
print('found InDevice', Innumber)
|
|
delattr(InObject, str(name))
|
|
print("InDevice", Innumber,"was removed")
|
|
else:
|
|
print("InDevice was not found")
|
|
|
|
|
|
|
|
def End():
|
|
global midiout
|
|
|
|
#midiin.close_port()
|
|
midiout.close_port()
|
|
|
|
#del virtual
|
|
if launchpad.Here != -1:
|
|
del launchpad.Here
|
|
if bhoreal.Here != -1:
|
|
del bhoreal.Here
|
|
if LPD8.Here != -1:
|
|
del LPD8.Here
|
|
|
|
# mididest : all or specifiname, won't be sent to launchpad or Bhoreal.
|
|
def MidiMsg(midimsg, mididest):
|
|
|
|
|
|
desterror = -1
|
|
|
|
print("miredis got midimsg", midimsg, "for", mididest)
|
|
|
|
for port in range(len(OutDevice)):
|
|
# To mididest
|
|
if midiname[port].find(mididest) != -1:
|
|
if debug == True:
|
|
print("miredis sending to name", midiname[port], "port", port, ":", midimsg)
|
|
midiport[port].send_message(midimsg)
|
|
desterror = 0
|
|
|
|
elif mididest == "pads" and midiname[port].find("Launchpad") > -1:
|
|
midiport[port].send_message(midimsg)
|
|
desterror = 0
|
|
|
|
if desterror == -1:
|
|
print("mididest",mididest, ": ** This midi destination doesn't exists **")
|
|
|
|
# send midi msg over ws.
|
|
#if clientmode == True:
|
|
# ws.send("/ocs2/cc/1 2")
|
|
|
|
|
|
'''
|
|
def NoteOn(note, velocity, mididest):
|
|
global MidInsNumber
|
|
|
|
|
|
for port in range(MidInsNumber):
|
|
|
|
# To mididest
|
|
if midiname[port].find(mididest) == 0:
|
|
midiport[port].send_message([NOTE_ON, note, velocity])
|
|
'''
|
|
|
|
|
|
# 1 CC -> 7 bits number : 0-127
|
|
# rerange end value between low high
|
|
# midifactor(ccnumber, channel, low, high, default)
|
|
def midifactor(ccnumber, channel = 0, low =0, high=127, default =1):
|
|
|
|
ccvalue = r.get('/midi/cc/'+str(channel)+'/'+str(ccnumber))
|
|
|
|
if ccvalue is not None:
|
|
return rerange(int(ccvalue), 0, 127, low, high)
|
|
|
|
else:
|
|
print('Default value returned. No midi value in redis for channel', channel, 'CC', ccnumber)
|
|
return default
|
|
|
|
# 2 CC -> 14 bits number : 0 - 16383
|
|
# rerange end value between low high
|
|
# midifactor14(highcc, lowcc, ccnumber, channel, low, high, default)
|
|
def midifactor14(highcc, lowcc, channel = 0, low =0, high=16383, default =1):
|
|
|
|
lowvalue = r.get('/midi/cc/'+str(channel)+'/'+str(lowcc))
|
|
highvalue = r.get('/midi/cc/'+str(channel)+'/'+str(highcc))
|
|
|
|
if lowvalue is not None and highcc is not None:
|
|
return rerange((int(highvalue) << 7)+ int(lowvalue), 0, 16383, low, high)
|
|
|
|
else:
|
|
print('Default value returned. No midi value in redis for channel', channel, 'CCs', highvalue, lowvalue)
|
|
return default
|
|
|
|
|
|
# 1 CC -> 14 bits number : 0 - 16383
|
|
# rerange end value between low high
|
|
# set low cc (=Byte) to 0
|
|
# midifactor140(ccnumber, channel, low, high, default)
|
|
def midifactor140(highcc, channel = 0, low =0, high=16383, default =1):
|
|
|
|
highvalue = r.get('/midi/cc/'+str(channel)+'/'+str(highcc))
|
|
|
|
if highcc is not None:
|
|
return rerange((int(highvalue) << 7), 0, 16383, low, high)
|
|
|
|
else:
|
|
print('Default value returned. No midi value in redis for channel', channel, 'CC', ccnumber)
|
|
return default
|
|
|
|
|
|
|
|
def rerange(s,a1,a2,b1,b2):
|
|
return b1 + ((s - a1) * (b2 - b1) / (a2 - a1))
|
|
|
|
|
|
#
|
|
# launchpad
|
|
#
|
|
|
|
|
|
def PadNoteOn(note,color):
|
|
(x,y) = BhorIndex(note)
|
|
#print('PadNoteon', note, x, y, color)
|
|
PadNoteOnXY(x,y,color)
|
|
|
|
|
|
def PadNoteOff(note):
|
|
(x,y) = BhorIndex(note)
|
|
#print('PadNoteOFF', note, x, y)
|
|
PadNoteOffXY(x,y)
|
|
|
|
def PadNoteOnXY(x,y,color):
|
|
msg= [NOTE_ON, PadNoteXY(x,y), color]
|
|
#print(msg)
|
|
MidiMsg(msg,"Launchpad")
|
|
PadLeds[BhorNoteXY(x,y)]=color
|
|
|
|
|
|
def PadNoteOffXY(x,y):
|
|
msg= [NOTE_OFF, PadNoteXY(x,y), 0]
|
|
#print(msg)
|
|
MidiMsg(msg,"Launchpad")
|
|
PadLeds[BhorNoteXY(x,y)]=0
|
|
|
|
def PadNoteXY(x,y):
|
|
#print(str(x),str(y))
|
|
note = LaunchLedMatrix[int(y-1)][int(x-1)]
|
|
return note
|
|
|
|
def PadIndex(note):
|
|
y=note/16
|
|
x=note%16
|
|
return int(x+1),int(y+1)
|
|
|
|
def BhorIndex(note):
|
|
y=note/8
|
|
x=note%8
|
|
#print "Note : ",note
|
|
#print "BhorIndex : ", x+1,y+1
|
|
return int(x+1),int(y+1)
|
|
|
|
def BhorNoteXY(x,y):
|
|
note = (x -1)+ (y-1) * 8
|
|
return note
|
|
|
|
# top raw and right column leds are numbered humanly 1-8. So -1 is for pythonic arrays position 0-7
|
|
def PadTopOn(number, color):
|
|
msg= [CONTROLLER_CHANGE, LaunchTop[number-1], color]
|
|
MidiMsg(msg,"Launchpad")
|
|
PadTops[number-1]=color
|
|
|
|
def PadTopOff(number):
|
|
msg= [CONTROLLER_CHANGE, LaunchTop[number-1], 0]
|
|
MidiMsg(msg,"Launchpad")
|
|
PadTops[number-1]=0
|
|
|
|
def PadRightOn(number, color):
|
|
msg= [NOTE_ON, LaunchRight[number-1], color]
|
|
MidiMsg(msg,"Launchpad")
|
|
PadRights[number-1] = color
|
|
#UpdateAllCCs(number-1)
|
|
|
|
def PadRightOff(number):
|
|
msg= [NOTE_OFF, LaunchRight[number-1], 0]
|
|
MidiMsg(msg,"Launchpad")
|
|
PadRights[number-1] = 0
|
|
|
|
def TopUpdate(button, color):
|
|
#print(PadTop)
|
|
PadTop = [0,0,0,0,0,0,0,0]
|
|
PadTop[button] = color
|
|
for pad in range(7):
|
|
PadTopOn(pad+1, PadTop[pad])
|
|
|
|
def RightUpdate():
|
|
for pad in range(8):
|
|
#print(pad,PadRight[pad])
|
|
PadRightOn(pad, PadRight[pad])
|
|
if PadRight[pad] == 0:
|
|
SendOSCUI('/pad/r'+ str(pad) +'/button', [0])
|
|
else:
|
|
SendOSCUI('/pad/r'+ str(pad) +'/button', [1])
|
|
|
|
def MatrixUpdate():
|
|
for pad in range(64):
|
|
PadNoteOn(pad, PadMatrix[pad])
|
|
|
|
def MatrixSelect():
|
|
MatrixUpdate()
|
|
return
|
|
|
|
|
|
|
|
|
|
# AllColor for launchpad on given port
|
|
def AllColorPad(color):
|
|
|
|
print('AllColorPad')
|
|
for led in range(0,64,1):
|
|
PadNoteOn(led,color)
|
|
'''
|
|
for line in LaunchLedMatrix:
|
|
for led in line:
|
|
midiport[port].send_message([NOTE_ON, led, color])
|
|
'''
|
|
for rightled in range(8):
|
|
PadRightOn(rightled+1, color)
|
|
for topled in range(8):
|
|
PadTopOn(topled+1,color)
|
|
#midiport[port].send_message([CONTROLLER_CHANGE, topled, color])
|
|
|
|
# Led line 1,2 color 58
|
|
def ClsCli1():
|
|
|
|
for x in range(8):
|
|
PadNoteOnXY(x,1,58)
|
|
PadNoteOnXY(x,2,58)
|
|
|
|
# Led line 3,4 color yellow
|
|
def ClsCli2():
|
|
|
|
for x in range(8):
|
|
PadNoteOnXY(x,3,yellow)
|
|
PadNoteOnXY(x,4,yellow)
|
|
|
|
# Led line 5,6 color 58
|
|
def ClsCli3():
|
|
|
|
for x in range(8):
|
|
PadNoteOnXY(x,5,58)
|
|
PadNoteOnXY(x,6,58)
|
|
|
|
# Led line 7,8 color yellow
|
|
def ClsCli4():
|
|
|
|
for x in range(8):
|
|
PadNoteOnXY(x,7,yellow)
|
|
PadNoteOnXY(x,8,yellow)
|
|
|
|
|
|
def ClsCli():
|
|
|
|
ClsCli1()
|
|
ClsCli2()
|
|
ClsCli3()
|
|
ClsCli4()
|
|
|
|
|
|
def ClsMatrix():
|
|
for led in range(0,64,1):
|
|
PadNoteOff(led)
|
|
|
|
def ClsTop():
|
|
for topled in range(8):
|
|
PadTopOff(topled+1)
|
|
|
|
def ClsRight():
|
|
|
|
for rightled in range(8):
|
|
PadRightOff(rightled+1)
|
|
|
|
def Cls():
|
|
|
|
ClsMatrix()
|
|
ClsTop()
|
|
ClsRight()
|
|
|
|
def Start():
|
|
|
|
#ClsPad(port)
|
|
#time.sleep(0.3)
|
|
ClsTop()
|
|
ClsRight()
|
|
#AllColorPad(20)
|
|
time.sleep(1)
|
|
|
|
for color in range(64,128,1):
|
|
#AllColorPad(color)
|
|
PadNoteOn(color-64, color)
|
|
#print("color", color)
|
|
time.sleep(0.5)
|
|
Cls()
|
|
|
|
if mode == "clitools":
|
|
ClsCli()
|
|
for y in range(0,8,2):
|
|
PadNoteOnXY(1,y-1,red)
|
|
|
|
def listdevice(number):
|
|
|
|
return midiname[number]
|
|
|
|
def loadConf():
|
|
global conf, nbparam
|
|
|
|
try:
|
|
ConFile = 'miredis.json'
|
|
f = open(ConFile,"r")
|
|
s = f.read()
|
|
conf = json.loads(s)
|
|
print(len(conf['params']), "custom actions")
|
|
nbparam = len(conf['params'])
|
|
|
|
#print(conf)
|
|
return True
|
|
except Exception as e:
|
|
print("_loadPlaylist error when loading '{}':{}".format(ConFile,e))
|
|
|
|
def check():
|
|
|
|
OutConfig()
|
|
InConfig()
|
|
|
|
|
|
'''
|
|
|
|
from rtmidi.midiconstants import (CONTROL_CHANGE, DATA_DECREMENT,
|
|
DATA_ENTRY_LSB, DATA_ENTRY_MSB,
|
|
DATA_INCREMENT, RPN_LSB, RPN_MSB)
|
|
from rtmidi.midiutil import open_midiinput
|
|
|
|
|
|
class RPNDecoder:
|
|
def __init__(self, channel=1):
|
|
self.channel = (channel - 1) & 0xF
|
|
self.rpn = 0
|
|
self.values = defaultdict(int)
|
|
self.last_changed = None
|
|
|
|
def __call__(self, event, data=None):
|
|
msg, deltatime = event
|
|
|
|
# event type = upper four bits of first byte
|
|
if msg[0] == (CONTROL_CHANGE | self.channel):
|
|
cc, value = msg[1], msg[2]
|
|
|
|
if cc == RPN_LSB:
|
|
self.rpn = (self.rpn >> 7) * 128 + value
|
|
elif cc == RPN_MSB:
|
|
self.rpn = value * 128 + (self.rpn & 0x7F)
|
|
elif cc == DATA_INCREMENT:
|
|
self.set_rpn(self.rpn, min(2 ** 14, self.values[self.rpn] + 1))
|
|
elif cc == DATA_DECREMENT:
|
|
self.set_rpn(self.rpn, max(0, self.values[self.rpn] - 1))
|
|
elif cc == DATA_ENTRY_LSB:
|
|
self.set_rpn(self.rpn,
|
|
(self.values[self.rpn] >> 7) * 128 + value)
|
|
elif cc == DATA_ENTRY_MSB:
|
|
self.set_rpn(self.rpn,
|
|
value * 128 + (self.values[self.rpn] & 0x7F))
|
|
|
|
def set_rpn(self, rpn, value):
|
|
self.values[rpn] = value
|
|
self.last_changed = rpn
|
|
|
|
'''
|