[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
##

# MIDI protocol decoder

import sigrokdecode as srd
from .lists import *

RX = 0
TX = 1

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

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

    def start(self):
        # self.out_proto = self.add(srd.OUTPUT_PROTO, 'midi')
        self.out_ann = self.add(srd.OUTPUT_ANN, 'midi')

    def report(self):
        pass

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

    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]
        self.putx([0, ['Channel %d: %s (note = %d, velocity = %d)' % \
                  (chan, status_bytes[msg], 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.ss
        msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
        s = 'note off' if (velocity == 0) else status_bytes[msg]
        self.putx([0, ['Channel %d: %s (note = %d, velocity = %d)' % \
                  (chan, s, 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
        pass # TODO

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

    def handle_controller_0x54(self):
        # Portamento control (PTC): Bn 54 kk
        # n = channel, kk = source note for pitch reference
        chan, kk = (self.cmd[0] & 0x0f) + 1, self.cmd[2]
        self.putx([0, ['Channel %d: control function \'%s\' (source note ' \
                  '= %d)' % (chan, control_functions[0x54], kk)]])

    def handle_controller_generic(self):
        c = self.cmd
        chan, fn, param = (c[0] & 0x0f) + 1, c[1], c[2]
        ctrl_fn = control_functions.get(fn, 'undefined')
        self.putx([0, ['Channel %d: control change to function \'%s\' ' \
                  '(param = 0x%02x)' % (chan, ctrl_fn, param)]])

    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) >= 2) and (c[1] in range(0x78, 0x7f + 1)):
            # This is not a control change, but rather a channel mode message.
            # TODO: Handle channel mode messages.
            return
        if len(c) < 3:
            return
        self.es_block = self.es
        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)
        pass # TODO

    def handle_channel_msg_0xd0(self):
        # Channel pressure / aftertouch: Dn vv
        # n = channel, vv = pressure value
        pass # TODO

    def handle_channel_msg_0xe0(self):
        # Pitch bend change: En ll mm
        # n = channel, ll = pitch bend change LSB, mm = pitch bend change MSB
        pass # TODO

    def handle_channel_msg_generic(self):
        msg_type = self.cmd[0] & 0xf0
        self.putx([0, ['Unknown channel message type: 0x%02x' % msg_type]])
        # TODO: Handle properly.

    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, x data bytes, EOX byte
        self.cmd.append(newbyte)
        if newbyte != 0xf7: # EOX
            return
        self.es_block = self.es
        # TODO: Get message ID, vendor ID, message contents, etc.
        self.putx([0, ['SysEx message']])
        self.cmd, self.state = [], 'IDLE'

    def handle_syscommon_msg(self, newbyte):
        pass # TODO

    def handle_sysrealtime_msg(self, newbyte):
        # System realtime message: 0b11111ttt (t = message type)
        self.es_block = self.ss
        self.putx([0, ['System realtime message: %s' % status_bytes[newbyte]]])
        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

        # 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:
                self.state = 'HANDLE SYSEX MSG'
            elif pdata in range(0xf1, 0xf7 + 1):
                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)
        else:
            raise Exception('Invalid state: %s' % self.state)
Commit	Line	Data
17f5df4f UH	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	# MIDI protocol decoder
	22
	23	import sigrokdecode as srd
	24	from .lists import *
	25
	26	RX = 0
	27	TX = 1
	28
	29	class Decoder(srd.Decoder):
	30	api_version = 1
	31	id = 'midi'
	32	name = 'MIDI'
	33	longname = 'Musical Instrument Digital Interface'
	34	desc = 'Musical Instrument Digital Interface (MIDI) protocol.'
	35	license = 'gplv2+'
	36	inputs = ['uart']
	37	outputs = ['midi']
	38	probes = []
	39	optional_probes = []
	40	options = {}
	41	annotations = [
	42	['Text (verbose)', 'Human-readable text (verbose)'],
	43	# ['Text', 'Human-readable text'],
	44	]
	45
	46	def __init__(self, **kwargs):
	47	self.cmd = []
	48	self.state = 'IDLE'
	49	self.ss = None
	50	self.es = None
	51	self.ss_block = None
	52	self.es_block = None
	53
8915b346	54	def start(self):
17f5df4f UH	55	# self.out_proto = self.add(srd.OUTPUT_PROTO, 'midi')
	56	self.out_ann = self.add(srd.OUTPUT_ANN, 'midi')
	57
	58	def report(self):
	59	pass
	60
	61	def putx(self, data):
	62	self.put(self.ss_block, self.es_block, self.out_ann, data)
	63
	64	def handle_channel_msg_0x80(self):
	65	# Note off: 8n kk vv
	66	# n = channel, kk = note, vv = velocity
	67	c = self.cmd
	68	if len(c) < 3:
	69	return
	70	self.es_block = self.es
	71	msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	72	self.putx([0, ['Channel %d: %s (note = %d, velocity = %d)' % \
	73	(chan, status_bytes[msg], note, velocity)]])
	74	self.cmd, self.state = [], 'IDLE'
	75
	76	def handle_channel_msg_0x90(self):
	77	# Note on: 9n kk vv
	78	# n = channel, kk = note, vv = velocity
	79	# If velocity == 0 that actually means 'note off', though.
	80	c = self.cmd
	81	if len(c) < 3:
	82	return
	83	self.es_block = self.ss
	84	msg, chan, note, velocity = c[0] & 0xf0, (c[0] & 0x0f) + 1, c[1], c[2]
	85	s = 'note off' if (velocity == 0) else status_bytes[msg]
	86	self.putx([0, ['Channel %d: %s (note = %d, velocity = %d)' % \
	87	(chan, s, note, velocity)]])
	88	self.cmd, self.state = [], 'IDLE'
	89
	90	def handle_channel_msg_0xa0(self):
	91	# Polyphonic key pressure / aftertouch: An kk vv
	92	# n = channel, kk = polyphonic key pressure, vv = pressure value
	93	pass # TODO
	94
	95	def handle_controller_0x44(self):
	96	# Legato footswitch: Bn 44 vv
	97	# n = channel, vv = value (<= 0x3f: normal, > 0x3f: legato)
	98	chan, vv = (self.cmd[0] & 0x0f) + 1, self.cmd[2]
	99	t = 'normal' if vv <= 0x3f else 'legato'
	100	self.putx([0, ['Channel %d: control function \'%s\' = %s' % \
	101	(chan, control_functions[0x44], t)]])
	102
	103	def handle_controller_0x54(self):
	104	# Portamento control (PTC): Bn 54 kk
	105	# n = channel, kk = source note for pitch reference
	106	chan, kk = (self.cmd[0] & 0x0f) + 1, self.cmd[2]
	107	self.putx([0, ['Channel %d: control function \'%s\' (source note ' \
	108	'= %d)' % (chan, control_functions[0x54], kk)]])
	109
	110	def handle_controller_generic(self):
	111	c = self.cmd
	112	chan, fn, param = (c[0] & 0x0f) + 1, c[1], c[2]
	113	ctrl_fn = control_functions.get(fn, 'undefined')
	114	self.putx([0, ['Channel %d: control change to function \'%s\' ' \
	115	'(param = 0x%02x)' % (chan, ctrl_fn, param)]])
	116
	117	def handle_channel_msg_0xb0(self):
	118	# Control change (or channel mode messages): Bn cc vv
119	# n = channel, cc = control number (0 - 119), vv = control value
120	c = self.cmd
121	if (len(c) >= 2) and (c[1] in range(0x78, 0x7f + 1)):
122	# This is not a control change, but rather a channel mode message.
123	# TODO: Handle channel mode messages.
124	return
125	if len(c) < 3:
126	return
127	self.es_block = self.es
128	handle_ctrl = getattr(self, 'handle_controller_0x%02x' % c[1],
129	self.handle_controller_generic)
130	handle_ctrl()
131	self.cmd, self.state = [], 'IDLE'
132
133	def handle_channel_msg_0xc0(self):
134	# Program change: Cn pp
135	# n = channel, pp = program number (0 - 127)
136	pass # TODO
137
138	def handle_channel_msg_0xd0(self):
139	# Channel pressure / aftertouch: Dn vv
140	# n = channel, vv = pressure value
141	pass # TODO
142
143	def handle_channel_msg_0xe0(self):
144	# Pitch bend change: En ll mm
145	# n = channel, ll = pitch bend change LSB, mm = pitch bend change MSB
146	pass # TODO
147
148	def handle_channel_msg_generic(self):
149	msg_type = self.cmd[0] & 0xf0
150	self.putx([0, ['Unknown channel message type: 0x%02x' % msg_type]])
151	# TODO: Handle properly.
152
153	def handle_channel_msg(self, newbyte):
154	self.cmd.append(newbyte)
155	msg_type = self.cmd[0] & 0xf0
156	handle_msg = getattr(self, 'handle_channel_msg_0x%02x' % msg_type,
157	self.handle_channel_msg_generic)
158	handle_msg()
159
160	def handle_sysex_msg(self, newbyte):
161	# SysEx message: 1 status byte, x data bytes, EOX byte
162	self.cmd.append(newbyte)
163	if newbyte != 0xf7: # EOX
164	return
165	self.es_block = self.es
166	# TODO: Get message ID, vendor ID, message contents, etc.
167	self.putx([0, ['SysEx message']])
168	self.cmd, self.state = [], 'IDLE'
169
170	def handle_syscommon_msg(self, newbyte):
171	pass # TODO
172
173	def handle_sysrealtime_msg(self, newbyte):
174	# System realtime message: 0b11111ttt (t = message type)
175	self.es_block = self.ss
176	self.putx([0, ['System realtime message: %s' % status_bytes[newbyte]]])
177	self.cmd, self.state = [], 'IDLE'
178
179	def decode(self, ss, es, data):
180	ptype, rxtx, pdata = data
181
182	# For now, ignore all UART packets except the actual data packets.
183	if ptype != 'DATA':
184	return
185
186	self.ss, self.es = ss, es
187
188	# Short MIDI overview:
189	# - Status bytes are 0x80-0xff, data bytes are 0x00-0x7f.
190	# - Most messages: 1 status byte, 1-2 data bytes.
191	# - Real-time system messages: always 1 byte.
192	# - SysEx messages: 1 status byte, n data bytes, EOX byte.
193
194	# State machine.
195	if self.state == 'IDLE':
196	# Wait until we see a status byte (bit 7 must be set).
197	if pdata < 0x80:
198	return # TODO: How to handle? Ignore?
199	# This is a status byte, remember the start sample.
200	self.ss_block = ss
201	if pdata in range(0x80, 0xef + 1):
202	self.state = 'HANDLE CHANNEL MSG'
203	elif pdata == 0xf0:
204	self.state = 'HANDLE SYSEX MSG'
205	elif pdata in range(0xf1, 0xf7 + 1):
206	self.state = 'HANDLE SYSCOMMON MSG'
207	elif pdata in range(0xf8, 0xff + 1):
208	self.state = 'HANDLE SYSREALTIME MSG'
209
210	# Yes, this is intentionally _not_ an 'elif' here.
211	if self.state == 'HANDLE CHANNEL MSG':
212	self.handle_channel_msg(pdata)
213	elif self.state == 'HANDLE SYSEX MSG':
214	self.handle_sysex_msg(pdata)
215	elif self.state == 'HANDLE SYSCOMMON MSG':
216	self.handle_syscommon_msg(pdata)
217	elif self.state == 'HANDLE SYSREALTIME MSG':
218	self.handle_sysrealtime_msg(pdata)
219	else:
220	raise Exception('Invalid state: %s' % self.state)
221