[libsigrokdecode.git] / decoders / midi / pd.py

##
## This file is part of the libsigrokdecode project.
##
## Copyright (C) 2013 Uwe Hermann <uwe@hermann-uwe.de>
##
## This program is free software; you can redistribute it and/or modify
## it under the terms of the GNU General Public License as published by
## the Free Software Foundation; either version 2 of the License, or
## (at your option) any later version.
##
## This program is distributed in the hope that it will be useful,
## but WITHOUT ANY WARRANTY; without even the implied warranty of
## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
## GNU General Public License for more details.
##
## You should have received a copy of the GNU General Public License
## along with this program; if not, write to the Free Software
## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA  02110-1301 USA
##

import sigrokdecode as srd
from .lists import *

RX = 0
TX = 1

class Decoder(srd.Decoder):
    api_version = 2
    id = 'midi'
    name = 'MIDI'
    longname = 'Musical Instrument Digital Interface'
    desc = 'Musical Instrument Digital Interface (MIDI) protocol.'
    license = 'gplv2+'
    inputs = ['uart']
    outputs = ['midi']
    annotations = (
        ('text-verbose', 'Human-readable text (verbose)'),
    )

    def __init__(self):
        self.cmd = []
        self.state = 'IDLE'
        self.ss = None
        self.es = None
        self.ss_block = None
        self.es_block = None

    def start(self):
        self.out_ann = self.register(srd.OUTPUT_ANN)

    def putx(self, data):
        self.put(self.ss_block, self.es_block, self.out_ann, data)

    def get_note_name(self, channel, note):
        if channel != 10:
            return chromatic_notes[note]
        else:
            return 'assuming ' + percussion_notes.get(note, 'undefined')

    def handle_channel_msg_0x80(self):
        # Note off: 8n kk vv
        # n = channel, kk = note, vv = velocity
        c = self.cmd
        if len(c) < 3:
            return
        self.es_block = self.es
        msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        note_name = self.get_note_name(chan, note)
        self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
                  (chan, status_bytes[msg][0], note, note_name, velocity),
                  'ch %d: %s %d, velocity = %d' % \
                  (chan, status_bytes[msg][1], note, velocity),
                  '%d: %s %d, vel %d' % \
                  (chan, status_bytes[msg][2], note, velocity)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0x90(self):
        # Note on: 9n kk vv
        # n = channel, kk = note, vv = velocity
        # If velocity == 0 that actually means 'note off', though.
        c = self.cmd
        if len(c) < 3:
            return
        self.es_block = self.es
        msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        s = status_bytes[0x80] if (velocity == 0) else status_bytes[msg]
        note_name = self.get_note_name(chan, note)
        self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
                  (chan, s[0], note, note_name, velocity),
                  'ch %d: %s %d, velocity = %d' % \
                  (chan, s[1], note, velocity),
                  '%d: %s %d, vel %d' % \
                  (chan, s[2], note, velocity)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0xa0(self):
        # Polyphonic key pressure / aftertouch: An kk vv
        # n = channel, kk = polyphonic key pressure, vv = pressure value
        c = self.cmd
        if len(c) < 3:
            return
        self.es_block = self.es
        msg, chan, note, pressure = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        note_name = self.get_note_name(chan, note)
        self.putx([0, ['Channel %d: %s of %d for note = %d \'%s\'' % \
                  (chan, status_bytes[msg][0], pressure, note, note_name),
                  'ch %d: %s %d for note %d' % \
                  (chan, status_bytes[msg][1], pressure, note),
                  '%d: %s %d, N %d' % \
                  (chan, status_bytes[msg][2], pressure, note)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_controller_0x44(self):
        # Legato footswitch: Bn 44 vv
        # n = channel, vv = value (<= 0x3f: normal, > 0x3f: legato)
        c = self.cmd
        msg, chan, vv = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[2]
        t = ('normal', 'no') if vv <= 0x3f else ('legato', 'yes')
        self.putx([0, ['Channel %d: %s \'%s\' = %s' % \
                  (chan, status_bytes[msg][0],
                  control_functions[0x44][0], t[0]),
                  'ch %d: %s \'%s\' = %s' % \
                  (chan, status_bytes[msg][1],
                  control_functions[0x44][1], t[0]),
                  '%d: %s \'%s\' = %s' % \
                  (chan, status_bytes[msg][2],
                  control_functions[0x44][2], t[1])]])

    def handle_controller_0x54(self):
        # Portamento control (PTC): Bn 54 kk
        # n = channel, kk = source note for pitch reference
        c = self.cmd
        msg, chan, kk = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[2]
        kk_name = self.get_note_name(chan, kk)
        self.putx([0, ['Channel %d: %s \'%s\' (source note = %d / %s)' % \
                  (chan, status_bytes[msg][0],
                  control_functions[0x54][0], kk, kk_name),
                  'ch %d: %s \'%s\' (source note = %d)' % \
                  (chan, status_bytes[msg][1],
                  control_functions[0x54][1], kk),
                  '%d: %s \'%s\' (src N %d)' % \
                  (chan, status_bytes[msg][2],
                  control_functions[0x54][2], kk)]])

    def handle_controller_generic(self):
        c = self.cmd
        msg, chan, fn, param = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        default_name = 'undefined'
        ctrl_fn = control_functions.get(fn, default_name)
        if ctrl_fn == default_name:
            ctrl_fn = ('undefined 0x%02x' % fn, 'undef 0x%02x' % fn, '0x%02x' % fn)
        self.putx([0, ['Channel %d: %s \'%s\' (param = 0x%02x)' % \
                  (chan, status_bytes[msg][0], ctrl_fn[0], param),
                  'ch %d: %s \'%s\' (param = 0x%02x)' % \
                  (chan, status_bytes[msg][1], ctrl_fn[1], param),
                  '%d: %s \'%s\' is 0x%02x' % \
                  (chan, status_bytes[msg][2], ctrl_fn[2], param)]])

    def handle_channel_mode(self):
        # Channel Mode: Bn mm vv
        # n = channel, mm = mode number (120 - 127), vv = value
        c = self.cmd
        msg, chan, mm, vv = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        mode_fn = control_functions.get(mm, ('undefined', 'undef', 'undef'))
        # Decode the value based on the mode number.
        vv_string = ('', '')
        if mm == 122:           # mode = local control?
            if vv == 0:
                vv_string = ('off', 'off')
            elif vv == 127:     # mode = poly mode on?
                vv_string = ('on', 'on')
            else:
                vv_string = ('(non-standard param value of 0x%02x)' % vv,
                            '0x%02x' % vv)
        elif mm == 126:         # mode = mono mode on?
            if vv != 0:
                vv_string = ('(%d channels)' % vv, '(%d ch)' % vv)
            else:
                vv_string = ('(channels \'basic\' through 16)',
                            '(ch \'basic\' thru 16)')
        elif vv != 0: # All other channel mode messages expect vv == 0.
            vv_string = ('(non-standard param value of 0x%02x)' % vv,
                        '0x%02x' % vv)
        self.putx([0, ['Channel %d: %s \'%s\' %s' % \
                      (chan, status_bytes[msg][0], mode_fn[0], vv_string[0]),
                      'ch %d: %s \'%s\' %s' % \
                      (chan, status_bytes[msg][1], mode_fn[1], vv_string[1]),
                      '%d: %s \'%s\' %s' % \
                      (chan, status_bytes[msg][2], mode_fn[2], vv_string[1])]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0xb0(self):
        # Control change (or channel mode messages): Bn cc vv
        # n = channel, cc = control number (0 - 119), vv = control value
        c = self.cmd
        if len(c) < 3:
            return
        self.es_block = self.es
        if c[1] in range(0x78, 0x7f + 1):
            self.handle_channel_mode()
            return
        handle_ctrl = getattr(self, 'handle_controller_0x%02x' % c[1],
                              self.handle_controller_generic)
        handle_ctrl()
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0xc0(self):
        # Program change: Cn pp
        # n = channel, pp = program number (0 - 127)
        c = self.cmd
        if len(c) < 2:
            return
        self.es_block = self.es
        msg, chan, pp = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
                        self.cmd[1] + 1
        change_type = 'instrument'
        name = ''
        if chan != 10:  # channel != percussion
            name = gm_instruments.get(pp, 'undefined')
        else:
            change_type = 'drum kit'
            name = drum_kit.get(pp, 'undefined')
        self.putx([0, ['Channel %d: %s to %s %d (assuming %s)' % \
            (chan, status_bytes[msg][0], change_type, pp, name),
            'ch %d: %s to %s %d' % \
            (chan, status_bytes[msg][1], change_type, pp),
            '%d: %s %d' % \
            (chan, status_bytes[msg][2], pp)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0xd0(self):
        # Channel pressure / aftertouch: Dn vv
        # n = channel, vv = pressure value
        c = self.cmd
        if len(c) < 2:
            return
        self.es_block = self.es
        msg, chan, vv = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
                        self.cmd[1]
        self.putx([0, ['Channel %d: %s %d' % (chan, status_bytes[msg][0], vv),
                      'ch %d: %s %d' % (chan, status_bytes[msg][1], vv),
                      '%d: %s %d' % (chan, status_bytes[msg][2], vv)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_0xe0(self):
        # Pitch bend change: En ll mm
        # n = channel, ll = pitch bend change LSB, mm = pitch bend change MSB
        c = self.cmd
        if len(c) < 3:
            return
        self.es_block = self.es
        msg, chan, ll, mm = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
                            self.cmd[1], self.cmd[2]
        decimal = (mm << 7) + ll
        self.putx([0, ['Channel %d: %s 0x%02x 0x%02x (%d)' % \
                      (chan, status_bytes[msg][0], ll, mm, decimal),
                      'ch %d: %s 0x%02x 0x%02x (%d)' % \
                      (chan, status_bytes[msg][1], ll, mm, decimal),
                      '%d: %s (%d)' % \
                      (chan, status_bytes[msg][2], decimal)]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg_generic(self):
        # TODO: It should not be possible to hit this code.
        # It currently can not be unit tested.
        msg_type = self.cmd[0] & 0xf0
        self.es_block = self.es
        self.putx([0, ['Unknown channel message type: 0x%02x' % msg_type]])
        self.cmd, self.state = [], 'IDLE'

    def handle_channel_msg(self, newbyte):
        self.cmd.append(newbyte)
        msg_type = self.cmd[0] & 0xf0
        handle_msg = getattr(self, 'handle_channel_msg_0x%02x' % msg_type,
                             self.handle_channel_msg_generic)
        handle_msg()

    def handle_sysex_msg(self, newbyte):
        # SysEx message: 1 status byte, 1-3 manuf. bytes, x data bytes, EOX byte
        self.cmd.append(newbyte)
        if newbyte != 0xf7: # EOX
            return
        self.es_block = self.es
        # Note: Unlike other methods, this code pops bytes out of self.cmd
        # to isolate the data.
        msg, eox = self.cmd.pop(0), self.cmd.pop()
        if len(self.cmd) < 1:
            self.putx([0, ['%s: truncated manufacturer code (<1 bytes)' % \
                          status_bytes[msg][0],
                          '%s: truncated manufacturer (<1 bytes)' % \
                          status_bytes[msg][1],
                          '%s: trunc. manu.' % status_bytes[msg][2]]])
            self.cmd, self.state = [], 'IDLE'
            return
        # Extract the manufacturer name (or SysEx realtime or non-realtime).
        m1 = self.cmd.pop(0)
        manu = (m1,)
        if m1 == 0x00:  # If byte == 0, then 2 more manufacturer bytes follow.
            if len(self.cmd) < 2:
                self.putx([0, ['%s: truncated manufacturer code (<3 bytes)' % \
                          status_bytes[msg][0],
                          '%s: truncated manufacturer (<3 bytes)' % \
                          status_bytes[msg][1],
                          '%s: trunc. manu.' % status_bytes[msg][2]]])
                self.cmd, self.state = [], 'IDLE'
                return
            manu = (m1, self.cmd.pop(0), self.cmd.pop(0))
        default_name = 'undefined'
        manu_name = sysex_manufacturer_ids.get(manu, default_name)
        if manu_name == default_name:
            if len(manu) == 3:
                manu_name = ('%s (0x%02x 0x%02x 0x%02x)' % \
                            (default_name, manu[0], manu[1], manu[2]),
                            default_name)
            else:
                manu_name = ('%s (0x%02x)' % (default_name, manu[0]),
                            default_name)
        else:
            manu_name = (manu_name, manu_name)
        # Extract the payload, display in 1 of 2 formats
        # TODO: Write methods to decode SysEx realtime & non-realtime payloads.
        payload0 = ''
        payload1 = ''
        while len(self.cmd) > 0:
            byte = self.cmd.pop(0)
            payload0 += '0x%02x ' % (byte)
            payload1 += '%02x ' % (byte)
        if payload0 == '':
            payload0 = '<empty>'
            payload1 = '<>'
        payload = (payload0, payload1)
        self.putx([0, ['%s: for \'%s\' with payload %s' % \
                      (status_bytes[msg][0], manu_name[0], payload[0]),
                      '%s: \'%s\', payload %s' % \
                      (status_bytes[msg][1], manu_name[1], payload[1]),
                      '%s: \'%s\', payload %s' % \
                      (status_bytes[msg][2], manu_name[1], payload[1])]])
        self.cmd, self.state = [], 'IDLE'

    def handle_syscommon_midi_time_code_quarter_frame_msg(self, newbyte):
        # MIDI time code quarter frame: F1 nd
        # n = message type
        # d = values
        c = self.cmd
        if len(c) < 2:
            return
        msg = self.cmd[0]
        nn, dd = (self.cmd[1] & 0x70) >> 4, self.cmd[1] & 0x0f
        group = ('System Common', 'SysCom', 'SC')
        self.es_block = self.es
        if nn != 7: # If message type does not contain SMPTE type.
            self.putx([0, ['%s: %s of %s, value 0x%01x' % \
                          (group[0], status_bytes[msg][0],
                          quarter_frame_type[nn][0], dd),
                          '%s: %s of %s, value 0x%01x' % \
                          (group[1], status_bytes[msg][1],
                          quarter_frame_type[nn][1], dd),
                          '%s: %s of %s, value 0x%01x' % \
                          (group[2], status_bytes[msg][2],
                          quarter_frame_type[nn][1], dd)]])
            self.cmd, self.state = [], 'IDLE'
            return
        tt = (dd & 0x6) >> 1
        self.putx([0, ['%s: %s of %s, value 0x%01x for %s' % \
                      (group[0], status_bytes[msg][0], \
                      quarter_frame_type[nn][0], dd, smpte_type[tt]),
                      '%s: %s of %s, value 0x%01x for %s' % \
                      (group[1], status_bytes[msg][1], \
                      quarter_frame_type[nn][1], dd, smpte_type[tt]),
                      '%s: %s of %s, value 0x%01x for %s' % \
                      (group[2], status_bytes[msg][2], \
                      quarter_frame_type[nn][1], dd, smpte_type[tt])]])
        self.cmd, self.state = [], 'IDLE'

    def handle_syscommon_msg(self, newbyte):
        # System common messages
        #
        # There are 5 simple formats (which are directly handled here) and
        # 1 complex one called MIDI time code quarter frame.
        #
        # Note: While the MIDI lists 0xf7 as a "system common" message, it
        # is actually only used with SysEx messages so it is processed there.
        self.cmd.append(newbyte)
        msg = self.cmd[0]
        c = self.cmd
        group = ('System Common', 'SysCom', 'SC')
        if msg == 0xf1:
            # MIDI time code quarter frame
            self.handle_syscommon_midi_time_code_quarter_frame_msg(newbyte)
            return
        elif msg == 0xf2:
            # Song position pointer: F2 ll mm
            # ll = LSB position, mm = MSB position
            if len(c) < 3:
                return
            ll, mm = self.cmd[1], self.cmd[2]
            decimal = (mm << 7) + ll
            self.es_block = self.es
            self.putx([0, ['%s: %s 0x%02x 0x%02x (%d)' % \
                          (group[0], status_bytes[msg][0], ll, mm, decimal),
                          '%s: %s 0x%02x 0x%02x (%d)' % \
                          (group[1], status_bytes[msg][1], ll, mm, decimal),
                          '%s: %s (%d)' % \
                          (group[2], status_bytes[msg][2], decimal)]])
        elif msg == 0xf3:
            # Song select: F3 ss
            # ss = song selection number
            if len(c) < 2:
                return
            ss = self.cmd[1]
            self.es_block = self.es
            self.putx([0, ['%s: %s number %d' % \
                          (group[0], status_bytes[msg][0], ss),
                          '%s: %s number %d' % \
                          (group[1], status_bytes[msg][1], ss),
                          '%s: %s # %d' % \
                          (group[2], status_bytes[msg][2], ss)]])
        elif msg == 0xf4 or msg == 0xf5 or msg == 0xf6:
            # Undefined 0xf4, Undefined 0xf5, and Tune Request (respectively).
            # All are only 1 byte long with no data bytes.
            self.es_block = self.es
            self.putx([0, ['%s: %s' % (group[0], status_bytes[msg][0]),
                          '%s: %s' % (group[1], status_bytes[msg][1]),
                          '%s: %s' % (group[2], status_bytes[msg][2])]])
        self.cmd, self.state = [], 'IDLE'

    def handle_sysrealtime_msg(self, newbyte):
        # System realtime message: 0b11111ttt (t = message type)
        self.es_block = self.es
        group = ('System Realtime', 'SysReal', 'SR')
        self.putx([0, ['%s: %s' % (group[0], status_bytes[newbyte][0]),
                      '%s: %s' % (group[1], status_bytes[newbyte][1]),
                      '%s: %s' % (group[2], status_bytes[newbyte][2])]])
        self.cmd, self.state = [], 'IDLE'

    def decode(self, ss, es, data):
        ptype, rxtx, pdata = data

        # For now, ignore all UART packets except the actual data packets.
        if ptype != 'DATA':
            return

        self.ss, self.es = ss, es

        # We're only interested in the byte value (not individual bits).
        pdata = pdata[0]

        # Short MIDI overview:
        #  - Status bytes are 0x80-0xff, data bytes are 0x00-0x7f.
        #  - Most messages: 1 status byte, 1-2 data bytes.
        #  - Real-time system messages: always 1 byte.
        #  - SysEx messages: 1 status byte, n data bytes, EOX byte.

        # State machine.
        if self.state == 'IDLE':
            # Wait until we see a status byte (bit 7 must be set).
            if pdata < 0x80:
                return # TODO: How to handle? Ignore?
            # This is a status byte, remember the start sample.
            self.ss_block = ss
            if pdata in range(0x80, 0xef + 1):
                self.state = 'HANDLE CHANNEL MSG'
            elif pdata == 0xf0 or pdata == 0xf7:
                self.state = 'HANDLE SYSEX MSG'
            elif pdata in range(0xf1, 0xf7):
                self.state = 'HANDLE SYSCOMMON MSG'
            elif pdata in range(0xf8, 0xff + 1):
                self.state = 'HANDLE SYSREALTIME MSG'

        # Yes, this is intentionally _not_ an 'elif' here.
        if self.state == 'HANDLE CHANNEL MSG':
            self.handle_channel_msg(pdata)
        elif self.state == 'HANDLE SYSEX MSG':
            self.handle_sysex_msg(pdata)
        elif self.state == 'HANDLE SYSCOMMON MSG':
            self.handle_syscommon_msg(pdata)
        elif self.state == 'HANDLE SYSREALTIME MSG':
            self.handle_sysrealtime_msg(pdata)
Commit	Line	Data
	1	##
	2	## This file is part of the libsigrokdecode project.
	3	##
	4	## Copyright (C) 2013 Uwe Hermann <uwe@hermann-uwe.de>
	5	##
	6	## This program is free software; you can redistribute it and/or modify
	7	## it under the terms of the GNU General Public License as published by
	8	## the Free Software Foundation; either version 2 of the License, or
	9	## (at your option) any later version.
	10	##
	11	## This program is distributed in the hope that it will be useful,
	12	## but WITHOUT ANY WARRANTY; without even the implied warranty of
	13	## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
	14	## GNU General Public License for more details.
	15	##
	16	## You should have received a copy of the GNU General Public License
	17	## along with this program; if not, write to the Free Software
	18	## Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA
	19	##
	20
	21	import sigrokdecode as srd
	22	from .lists import *
	23
	24	RX = 0
	25	TX = 1
	26
	27	class Decoder(srd.Decoder):
	28	api_version = 2
	29	id = 'midi'
	30	name = 'MIDI'
	31	longname = 'Musical Instrument Digital Interface'
	32	desc = 'Musical Instrument Digital Interface (MIDI) protocol.'
	33	license = 'gplv2+'
	34	inputs = ['uart']
	35	outputs = ['midi']
	36	annotations = (
	37	('text-verbose', 'Human-readable text (verbose)'),
	38	)
	39
	40	def __init__(self):
	41	self.cmd = []
	42	self.state = 'IDLE'
	43	self.ss = None
	44	self.es = None
	45	self.ss_block = None
	46	self.es_block = None
	47
	48	def start(self):
	49	self.out_ann = self.register(srd.OUTPUT_ANN)
	50
	51	def putx(self, data):
	52	self.put(self.ss_block, self.es_block, self.out_ann, data)
	53
	54	def get_note_name(self, channel, note):
	55	if channel != 10:
	56	return chromatic_notes[note]
	57	else:
	58	return 'assuming ' + percussion_notes.get(note, 'undefined')
	59
	60	def handle_channel_msg_0x80(self):
	61	# Note off: 8n kk vv
	62	# n = channel, kk = note, vv = velocity
	63	c = self.cmd
	64	if len(c) < 3:
	65	return
	66	self.es_block = self.es
	67	msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	68	note_name = self.get_note_name(chan, note)
	69	self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
	70	(chan, status_bytes[msg][0], note, note_name, velocity),
	71	'ch %d: %s %d, velocity = %d' % \
	72	(chan, status_bytes[msg][1], note, velocity),
	73	'%d: %s %d, vel %d' % \
	74	(chan, status_bytes[msg][2], note, velocity)]])
	75	self.cmd, self.state = [], 'IDLE'
	76
	77	def handle_channel_msg_0x90(self):
	78	# Note on: 9n kk vv
	79	# n = channel, kk = note, vv = velocity
	80	# If velocity == 0 that actually means 'note off', though.
	81	c = self.cmd
	82	if len(c) < 3:
	83	return
	84	self.es_block = self.es
	85	msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	86	s = status_bytes[0x80] if (velocity == 0) else status_bytes[msg]
	87	note_name = self.get_note_name(chan, note)
	88	self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
	89	(chan, s[0], note, note_name, velocity),
	90	'ch %d: %s %d, velocity = %d' % \
	91	(chan, s[1], note, velocity),
	92	'%d: %s %d, vel %d' % \
	93	(chan, s[2], note, velocity)]])
	94	self.cmd, self.state = [], 'IDLE'
	95
	96	def handle_channel_msg_0xa0(self):
	97	# Polyphonic key pressure / aftertouch: An kk vv
	98	# n = channel, kk = polyphonic key pressure, vv = pressure value
	99	c = self.cmd
	100	if len(c) < 3:
	101	return
	102	self.es_block = self.es
	103	msg, chan, note, pressure = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	104	note_name = self.get_note_name(chan, note)
	105	self.putx([0, ['Channel %d: %s of %d for note = %d \'%s\'' % \
	106	(chan, status_bytes[msg][0], pressure, note, note_name),
	107	'ch %d: %s %d for note %d' % \
	108	(chan, status_bytes[msg][1], pressure, note),
	109	'%d: %s %d, N %d' % \
	110	(chan, status_bytes[msg][2], pressure, note)]])
	111	self.cmd, self.state = [], 'IDLE'
	112
	113	def handle_controller_0x44(self):
	114	# Legato footswitch: Bn 44 vv
	115	# n = channel, vv = value (<= 0x3f: normal, > 0x3f: legato)
	116	c = self.cmd
	117	msg, chan, vv = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[2]
	118	t = ('normal', 'no') if vv <= 0x3f else ('legato', 'yes')
	119	self.putx([0, ['Channel %d: %s \'%s\' = %s' % \
	120	(chan, status_bytes[msg][0],
	121	control_functions[0x44][0], t[0]),
	122	'ch %d: %s \'%s\' = %s' % \
	123	(chan, status_bytes[msg][1],
	124	control_functions[0x44][1], t[0]),
	125	'%d: %s \'%s\' = %s' % \
	126	(chan, status_bytes[msg][2],
	127	control_functions[0x44][2], t[1])]])
	128
	129	def handle_controller_0x54(self):
	130	# Portamento control (PTC): Bn 54 kk
	131	# n = channel, kk = source note for pitch reference
	132	c = self.cmd
	133	msg, chan, kk = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[2]
	134	kk_name = self.get_note_name(chan, kk)
	135	self.putx([0, ['Channel %d: %s \'%s\' (source note = %d / %s)' % \
	136	(chan, status_bytes[msg][0],
	137	control_functions[0x54][0], kk, kk_name),
	138	'ch %d: %s \'%s\' (source note = %d)' % \
	139	(chan, status_bytes[msg][1],
	140	control_functions[0x54][1], kk),
	141	'%d: %s \'%s\' (src N %d)' % \
	142	(chan, status_bytes[msg][2],
	143	control_functions[0x54][2], kk)]])
	144
	145	def handle_controller_generic(self):
	146	c = self.cmd
	147	msg, chan, fn, param = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	148	default_name = 'undefined'
	149	ctrl_fn = control_functions.get(fn, default_name)
	150	if ctrl_fn == default_name:
	151	ctrl_fn = ('undefined 0x%02x' % fn, 'undef 0x%02x' % fn, '0x%02x' % fn)
	152	self.putx([0, ['Channel %d: %s \'%s\' (param = 0x%02x)' % \
	153	(chan, status_bytes[msg][0], ctrl_fn[0], param),
	154	'ch %d: %s \'%s\' (param = 0x%02x)' % \
	155	(chan, status_bytes[msg][1], ctrl_fn[1], param),
	156	'%d: %s \'%s\' is 0x%02x' % \
	157	(chan, status_bytes[msg][2], ctrl_fn[2], param)]])
	158
	159	def handle_channel_mode(self):
	160	# Channel Mode: Bn mm vv
	161	# n = channel, mm = mode number (120 - 127), vv = value
	162	c = self.cmd
	163	msg, chan, mm, vv = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	164	mode_fn = control_functions.get(mm, ('undefined', 'undef', 'undef'))
	165	# Decode the value based on the mode number.
	166	vv_string = ('', '')
	167	if mm == 122: # mode = local control?
	168	if vv == 0:
	169	vv_string = ('off', 'off')
	170	elif vv == 127: # mode = poly mode on?
	171	vv_string = ('on', 'on')
	172	else:
	173	vv_string = ('(non-standard param value of 0x%02x)' % vv,
	174	'0x%02x' % vv)
	175	elif mm == 126: # mode = mono mode on?
	176	if vv != 0:
	177	vv_string = ('(%d channels)' % vv, '(%d ch)' % vv)
	178	else:
	179	vv_string = ('(channels \'basic\' through 16)',
	180	'(ch \'basic\' thru 16)')
	181	elif vv != 0: # All other channel mode messages expect vv == 0.
	182	vv_string = ('(non-standard param value of 0x%02x)' % vv,
	183	'0x%02x' % vv)
	184	self.putx([0, ['Channel %d: %s \'%s\' %s' % \
	185	(chan, status_bytes[msg][0], mode_fn[0], vv_string[0]),
	186	'ch %d: %s \'%s\' %s' % \
	187	(chan, status_bytes[msg][1], mode_fn[1], vv_string[1]),
	188	'%d: %s \'%s\' %s' % \
	189	(chan, status_bytes[msg][2], mode_fn[2], vv_string[1])]])
	190	self.cmd, self.state = [], 'IDLE'
	191
	192	def handle_channel_msg_0xb0(self):
	193	# Control change (or channel mode messages): Bn cc vv
	194	# n = channel, cc = control number (0 - 119), vv = control value
	195	c = self.cmd
	196	if len(c) < 3:
	197	return
	198	self.es_block = self.es
	199	if c[1] in range(0x78, 0x7f + 1):
	200	self.handle_channel_mode()
	201	return
	202	handle_ctrl = getattr(self, 'handle_controller_0x%02x' % c[1],
	203	self.handle_controller_generic)
	204	handle_ctrl()
	205	self.cmd, self.state = [], 'IDLE'
	206
	207	def handle_channel_msg_0xc0(self):
	208	# Program change: Cn pp
	209	# n = channel, pp = program number (0 - 127)
	210	c = self.cmd
	211	if len(c) < 2:
	212	return
	213	self.es_block = self.es
	214	msg, chan, pp = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
	215	self.cmd[1] + 1
	216	change_type = 'instrument'
	217	name = ''
	218	if chan != 10: # channel != percussion
	219	name = gm_instruments.get(pp, 'undefined')
	220	else:
	221	change_type = 'drum kit'
	222	name = drum_kit.get(pp, 'undefined')
	223	self.putx([0, ['Channel %d: %s to %s %d (assuming %s)' % \
	224	(chan, status_bytes[msg][0], change_type, pp, name),
	225	'ch %d: %s to %s %d' % \
	226	(chan, status_bytes[msg][1], change_type, pp),
	227	'%d: %s %d' % \
	228	(chan, status_bytes[msg][2], pp)]])
	229	self.cmd, self.state = [], 'IDLE'
	230
	231	def handle_channel_msg_0xd0(self):
	232	# Channel pressure / aftertouch: Dn vv
	233	# n = channel, vv = pressure value
	234	c = self.cmd
	235	if len(c) < 2:
	236	return
	237	self.es_block = self.es
	238	msg, chan, vv = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
	239	self.cmd[1]
	240	self.putx([0, ['Channel %d: %s %d' % (chan, status_bytes[msg][0], vv),
	241	'ch %d: %s %d' % (chan, status_bytes[msg][1], vv),
	242	'%d: %s %d' % (chan, status_bytes[msg][2], vv)]])
	243	self.cmd, self.state = [], 'IDLE'
	244
	245	def handle_channel_msg_0xe0(self):
	246	# Pitch bend change: En ll mm
	247	# n = channel, ll = pitch bend change LSB, mm = pitch bend change MSB
	248	c = self.cmd
	249	if len(c) < 3:
	250	return
	251	self.es_block = self.es
	252	msg, chan, ll, mm = self.cmd[0] & 0xf0, (self.cmd[0] & 0x0f) + 1, \
	253	self.cmd[1], self.cmd[2]
	254	decimal = (mm << 7) + ll
	255	self.putx([0, ['Channel %d: %s 0x%02x 0x%02x (%d)' % \
	256	(chan, status_bytes[msg][0], ll, mm, decimal),
	257	'ch %d: %s 0x%02x 0x%02x (%d)' % \
	258	(chan, status_bytes[msg][1], ll, mm, decimal),
	259	'%d: %s (%d)' % \
	260	(chan, status_bytes[msg][2], decimal)]])
	261	self.cmd, self.state = [], 'IDLE'
	262
	263	def handle_channel_msg_generic(self):
	264	# TODO: It should not be possible to hit this code.
	265	# It currently can not be unit tested.
	266	msg_type = self.cmd[0] & 0xf0
	267	self.es_block = self.es
	268	self.putx([0, ['Unknown channel message type: 0x%02x' % msg_type]])
	269	self.cmd, self.state = [], 'IDLE'
	270
	271	def handle_channel_msg(self, newbyte):
	272	self.cmd.append(newbyte)
	273	msg_type = self.cmd[0] & 0xf0
	274	handle_msg = getattr(self, 'handle_channel_msg_0x%02x' % msg_type,
	275	self.handle_channel_msg_generic)
	276	handle_msg()
	277
	278	def handle_sysex_msg(self, newbyte):
	279	# SysEx message: 1 status byte, 1-3 manuf. bytes, x data bytes, EOX byte
	280	self.cmd.append(newbyte)
	281	if newbyte != 0xf7: # EOX
	282	return
	283	self.es_block = self.es
	284	# Note: Unlike other methods, this code pops bytes out of self.cmd
	285	# to isolate the data.
	286	msg, eox = self.cmd.pop(0), self.cmd.pop()
	287	if len(self.cmd) < 1:
	288	self.putx([0, ['%s: truncated manufacturer code (<1 bytes)' % \
	289	status_bytes[msg][0],
	290	'%s: truncated manufacturer (<1 bytes)' % \
	291	status_bytes[msg][1],
	292	'%s: trunc. manu.' % status_bytes[msg][2]]])
	293	self.cmd, self.state = [], 'IDLE'
	294	return
	295	# Extract the manufacturer name (or SysEx realtime or non-realtime).
	296	m1 = self.cmd.pop(0)
	297	manu = (m1,)
	298	if m1 == 0x00: # If byte == 0, then 2 more manufacturer bytes follow.
	299	if len(self.cmd) < 2:
	300	self.putx([0, ['%s: truncated manufacturer code (<3 bytes)' % \
	301	status_bytes[msg][0],
	302	'%s: truncated manufacturer (<3 bytes)' % \
	303	status_bytes[msg][1],
	304	'%s: trunc. manu.' % status_bytes[msg][2]]])
	305	self.cmd, self.state = [], 'IDLE'
	306	return
	307	manu = (m1, self.cmd.pop(0), self.cmd.pop(0))
	308	default_name = 'undefined'
	309	manu_name = sysex_manufacturer_ids.get(manu, default_name)
	310	if manu_name == default_name:
	311	if len(manu) == 3:
	312	manu_name = ('%s (0x%02x 0x%02x 0x%02x)' % \
	313	(default_name, manu[0], manu[1], manu[2]),
	314	default_name)
	315	else:
	316	manu_name = ('%s (0x%02x)' % (default_name, manu[0]),
	317	default_name)
	318	else:
	319	manu_name = (manu_name, manu_name)
	320	# Extract the payload, display in 1 of 2 formats
	321	# TODO: Write methods to decode SysEx realtime & non-realtime payloads.
	322	payload0 = ''
	323	payload1 = ''
	324	while len(self.cmd) > 0:
	325	byte = self.cmd.pop(0)
	326	payload0 += '0x%02x ' % (byte)
	327	payload1 += '%02x ' % (byte)
	328	if payload0 == '':
	329	payload0 = '<empty>'
	330	payload1 = '<>'
	331	payload = (payload0, payload1)
	332	self.putx([0, ['%s: for \'%s\' with payload %s' % \
	333	(status_bytes[msg][0], manu_name[0], payload[0]),
	334	'%s: \'%s\', payload %s' % \
	335	(status_bytes[msg][1], manu_name[1], payload[1]),
	336	'%s: \'%s\', payload %s' % \
	337	(status_bytes[msg][2], manu_name[1], payload[1])]])
	338	self.cmd, self.state = [], 'IDLE'
	339
	340	def handle_syscommon_midi_time_code_quarter_frame_msg(self, newbyte):
	341	# MIDI time code quarter frame: F1 nd
	342	# n = message type
	343	# d = values
	344	c = self.cmd
	345	if len(c) < 2:
	346	return
	347	msg = self.cmd[0]
	348	nn, dd = (self.cmd[1] & 0x70) >> 4, self.cmd[1] & 0x0f
	349	group = ('System Common', 'SysCom', 'SC')
	350	self.es_block = self.es
	351	if nn != 7: # If message type does not contain SMPTE type.
	352	self.putx([0, ['%s: %s of %s, value 0x%01x' % \
	353	(group[0], status_bytes[msg][0],
	354	quarter_frame_type[nn][0], dd),
	355	'%s: %s of %s, value 0x%01x' % \
	356	(group[1], status_bytes[msg][1],
	357	quarter_frame_type[nn][1], dd),
	358	'%s: %s of %s, value 0x%01x' % \
	359	(group[2], status_bytes[msg][2],
	360	quarter_frame_type[nn][1], dd)]])
	361	self.cmd, self.state = [], 'IDLE'
	362	return
	363	tt = (dd & 0x6) >> 1
	364	self.putx([0, ['%s: %s of %s, value 0x%01x for %s' % \
	365	(group[0], status_bytes[msg][0], \
	366	quarter_frame_type[nn][0], dd, smpte_type[tt]),
	367	'%s: %s of %s, value 0x%01x for %s' % \
	368	(group[1], status_bytes[msg][1], \
	369	quarter_frame_type[nn][1], dd, smpte_type[tt]),
	370	'%s: %s of %s, value 0x%01x for %s' % \
	371	(group[2], status_bytes[msg][2], \
	372	quarter_frame_type[nn][1], dd, smpte_type[tt])]])
	373	self.cmd, self.state = [], 'IDLE'
	374
	375	def handle_syscommon_msg(self, newbyte):
	376	# System common messages
	377	#
	378	# There are 5 simple formats (which are directly handled here) and
	379	# 1 complex one called MIDI time code quarter frame.
	380	#
	381	# Note: While the MIDI lists 0xf7 as a "system common" message, it
	382	# is actually only used with SysEx messages so it is processed there.
	383	self.cmd.append(newbyte)
	384	msg = self.cmd[0]
	385	c = self.cmd
	386	group = ('System Common', 'SysCom', 'SC')
	387	if msg == 0xf1:
	388	# MIDI time code quarter frame
	389	self.handle_syscommon_midi_time_code_quarter_frame_msg(newbyte)
	390	return
	391	elif msg == 0xf2:
	392	# Song position pointer: F2 ll mm
	393	# ll = LSB position, mm = MSB position
	394	if len(c) < 3:
	395	return
	396	ll, mm = self.cmd[1], self.cmd[2]
	397	decimal = (mm << 7) + ll
	398	self.es_block = self.es
	399	self.putx([0, ['%s: %s 0x%02x 0x%02x (%d)' % \
	400	(group[0], status_bytes[msg][0], ll, mm, decimal),
	401	'%s: %s 0x%02x 0x%02x (%d)' % \
	402	(group[1], status_bytes[msg][1], ll, mm, decimal),
	403	'%s: %s (%d)' % \
	404	(group[2], status_bytes[msg][2], decimal)]])
	405	elif msg == 0xf3:
	406	# Song select: F3 ss
	407	# ss = song selection number
	408	if len(c) < 2:
	409	return
	410	ss = self.cmd[1]
	411	self.es_block = self.es
	412	self.putx([0, ['%s: %s number %d' % \
	413	(group[0], status_bytes[msg][0], ss),
	414	'%s: %s number %d' % \
	415	(group[1], status_bytes[msg][1], ss),
	416	'%s: %s # %d' % \
	417	(group[2], status_bytes[msg][2], ss)]])
	418	elif msg == 0xf4 or msg == 0xf5 or msg == 0xf6:
	419	# Undefined 0xf4, Undefined 0xf5, and Tune Request (respectively).
	420	# All are only 1 byte long with no data bytes.
	421	self.es_block = self.es
	422	self.putx([0, ['%s: %s' % (group[0], status_bytes[msg][0]),
	423	'%s: %s' % (group[1], status_bytes[msg][1]),
	424	'%s: %s' % (group[2], status_bytes[msg][2])]])
	425	self.cmd, self.state = [], 'IDLE'
	426
	427	def handle_sysrealtime_msg(self, newbyte):
	428	# System realtime message: 0b11111ttt (t = message type)
	429	self.es_block = self.es
	430	group = ('System Realtime', 'SysReal', 'SR')
	431	self.putx([0, ['%s: %s' % (group[0], status_bytes[newbyte][0]),
	432	'%s: %s' % (group[1], status_bytes[newbyte][1]),
	433	'%s: %s' % (group[2], status_bytes[newbyte][2])]])
	434	self.cmd, self.state = [], 'IDLE'
	435
	436	def decode(self, ss, es, data):
	437	ptype, rxtx, pdata = data
	438
	439	# For now, ignore all UART packets except the actual data packets.
	440	if ptype != 'DATA':
	441	return
	442
	443	self.ss, self.es = ss, es
	444
	445	# We're only interested in the byte value (not individual bits).
	446	pdata = pdata[0]
	447
	448	# Short MIDI overview:
	449	# - Status bytes are 0x80-0xff, data bytes are 0x00-0x7f.
	450	# - Most messages: 1 status byte, 1-2 data bytes.
	451	# - Real-time system messages: always 1 byte.
	452	# - SysEx messages: 1 status byte, n data bytes, EOX byte.
	453
	454	# State machine.
	455	if self.state == 'IDLE':
	456	# Wait until we see a status byte (bit 7 must be set).
	457	if pdata < 0x80:
	458	return # TODO: How to handle? Ignore?
	459	# This is a status byte, remember the start sample.
	460	self.ss_block = ss
	461	if pdata in range(0x80, 0xef + 1):
	462	self.state = 'HANDLE CHANNEL MSG'
	463	elif pdata == 0xf0 or pdata == 0xf7:
	464	self.state = 'HANDLE SYSEX MSG'
	465	elif pdata in range(0xf1, 0xf7):
	466	self.state = 'HANDLE SYSCOMMON MSG'
	467	elif pdata in range(0xf8, 0xff + 1):
	468	self.state = 'HANDLE SYSREALTIME MSG'
	469
	470	# Yes, this is intentionally _not_ an 'elif' here.
	471	if self.state == 'HANDLE CHANNEL MSG':
	472	self.handle_channel_msg(pdata)
	473	elif self.state == 'HANDLE SYSEX MSG':
	474	self.handle_sysex_msg(pdata)
	475	elif self.state == 'HANDLE SYSCOMMON MSG':
	476	self.handle_syscommon_msg(pdata)
	477	elif self.state == 'HANDLE SYSREALTIME MSG':
	478	self.handle_sysrealtime_msg(pdata)