decoders/midi/pd.py

   1 ##
   2 ## This file is part of the libsigrokdecode project.
   3 ##
   4 ## Copyright (C) 2013-2016 Uwe Hermann <uwe@hermann-uwe.de>
   5 ## Copyright (C) 2016 Chris Dreher <chrisdreher@hotmail.com>
   6 ##
   7 ## This program is free software; you can redistribute it and/or modify
   8 ## it under the terms of the GNU General Public License as published by
   9 ## the Free Software Foundation; either version 2 of the License, or
  10 ## (at your option) any later version.
  11 ##
  12 ## This program is distributed in the hope that it will be useful,
  13 ## but WITHOUT ANY WARRANTY; without even the implied warranty of
  14 ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  15 ## GNU General Public License for more details.
  16 ##
  17 ## You should have received a copy of the GNU General Public License
  18 ## along with this program; if not, see <http://www.gnu.org/licenses/>.
  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 = 3
  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         ('text-sysreal-verbose', 'Human-readable SysReal text (verbose)'),
  39         ('text-error', 'Human-readable Error text'),
  40     )
  41     annotation_rows = (
  42         ('normal', 'Normal', (0, 2)),
  43         ('sys-real', 'SysReal', (1,)),
  44     )
  45
  46     def __init__(self):
  47         self.reset()
  48
  49     def reset(self):
  50         self.state = 'IDLE'
  51         self.status_byte = 0
  52         self.explicit_status_byte = False
  53         self.cmd = []
  54         self.ss = None
  55         self.es = None
  56         self.ss_block = None
  57         self.es_block = None
  58
  59     def start(self):
  60         self.out_ann = self.register(srd.OUTPUT_ANN)
  61
  62     def putx(self, data):
  63         self.put(self.ss_block, self.es_block, self.out_ann, data)
  64
  65     def get_note_name(self, channel, note):
  66         if channel != 10:
  67             return chromatic_notes[note]
  68         else:
  69             return 'assuming ' + percussion_notes.get(note, 'undefined')
  70
  71     def check_for_garbage_flush(self, is_flushed):
  72         if is_flushed:
  73             if self.explicit_status_byte:
  74                 self.cmd.insert(0, self.status_byte)
  75             self.handle_garbage_msg(None)
  76
  77     def soft_clear_status_byte(self):
  78         self.explicit_status_byte = False
  79
  80     def hard_clear_status_byte(self):
  81         self.status_byte = 0
  82         self.explicit_status_byte = False
  83
  84     def set_status_byte(self, newbyte):
  85         self.status_byte = newbyte
  86         self.explicit_status_byte = True
  87
  88     def handle_channel_msg_0x80(self, is_flushed):
  89         # Note off: 8n kk vv
  90         # n = channel, kk = note, vv = velocity
  91         c = self.cmd
  92         if len(c) < 2:
  93             self.check_for_garbage_flush(is_flushed)
  94             return
  95         self.es_block = self.es
  96         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
  97         note, velocity = c[0], c[1]
  98         note_name = self.get_note_name(chan, note)
  99         self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
 100                   (chan, status_bytes[msg][0], note, note_name, velocity),
 101                   'ch %d: %s %d, velocity = %d' % \
 102                   (chan, status_bytes[msg][1], note, velocity),
 103                   '%d: %s %d, vel %d' % \
 104                   (chan, status_bytes[msg][2], note, velocity)]])
 105         self.cmd, self.state = [], 'IDLE'
 106         self.soft_clear_status_byte()
 107
 108     def handle_channel_msg_0x90(self, is_flushed):
 109         # Note on: 9n kk vv
 110         # n = channel, kk = note, vv = velocity
 111         # If velocity == 0 that actually means 'note off', though.
 112         c = self.cmd
 113         if len(c) < 2:
 114             self.check_for_garbage_flush(is_flushed)
 115             return
 116         self.es_block = self.es
 117         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 118         note, velocity = c[0], c[1]
 119         s = status_bytes[0x80] if (velocity == 0) else status_bytes[msg]
 120         note_name = self.get_note_name(chan, note)
 121         self.putx([0, ['Channel %d: %s (note = %d \'%s\', velocity = %d)' % \
 122                   (chan, s[0], note, note_name, velocity),
 123                   'ch %d: %s %d, velocity = %d' % \
 124                   (chan, s[1], note, velocity),
 125                   '%d: %s %d, vel %d' % \
 126                   (chan, s[2], note, velocity)]])
 127         self.cmd, self.state = [], 'IDLE'
 128         self.soft_clear_status_byte()
 129
 130     def handle_channel_msg_0xa0(self, is_flushed):
 131         # Polyphonic key pressure / aftertouch: An kk vv
 132         # n = channel, kk = polyphonic key pressure, vv = pressure value
 133         c = self.cmd
 134         if len(c) < 2:
 135             self.check_for_garbage_flush(is_flushed)
 136             return
 137         self.es_block = self.es
 138         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 139         note, pressure = c[0], c[1]
 140         note_name = self.get_note_name(chan, note)
 141         self.putx([0, ['Channel %d: %s of %d for note = %d \'%s\'' % \
 142                   (chan, status_bytes[msg][0], pressure, note, note_name),
 143                   'ch %d: %s %d for note %d' % \
 144                   (chan, status_bytes[msg][1], pressure, note),
 145                   '%d: %s %d, N %d' % \
 146                   (chan, status_bytes[msg][2], pressure, note)]])
 147         self.cmd, self.state = [], 'IDLE'
 148         self.soft_clear_status_byte()
 149
 150     def handle_controller_0x44(self):
 151         # Legato footswitch: Bn 44 vv
 152         # n = channel, vv = value (<= 0x3f: normal, > 0x3f: legato)
 153         c = self.cmd
 154         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 155         vv = c[1]
 156         t = ('normal', 'no') if vv <= 0x3f else ('legato', 'yes')
 157         self.putx([0, ['Channel %d: %s \'%s\' = %s' % \
 158                   (chan, status_bytes[msg][0],
 159                   control_functions[0x44][0], t[0]),
 160                   'ch %d: %s \'%s\' = %s' % \
 161                   (chan, status_bytes[msg][1],
 162                   control_functions[0x44][1], t[0]),
 163                   '%d: %s \'%s\' = %s' % \
 164                   (chan, status_bytes[msg][2],
 165                   control_functions[0x44][2], t[1])]])
 166
 167     def handle_controller_0x54(self):
 168         # Portamento control (PTC): Bn 54 kk
 169         # n = channel, kk = source note for pitch reference
 170         c = self.cmd
 171         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 172         kk = c[1]
 173         kk_name = self.get_note_name(chan, kk)
 174         self.putx([0, ['Channel %d: %s \'%s\' (source note = %d / %s)' % \
 175                   (chan, status_bytes[msg][0],
 176                   control_functions[0x54][0], kk, kk_name),
 177                   'ch %d: %s \'%s\' (source note = %d)' % \
 178                   (chan, status_bytes[msg][1],
 179                   control_functions[0x54][1], kk),
 180                   '%d: %s \'%s\' (src N %d)' % \
 181                   (chan, status_bytes[msg][2],
 182                   control_functions[0x54][2], kk)]])
 183
 184     def handle_controller_generic(self):
 185         c = self.cmd
 186         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 187         fn, param = c[0], c[1]
 188         default_name = 'undefined'
 189         ctrl_fn = control_functions.get(fn, default_name)
 190         if ctrl_fn == default_name:
 191             ctrl_fn = ('undefined 0x%02x' % fn, 'undef 0x%02x' % fn, '0x%02x' % fn)
 192         self.putx([0, ['Channel %d: %s \'%s\' (param = 0x%02x)' % \
 193                   (chan, status_bytes[msg][0], ctrl_fn[0], param),
 194                   'ch %d: %s \'%s\' (param = 0x%02x)' % \
 195                   (chan, status_bytes[msg][1], ctrl_fn[1], param),
 196                   '%d: %s \'%s\' is 0x%02x' % \
 197                   (chan, status_bytes[msg][2], ctrl_fn[2], param)]])
 198
 199     def handle_channel_mode(self):
 200         # Channel Mode: Bn mm vv
 201         # n = channel, mm = mode number (120 - 127), vv = value
 202         c = self.cmd
 203         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 204         mm, vv = c[0], c[1]
 205         mode_fn = control_functions.get(mm, ('undefined', 'undef', 'undef'))
 206         # Decode the value based on the mode number.
 207         vv_string = ('', '')
 208         if mm == 122:           # mode = local control?
 209             if vv == 0:
 210                 vv_string = ('off', 'off')
 211             elif vv == 127:     # mode = poly mode on?
 212                 vv_string = ('on', 'on')
 213             else:
 214                 vv_string = ('(non-standard param value of 0x%02x)' % vv,
 215                             '0x%02x' % vv)
 216         elif mm == 126:         # mode = mono mode on?
 217             if vv != 0:
 218                 vv_string = ('(%d channels)' % vv, '(%d ch)' % vv)
 219             else:
 220                 vv_string = ('(channels \'basic\' through 16)',
 221                             '(ch \'basic\' thru 16)')
 222         elif vv != 0: # All other channel mode messages expect vv == 0.
 223             vv_string = ('(non-standard param value of 0x%02x)' % vv,
 224                         '0x%02x' % vv)
 225         self.putx([0, ['Channel %d: %s \'%s\' %s' % \
 226                       (chan, status_bytes[msg][0], mode_fn[0], vv_string[0]),
 227                       'ch %d: %s \'%s\' %s' % \
 228                       (chan, status_bytes[msg][1], mode_fn[1], vv_string[1]),
 229                       '%d: %s \'%s\' %s' % \
 230                       (chan, status_bytes[msg][2], mode_fn[2], vv_string[1])]])
 231         self.cmd, self.state = [], 'IDLE'
 232         self.soft_clear_status_byte()
 233
 234     def handle_channel_msg_0xb0(self, is_flushed):
 235         # Control change (or channel mode messages): Bn cc vv
 236         # n = channel, cc = control number (0 - 119), vv = control value
 237         c = self.cmd
 238         if len(c) < 2:
 239             self.check_for_garbage_flush(is_flushed)
 240             return
 241         self.es_block = self.es
 242         if c[0] in range(0x78, 0x7f + 1):
 243             self.handle_channel_mode()
 244             return
 245         handle_ctrl = getattr(self, 'handle_controller_0x%02x' % c[0],
 246                               self.handle_controller_generic)
 247         handle_ctrl()
 248         self.cmd, self.state = [], 'IDLE'
 249         self.soft_clear_status_byte()
 250
 251     def handle_channel_msg_0xc0(self, is_flushed):
 252         # Program change: Cn pp
 253         # n = channel, pp = program number (0 - 127)
 254         c = self.cmd
 255         if len(c) < 1:
 256             self.check_for_garbage_flush(is_flushed)
 257             return
 258         self.es_block = self.es
 259         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 260         pp = self.cmd[0] + 1
 261         change_type = 'instrument'
 262         name = ''
 263         if chan != 10:  # channel != percussion
 264             name = gm_instruments.get(pp, 'undefined')
 265         else:
 266             change_type = 'drum kit'
 267             name = drum_kit.get(pp, 'undefined')
 268         self.putx([0, ['Channel %d: %s to %s %d (assuming %s)' % \
 269             (chan, status_bytes[msg][0], change_type, pp, name),
 270             'ch %d: %s to %s %d' % \
 271             (chan, status_bytes[msg][1], change_type, pp),
 272             '%d: %s %d' % \
 273             (chan, status_bytes[msg][2], pp)]])
 274         self.cmd, self.state = [], 'IDLE'
 275         self.soft_clear_status_byte()
 276
 277     def handle_channel_msg_0xd0(self, is_flushed):
 278         # Channel pressure / aftertouch: Dn vv
 279         # n = channel, vv = pressure value
 280         c = self.cmd
 281         if len(c) < 1:
 282             self.check_for_garbage_flush(is_flushed)
 283             return
 284         self.es_block = self.es
 285         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 286         vv = self.cmd[0]
 287         self.putx([0, ['Channel %d: %s %d' % (chan, status_bytes[msg][0], vv),
 288                       'ch %d: %s %d' % (chan, status_bytes[msg][1], vv),
 289                       '%d: %s %d' % (chan, status_bytes[msg][2], vv)]])
 290         self.cmd, self.state = [], 'IDLE'
 291         self.soft_clear_status_byte()
 292
 293     def handle_channel_msg_0xe0(self, is_flushed):
 294         # Pitch bend change: En ll mm
 295         # n = channel, ll = pitch bend change LSB, mm = pitch bend change MSB
 296         c = self.cmd
 297         if len(c) < 2:
 298             self.check_for_garbage_flush(is_flushed)
 299             return
 300         self.es_block = self.es
 301         msg, chan = self.status_byte & 0xf0, (self.status_byte & 0x0f) + 1
 302         ll, mm = self.cmd[0], self.cmd[1]
 303         decimal = (mm << 7) + ll
 304         self.putx([0, ['Channel %d: %s 0x%02x 0x%02x (%d)' % \
 305                       (chan, status_bytes[msg][0], ll, mm, decimal),
 306                       'ch %d: %s 0x%02x 0x%02x (%d)' % \
 307                       (chan, status_bytes[msg][1], ll, mm, decimal),
 308                       '%d: %s (%d)' % \
 309                       (chan, status_bytes[msg][2], decimal)]])
 310         self.cmd, self.state = [], 'IDLE'
 311         self.soft_clear_status_byte()
 312
 313     def handle_channel_msg_generic(self, is_flushed):
 314         # TODO: It should not be possible to hit this code.
 315         # It currently can not be unit tested.
 316         msg_type = self.status_byte & 0xf0
 317         self.es_block = self.es
 318         self.putx([2, ['Unknown channel message type: 0x%02x' % msg_type]])
 319         self.cmd, self.state = [], 'IDLE'
 320         self.soft_clear_status_byte()
 321
 322     def handle_channel_msg(self, newbyte):
 323         if newbyte is not None:
 324             if newbyte >= 0x80:
 325                 self.set_status_byte(newbyte)
 326             else:
 327                 self.cmd.append(newbyte)
 328         msg_type = self.status_byte & 0xf0
 329         handle_msg = getattr(self, 'handle_channel_msg_0x%02x' % msg_type,
 330                              self.handle_channel_msg_generic)
 331         handle_msg(newbyte is None)
 332
 333     def handle_sysex_msg(self, newbyte):
 334         # SysEx message: 1 status byte, 1-3 manuf. bytes, x data bytes, EOX byte
 335         #
 336         # SysEx messages are variable length, can be terminated by EOX or
 337         # by any non-SysReal status byte, and it clears self.status_byte.
 338         #
 339         # Note: All System message codes don't utilize self.status_byte.
 340         self.hard_clear_status_byte()
 341         if newbyte != 0xf7 and newbyte is not None: # EOX
 342             self.cmd.append(newbyte)
 343             return
 344         self.es_block = self.es
 345         # Note: Unlike other methods, this code pops bytes out of self.cmd
 346         # to isolate the data.
 347         msg = self.cmd.pop(0)
 348         if len(self.cmd) < 1:
 349             self.putx([2, ['%s: truncated manufacturer code (<1 bytes)' % \
 350                           status_bytes[msg][0],
 351                           '%s: truncated manufacturer (<1 bytes)' % \
 352                           status_bytes[msg][1],
 353                           '%s: trunc. manu.' % status_bytes[msg][2]]])
 354             self.cmd, self.state = [], 'IDLE'
 355             return
 356         # Extract the manufacturer name (or SysEx realtime or non-realtime).
 357         m1 = self.cmd.pop(0)
 358         manu = (m1,)
 359         if m1 == 0x00:  # If byte == 0, then 2 more manufacturer bytes follow.
 360             if len(self.cmd) < 2:
 361                 self.putx([2, ['%s: truncated manufacturer code (<3 bytes)' % \
 362                           status_bytes[msg][0],
 363                           '%s: truncated manufacturer (<3 bytes)' % \
 364                           status_bytes[msg][1],
 365                           '%s: trunc. manu.' % status_bytes[msg][2]]])
 366                 self.cmd, self.state = [], 'IDLE'
 367                 return
 368             manu = (m1, self.cmd.pop(0), self.cmd.pop(0))
 369         default_name = 'undefined'
 370         manu_name = sysex_manufacturer_ids.get(manu, default_name)
 371         if manu_name == default_name:
 372             if len(manu) == 3:
 373                 manu_name = ('%s (0x%02x 0x%02x 0x%02x)' % \
 374                             (default_name, manu[0], manu[1], manu[2]),
 375                             default_name)
 376             else:
 377                 manu_name = ('%s (0x%02x)' % (default_name, manu[0]),
 378                             default_name)
 379         else:
 380             manu_name = (manu_name, manu_name)
 381         # Extract the payload, display in 1 of 2 formats
 382         # TODO: Write methods to decode SysEx realtime & non-realtime payloads.
 383         payload0 = ''
 384         payload1 = ''
 385         while len(self.cmd) > 0:
 386             byte = self.cmd.pop(0)
 387             payload0 += '0x%02x ' % (byte)
 388             payload1 += '%02x ' % (byte)
 389         if payload0 == '':
 390             payload0 = '<empty>'
 391             payload1 = '<>'
 392         payload = (payload0, payload1)
 393         self.putx([0, ['%s: for \'%s\' with payload %s' % \
 394                       (status_bytes[msg][0], manu_name[0], payload[0]),
 395                       '%s: \'%s\', payload %s' % \
 396                       (status_bytes[msg][1], manu_name[1], payload[1]),
 397                       '%s: \'%s\', payload %s' % \
 398                       (status_bytes[msg][2], manu_name[1], payload[1])]])
 399         self.cmd, self.state = [], 'IDLE'
 400
 401     def handle_syscommon_midi_time_code_quarter_frame_msg(self, newbyte):
 402         # MIDI time code quarter frame: F1 nd
 403         # n = message type
 404         # d = values
 405         #
 406         # Note: All System message codes don't utilize self.status_byte,
 407         # and System Exclusive and System Common clear it.
 408         c = self.cmd
 409         if len(c) < 2:
 410             if newbyte is None:
 411                 self.handle_garbage_msg(None)
 412             return
 413         msg = c[0]
 414         nn, dd = (c[1] & 0x70) >> 4, c[1] & 0x0f
 415         group = ('System Common', 'SysCom', 'SC')
 416         self.es_block = self.es
 417         if nn != 7: # If message type does not contain SMPTE type.
 418             self.putx([0, ['%s: %s of %s, value 0x%01x' % \
 419                           (group[0], status_bytes[msg][0],
 420                           quarter_frame_type[nn][0], dd),
 421                           '%s: %s of %s, value 0x%01x' % \
 422                           (group[1], status_bytes[msg][1],
 423                           quarter_frame_type[nn][1], dd),
 424                           '%s: %s of %s, value 0x%01x' % \
 425                           (group[2], status_bytes[msg][2],
 426                           quarter_frame_type[nn][1], dd)]])
 427             self.cmd, self.state = [], 'IDLE'
 428             return
 429         tt = (dd & 0x6) >> 1
 430         self.putx([0, ['%s: %s of %s, value 0x%01x for %s' % \
 431                       (group[0], status_bytes[msg][0], \
 432                       quarter_frame_type[nn][0], dd, smpte_type[tt]),
 433                       '%s: %s of %s, value 0x%01x for %s' % \
 434                       (group[1], status_bytes[msg][1], \
 435                       quarter_frame_type[nn][1], dd, smpte_type[tt]),
 436                       '%s: %s of %s, value 0x%01x for %s' % \
 437                       (group[2], status_bytes[msg][2], \
 438                       quarter_frame_type[nn][1], dd, smpte_type[tt])]])
 439         self.cmd, self.state = [], 'IDLE'
 440
 441     def handle_syscommon_msg(self, newbyte):
 442         # System common messages
 443         #
 444         # There are 5 simple formats (which are directly handled here) and
 445         # 1 complex one called MIDI time code quarter frame.
 446         #
 447         # Note: While the MIDI lists 0xf7 as a "system common" message, it
 448         # is actually only used with SysEx messages so it is processed there.
 449         #
 450         # Note: All System message codes don't utilize self.status_byte.
 451         self.hard_clear_status_byte()
 452         if newbyte is not None:
 453             self.cmd.append(newbyte)
 454         c = self.cmd
 455         msg = c[0]
 456         group = ('System Common', 'SysCom', 'SC')
 457         if msg == 0xf1:
 458             # MIDI time code quarter frame
 459             self.handle_syscommon_midi_time_code_quarter_frame_msg(newbyte)
 460             return
 461         elif msg == 0xf2:
 462             # Song position pointer: F2 ll mm
 463             # ll = LSB position, mm = MSB position
 464             if len(c) < 3:
 465                 if newbyte is None:
 466                     self.handle_garbage_msg(None)
 467                 return
 468             ll, mm = c[1], c[2]
 469             decimal = (mm << 7) + ll
 470             self.es_block = self.es
 471             self.putx([0, ['%s: %s 0x%02x 0x%02x (%d)' % \
 472                           (group[0], status_bytes[msg][0], ll, mm, decimal),
 473                           '%s: %s 0x%02x 0x%02x (%d)' % \
 474                           (group[1], status_bytes[msg][1], ll, mm, decimal),
 475                           '%s: %s (%d)' % \
 476                           (group[2], status_bytes[msg][2], decimal)]])
 477         elif msg == 0xf3:
 478             # Song select: F3 ss
 479             # ss = song selection number
 480             if len(c) < 2:
 481                 if newbyte is None:
 482                     self.handle_garbage_msg(None)
 483                 return
 484             ss = c[1]
 485             self.es_block = self.es
 486             self.putx([0, ['%s: %s number %d' % \
 487                           (group[0], status_bytes[msg][0], ss),
 488                           '%s: %s number %d' % \
 489                           (group[1], status_bytes[msg][1], ss),
 490                           '%s: %s # %d' % \
 491                           (group[2], status_bytes[msg][2], ss)]])
 492         elif msg == 0xf4 or msg == 0xf5 or msg == 0xf6:
 493             # Undefined 0xf4, Undefined 0xf5, and Tune Request (respectively).
 494             # All are only 1 byte long with no data bytes.
 495             self.es_block = self.es
 496             self.putx([0, ['%s: %s' % (group[0], status_bytes[msg][0]),
 497                           '%s: %s' % (group[1], status_bytes[msg][1]),
 498                           '%s: %s' % (group[2], status_bytes[msg][2])]])
 499         self.cmd, self.state = [], 'IDLE'
 500
 501     def handle_sysrealtime_msg(self, newbyte):
 502         # System realtime message: 0b11111ttt (t = message type)
 503         #
 504         # Important: These messages are handled differently from all others
 505         # because they are allowed to temporarily interrupt other messages.
 506         # The interrupted messages resume after the realtime message is done.
 507         # Thus, they mostly leave 'self' the way it was found.
 508         #
 509         # Note: All System message codes don't utilize self.status_byte.
 510         old_ss_block, old_es_block = self.ss_block, self.es_block
 511         self.ss_block, self.es_block = self.ss, self.es
 512         group = ('System Realtime', 'SysReal', 'SR')
 513         self.putx([1, ['%s: %s' % (group[0], status_bytes[newbyte][0]),
 514                       '%s: %s' % (group[1], status_bytes[newbyte][1]),
 515                       '%s: %s' % (group[2], status_bytes[newbyte][2])]])
 516         self.ss_block, self.es_block = old_ss_block, old_es_block
 517         # Deliberately not resetting self.cmd or self.state.
 518
 519     def handle_garbage_msg(self, newbyte):
 520         # Handle messages that are either not handled or are corrupt.
 521         self.es_block = self.es
 522         if newbyte is not None:
 523             self.cmd.append(newbyte)
 524             return
 525         payload = '<empty>'
 526         max_bytes = 16 # Put a limit on the length on the hex dump.
 527         for index in range(len(self.cmd)):
 528             if index == max_bytes:
 529                 payload += ' ...'
 530                 break
 531             if index == 0:
 532                 payload = '0x%02x' % self.cmd[index]
 533             else:
 534                 payload += ' 0x%02x' % self.cmd[index]
 535         self.putx([2, ['UNHANDLED DATA: %s' % payload,
 536                       'UNHANDLED', '???', '?']])
 537         self.cmd, self.state = [], 'IDLE'
 538         self.hard_clear_status_byte()
 539
 540     def handle_state(self, state, newbyte):
 541         # 'newbyte' can either be:
 542         # 1. Value between 0x00-0xff, deal with the byte normally.
 543         # 2. Value of 'None' which means "flush any buffered data".
 544         if state == 'HANDLE CHANNEL MSG':
 545             self.handle_channel_msg(newbyte)
 546         elif state == 'HANDLE SYSEX MSG':
 547             self.handle_sysex_msg(newbyte)
 548         elif state == 'HANDLE SYSCOMMON MSG':
 549             self.handle_syscommon_msg(newbyte)
 550         elif state == 'HANDLE SYSREALTIME MSG':
 551             self.handle_sysrealtime_msg(newbyte)
 552         elif state == 'BUFFER GARBAGE MSG':
 553             self.handle_garbage_msg(newbyte)
 554
 555     def get_next_state(self, newbyte):
 556         # 'newbyte' must be a valid byte between 0x00 and 0xff.
 557         #
 558         # Try to determine the state based off of the 'newbyte' parameter.
 559         if newbyte in range(0x80, 0xef + 1):
 560             return 'HANDLE CHANNEL MSG'
 561         if newbyte == 0xf0:
 562             return 'HANDLE SYSEX MSG'
 563         if newbyte in range(0xf1, 0xf7):
 564             return'HANDLE SYSCOMMON MSG'
 565         if newbyte in range(0xf8, 0xff + 1):
 566             return 'HANDLE SYSREALTIME MSG'
 567         # Passing 0xf7 is an error; messages don't start with 0xf7.
 568         if newbyte == 0xf7:
 569             return 'BUFFER GARBAGE MSG'
 570         # Next, base the state off of self.status_byte.
 571         if self.status_byte < 0x80:
 572             return 'BUFFER GARBAGE MSG'
 573         return self.get_next_state(self.status_byte)
 574
 575     def decode(self, ss, es, data):
 576         ptype, rxtx, pdata = data
 577         state = 'IDLE'
 578
 579         # For now, ignore all UART packets except the actual data packets.
 580         if ptype != 'DATA':
 581             return
 582
 583         # We're only interested in the byte value (not individual bits).
 584         pdata = pdata[0]
 585
 586         # Short MIDI overview:
 587         #  - Status bytes are 0x80-0xff, data bytes are 0x00-0x7f.
 588         #  - Most messages: 1 status byte, 1-2 data bytes.
 589         #  - Real-time system messages: always 1 byte.
 590         #  - SysEx messages: 1 status byte, n data bytes, EOX byte.
 591         #
 592         # Aspects of the MIDI protocol that complicate decoding:
 593         #  - MIDI System Realtime messages can briefly interrupt other
 594         #    messages already in progress.
 595         #  - "Running Status" allows for omitting the status byte in most
 596         #    scenarios if sequential messages have the same status byte.
 597         #  - System Exclusive (SysEx) messages can be terminated by ANY
 598         #    status byte (not limited to EOX byte).
 599
 600         # State machine.
 601         if pdata >= 0x80 and pdata != 0xf7:
 602             state = self.get_next_state(pdata)
 603             if state != 'HANDLE SYSREALTIME MSG' and self.state != 'IDLE':
 604                 # Flush the previous data since a new message is starting.
 605                 self.handle_state(self.state, None)
 606             # Cache ss and es -after- flushing previous data.
 607             self.ss, self.es = ss, es
 608             # This is a status byte, remember the start sample.
 609             if state != 'HANDLE SYSREALTIME MSG':
 610                 self.ss_block = ss
 611         elif self.state == 'IDLE' or self.state == 'BUFFER GARBAGE MSG':
 612             # Deal with "running status" or that we're buffering garbage.
 613             self.ss, self.es = ss, es
 614             if self.state == 'IDLE':
 615                 self.ss_block = ss
 616             state = self.get_next_state(pdata)
 617         else:
 618             self.ss, self.es = ss, es
 619             state = self.state
 620
 621         # Yes, this is intentionally _not_ an 'elif' here.
 622         if state != 'HANDLE SYSREALTIME MSG':
 623             self.state = state
 624         if state == 'BUFFER GARBAGE MSG':
 625             self.status_byte = 0
 626         self.handle_state(state, pdata)