src/mxmMidi/EventDispatcher.py

   1 # -*- coding: ISO-8859-1 -*-
   2
   3 # std library
   4 from struct import unpack
   5
   6 # custom
   7 from DataTypeConverters import readBew, readVar, varLen, toBytes
   8
   9 # uhh I don't really like this, but there are so many constants to
  10 # import otherwise
  11 from constants import *
  12
  13
  14 class EventDispatcher:
  15
  16
  17     def __init__(self, outstream):
  18
  19         """
  20
  21         The event dispatcher generates events on the outstream.
  22
  23         """
  24
  25         # internal values, don't mess with 'em directly
  26         self.outstream = outstream
  27
  28         # public flags
  29
  30         # A note_on with a velocity of 0x00 is actually the same as a
  31         # note_off with a velocity of 0x40. When
  32         # "convert_zero_velocity" is set, the zero velocity note_on's
  33         # automatically gets converted into note_off's. This is a less
  34         # suprising behaviour for those that are not into the intimate
  35         # details of the midi spec.
  36         self.convert_zero_velocity = 1
  37
  38         # If dispatch_continuos_controllers is true, continuos
  39         # controllers gets dispatched to their defined handlers. Else
  40         # they just trigger the "continuous_controller" event handler.
  41         self.dispatch_continuos_controllers = 1 # NOT IMPLEMENTED YET
  42
  43         # If dispatch_meta_events is true, meta events get's dispatched
  44         # to their defined events. Else they all they trigger the
  45         # "meta_event" handler.
  46         self.dispatch_meta_events = 1
  47
  48
  49
  50     def header(self, format, nTracks, division):
  51         "Triggers the header event"
  52         self.outstream.header(format, nTracks, division)
  53
  54
  55     def start_of_track(self, current_track):
  56         "Triggers the start of track event"
  57
  58         # I do this twice so that users can overwrite the
  59         # start_of_track event handler without worrying whether the
  60         # track number is updated correctly.
  61         self.outstream.set_current_track(current_track)
  62         self.outstream.start_of_track(current_track)
  63
  64
  65     def sysex_event(self, data):
  66         "Dispatcher for sysex events"
  67         self.outstream.sysex_event(data)
  68
  69
  70     def eof(self):
  71         "End of file!"
  72         self.outstream.eof()
  73
  74
  75     def update_time(self, new_time=0, relative=1):
  76         "Updates relative/absolute time."
  77         self.outstream.update_time(new_time, relative)
  78
  79
  80     def reset_time(self):
  81         "Updates relative/absolute time."
  82         self.outstream.reset_time()
  83
  84
  85     # Event dispatchers for similar types of events
  86
  87
  88     def channel_messages(self, hi_nible, channel, data):
  89
  90         "Dispatches channel messages"
  91
  92         stream = self.outstream
  93         data = toBytes(data)
  94
  95         if (NOTE_ON & 0xF0) == hi_nible:
  96             note, velocity = data
  97             # note_on with velocity 0x00 are same as note
  98             # off with velocity 0x40 according to spec!
  99             if velocity==0 and self.convert_zero_velocity:
 100                 stream.note_off(channel, note, 0x40)
 101             else:
 102                 stream.note_on(channel, note, velocity)
 103
 104         elif (NOTE_OFF & 0xF0) == hi_nible:
 105             note, velocity = data
 106             stream.note_off(channel, note, velocity)
 107
 108         elif (AFTERTOUCH & 0xF0) == hi_nible:
 109             note, velocity = data
 110             stream.aftertouch(channel, note, velocity)
 111
 112         elif (CONTINUOUS_CONTROLLER & 0xF0) == hi_nible:
 113             controller, value = data
 114             # A lot of the cc's are defined, so we trigger those directly
 115             if self.dispatch_continuos_controllers:
 116                 self.continuous_controllers(channel, controller, value)
 117             else:
 118                 stream.continuous_controller(channel, controller, value)
 119
 120         elif (PATCH_CHANGE & 0xF0) == hi_nible:
 121             program = data[0]
 122             stream.patch_change(channel, program)
 123
 124         elif (CHANNEL_PRESSURE & 0xF0) == hi_nible:
 125             pressure = data[0]
 126             stream.channel_pressure(channel, pressure)
 127
 128         elif (PITCH_BEND & 0xF0) == hi_nible:
 129             hibyte, lobyte = data
 130             value = (hibyte<<7) + lobyte
 131             stream.pitch_bend(channel, value)
 132
 133         else:
 134
 135             raise ValueError, 'Illegal channel message!'
 136
 137
 138
 139     def continuous_controllers(self, channel, controller, value):
 140
 141         "Dispatches channel messages"
 142
 143         stream = self.outstream
 144
 145         # I am not really shure if I ought to dispatch continuous controllers
 146         # There's so many of them that it can clutter up the OutStream
 147         # classes.
 148
 149         # So I just trigger the default event handler
 150         stream.continuous_controller(channel, controller, value)
 151
 152
 153
 154     def system_commons(self, common_type, common_data):
 155
 156         "Dispatches system common messages"
 157
 158         stream = self.outstream
 159
 160         # MTC Midi time code Quarter value
 161         if common_type == MTC:
 162             data = readBew(common_data)
 163             msg_type = (data & 0x07) >> 4
 164             values = (data & 0x0F)
 165             stream.midi_time_code(msg_type, values)
 166
 167         elif common_type == SONG_POSITION_POINTER:
 168             hibyte, lobyte = toBytes(common_data)
 169             value = (hibyte<<7) + lobyte
 170             stream.song_position_pointer(value)
 171
 172         elif common_type == SONG_SELECT:
 173             data = readBew(common_data)
 174             stream.song_select(data)
 175
 176         elif common_type == TUNING_REQUEST:
 177             # no data then
 178             stream.tuning_request(time=None)
 179
 180
 181
 182     def meta_event(self, meta_type, data):
 183
 184         "Dispatches meta events"
 185
 186         stream = self.outstream
 187
 188         # SEQUENCE_NUMBER = 0x00 (00 02 ss ss (seq-number))
 189         if meta_type == SEQUENCE_NUMBER:
 190             number = readBew(data)
 191             stream.sequence_number(number)
 192
 193         # TEXT = 0x01 (01 len text...)
 194         elif meta_type == TEXT:
 195             stream.text(data)
 196
 197         # COPYRIGHT = 0x02 (02 len text...)
 198         elif meta_type == COPYRIGHT:
 199             stream.copyright(data)
 200
 201         # SEQUENCE_NAME = 0x03 (03 len text...)
 202         elif meta_type == SEQUENCE_NAME:
 203             stream.sequence_name(data)
 204
 205         # INSTRUMENT_NAME = 0x04 (04 len text...)
 206         elif meta_type == INSTRUMENT_NAME:
 207             stream.instrument_name(data)
 208
 209         # LYRIC = 0x05 (05 len text...)
 210         elif meta_type == LYRIC:
 211             stream.lyric(data)
 212
 213         # MARKER = 0x06 (06 len text...)
 214         elif meta_type == MARKER:
 215             stream.marker(data)
 216
 217         # CUEPOINT = 0x07 (07 len text...)
 218         elif meta_type == CUEPOINT:
 219             stream.cuepoint(data)
 220
 221         # PROGRAM_NAME = 0x08 (05 len text...)
 222         elif meta_type == PROGRAM_NAME:
 223             stream.program_name(data)
 224
 225         # DEVICE_NAME = 0x09 (09 len text...)
 226         elif meta_type == DEVICE_NAME:
 227             stream.device_name(data)
 228
 229         # MIDI_CH_PREFIX = 0x20 (20 01 channel)
 230         elif meta_type == MIDI_CH_PREFIX:
 231             channel = readBew(data)
 232             stream.midi_ch_prefix(channel)
 233
 234         # MIDI_PORT  = 0x21 (21 01 port (legacy stuff))
 235         elif meta_type == MIDI_PORT:
 236             port = readBew(data)
 237             stream.midi_port(port)
 238
 239         # END_OFF_TRACK = 0x2F (2F 00)
 240         elif meta_type == END_OF_TRACK:
 241             stream.end_of_track()
 242
 243         # TEMPO = 0x51 (51 03 tt tt tt (tempo in us/quarternote))
 244         elif meta_type == TEMPO:
 245             b1, b2, b3 = toBytes(data)
 246             # uses 3 bytes to represent time between quarter
 247             # notes in microseconds
 248             stream.tempo((b1<<16) + (b2<<8) + b3)
 249
 250         # SMTP_OFFSET = 0x54 (54 05 hh mm ss ff xx)
 251         elif meta_type == SMTP_OFFSET:
 252             hour, minute, second, frame, framePart = toBytes(data)
 253             stream.smtp_offset(
 254                     hour, minute, second, frame, framePart)
 255
 256         # TIME_SIGNATURE = 0x58 (58 04 nn dd cc bb)
 257         elif meta_type == TIME_SIGNATURE:
 258             nn, dd, cc, bb = toBytes(data)
 259             stream.time_signature(nn, dd, cc, bb)
 260
 261         # KEY_SIGNATURE = 0x59 (59 02 sf mi)
 262         elif meta_type == KEY_SIGNATURE:
 263             sf, mi = toBytes(data)
 264             stream.key_signature(sf, mi)
 265
 266         # SPECIFIC = 0x7F (Sequencer specific event)
 267         elif meta_type == SPECIFIC:
 268             meta_data = toBytes(data)
 269             stream.sequencer_specific(meta_data)
 270
 271         # Handles any undefined meta events
 272         else: # undefined meta type
 273             meta_data = toBytes(data)
 274             stream.meta_event(meta_type, meta_data)
 275
 276
 277
 278
 279
 280 if __name__ == '__main__':
 281
 282
 283     from MidiToText import MidiToText
 284
 285     outstream = MidiToText()
 286     dispatcher = EventDispatcher(outstream)
 287     dispatcher.channel_messages(NOTE_ON, 0x00, '\x40\x40')