+++ /dev/null
-# -*- coding: ISO-8859-1 -*-\r
-\r
-# std library\r
-from struct import unpack\r
-\r
-# custom\r
-from DataTypeConverters import readBew, readVar, varLen, toBytes\r
-\r
-# uhh I don't really like this, but there are so many constants to \r
-# import otherwise\r
-from constants import *\r
-\r
-\r
-class EventDispatcher:\r
-\r
-\r
- def __init__(self, outstream):\r
- \r
- """\r
- \r
- The event dispatcher generates events on the outstream.\r
- \r
- """\r
- \r
- # internal values, don't mess with 'em directly\r
- self.outstream = outstream\r
- \r
- # public flags\r
-\r
- # A note_on with a velocity of 0x00 is actually the same as a \r
- # note_off with a velocity of 0x40. When \r
- # "convert_zero_velocity" is set, the zero velocity note_on's \r
- # automatically gets converted into note_off's. This is a less \r
- # suprising behaviour for those that are not into the intimate \r
- # details of the midi spec.\r
- self.convert_zero_velocity = 1\r
- \r
- # If dispatch_continuos_controllers is true, continuos \r
- # controllers gets dispatched to their defined handlers. Else \r
- # they just trigger the "continuous_controller" event handler.\r
- self.dispatch_continuos_controllers = 1 # NOT IMPLEMENTED YET\r
- \r
- # If dispatch_meta_events is true, meta events get's dispatched \r
- # to their defined events. Else they all they trigger the \r
- # "meta_event" handler.\r
- self.dispatch_meta_events = 1\r
-\r
-\r
-\r
- def header(self, format, nTracks, division):\r
- "Triggers the header event"\r
- self.outstream.header(format, nTracks, division)\r
-\r
-\r
- def start_of_track(self, current_track):\r
- "Triggers the start of track event"\r
- \r
- # I do this twice so that users can overwrite the \r
- # start_of_track event handler without worrying whether the \r
- # track number is updated correctly.\r
- self.outstream.set_current_track(current_track)\r
- self.outstream.start_of_track(current_track)\r
- \r
- \r
- def sysex_event(self, data):\r
- "Dispatcher for sysex events"\r
- self.outstream.sysex_event(data)\r
- \r
- \r
- def eof(self):\r
- "End of file!"\r
- self.outstream.eof()\r
-\r
-\r
- def update_time(self, new_time=0, relative=1):\r
- "Updates relative/absolute time."\r
- self.outstream.update_time(new_time, relative)\r
- \r
- \r
- def reset_time(self):\r
- "Updates relative/absolute time."\r
- self.outstream.reset_time()\r
- \r
- \r
- # Event dispatchers for similar types of events\r
- \r
- \r
- def channel_messages(self, hi_nible, channel, data):\r
- \r
- "Dispatches channel messages"\r
- \r
- stream = self.outstream\r
- data = toBytes(data)\r
- \r
- if (NOTE_ON & 0xF0) == hi_nible:\r
- note, velocity = data\r
- # note_on with velocity 0x00 are same as note \r
- # off with velocity 0x40 according to spec!\r
- if velocity==0 and self.convert_zero_velocity:\r
- stream.note_off(channel, note, 0x40)\r
- else:\r
- stream.note_on(channel, note, velocity)\r
- \r
- elif (NOTE_OFF & 0xF0) == hi_nible:\r
- note, velocity = data\r
- stream.note_off(channel, note, velocity)\r
- \r
- elif (AFTERTOUCH & 0xF0) == hi_nible:\r
- note, velocity = data\r
- stream.aftertouch(channel, note, velocity)\r
- \r
- elif (CONTINUOUS_CONTROLLER & 0xF0) == hi_nible:\r
- controller, value = data\r
- # A lot of the cc's are defined, so we trigger those directly\r
- if self.dispatch_continuos_controllers:\r
- self.continuous_controllers(channel, controller, value)\r
- else:\r
- stream.continuous_controller(channel, controller, value)\r
- \r
- elif (PATCH_CHANGE & 0xF0) == hi_nible:\r
- program = data[0]\r
- stream.patch_change(channel, program)\r
- \r
- elif (CHANNEL_PRESSURE & 0xF0) == hi_nible:\r
- pressure = data[0]\r
- stream.channel_pressure(channel, pressure)\r
- \r
- elif (PITCH_BEND & 0xF0) == hi_nible:\r
- hibyte, lobyte = data\r
- value = (hibyte<<7) + lobyte\r
- stream.pitch_bend(channel, value)\r
-\r
- else:\r
-\r
- raise ValueError, 'Illegal channel message!'\r
-\r
-\r
-\r
- def continuous_controllers(self, channel, controller, value):\r
- \r
- "Dispatches channel messages"\r
-\r
- stream = self.outstream\r
- \r
- # I am not really shure if I ought to dispatch continuous controllers\r
- # There's so many of them that it can clutter up the OutStream \r
- # classes.\r
- \r
- # So I just trigger the default event handler\r
- stream.continuous_controller(channel, controller, value)\r
-\r
-\r
-\r
- def system_commons(self, common_type, common_data):\r
- \r
- "Dispatches system common messages"\r
- \r
- stream = self.outstream\r
- \r
- # MTC Midi time code Quarter value\r
- if common_type == MTC:\r
- data = readBew(common_data)\r
- msg_type = (data & 0x07) >> 4\r
- values = (data & 0x0F)\r
- stream.midi_time_code(msg_type, values)\r
- \r
- elif common_type == SONG_POSITION_POINTER:\r
- hibyte, lobyte = toBytes(common_data)\r
- value = (hibyte<<7) + lobyte\r
- stream.song_position_pointer(value)\r
-\r
- elif common_type == SONG_SELECT:\r
- data = readBew(common_data)\r
- stream.song_select(data)\r
-\r
- elif common_type == TUNING_REQUEST:\r
- # no data then\r
- stream.tuning_request(time=None)\r
-\r
-\r
-\r
- def meta_event(self, meta_type, data):\r
- \r
- "Dispatches meta events"\r
- \r
- stream = self.outstream\r
- \r
- # SEQUENCE_NUMBER = 0x00 (00 02 ss ss (seq-number))\r
- if meta_type == SEQUENCE_NUMBER:\r
- number = readBew(data)\r
- stream.sequence_number(number)\r
- \r
- # TEXT = 0x01 (01 len text...)\r
- elif meta_type == TEXT:\r
- stream.text(data)\r
- \r
- # COPYRIGHT = 0x02 (02 len text...)\r
- elif meta_type == COPYRIGHT:\r
- stream.copyright(data)\r
- \r
- # SEQUENCE_NAME = 0x03 (03 len text...)\r
- elif meta_type == SEQUENCE_NAME:\r
- stream.sequence_name(data)\r
- \r
- # INSTRUMENT_NAME = 0x04 (04 len text...)\r
- elif meta_type == INSTRUMENT_NAME:\r
- stream.instrument_name(data)\r
- \r
- # LYRIC = 0x05 (05 len text...)\r
- elif meta_type == LYRIC:\r
- stream.lyric(data)\r
- \r
- # MARKER = 0x06 (06 len text...)\r
- elif meta_type == MARKER:\r
- stream.marker(data)\r
- \r
- # CUEPOINT = 0x07 (07 len text...)\r
- elif meta_type == CUEPOINT:\r
- stream.cuepoint(data)\r
- \r
- # PROGRAM_NAME = 0x08 (05 len text...)\r
- elif meta_type == PROGRAM_NAME:\r
- stream.program_name(data)\r
- \r
- # DEVICE_NAME = 0x09 (09 len text...)\r
- elif meta_type == DEVICE_NAME:\r
- stream.device_name(data)\r
- \r
- # MIDI_CH_PREFIX = 0x20 (20 01 channel)\r
- elif meta_type == MIDI_CH_PREFIX:\r
- channel = readBew(data)\r
- stream.midi_ch_prefix(channel)\r
- \r
- # MIDI_PORT = 0x21 (21 01 port (legacy stuff))\r
- elif meta_type == MIDI_PORT:\r
- port = readBew(data)\r
- stream.midi_port(port)\r
- \r
- # END_OFF_TRACK = 0x2F (2F 00)\r
- elif meta_type == END_OF_TRACK:\r
- stream.end_of_track()\r
- \r
- # TEMPO = 0x51 (51 03 tt tt tt (tempo in us/quarternote))\r
- elif meta_type == TEMPO:\r
- b1, b2, b3 = toBytes(data)\r
- # uses 3 bytes to represent time between quarter \r
- # notes in microseconds\r
- stream.tempo((b1<<16) + (b2<<8) + b3)\r
- \r
- # SMTP_OFFSET = 0x54 (54 05 hh mm ss ff xx)\r
- elif meta_type == SMTP_OFFSET:\r
- hour, minute, second, frame, framePart = toBytes(data)\r
- stream.smtp_offset(\r
- hour, minute, second, frame, framePart)\r
- \r
- # TIME_SIGNATURE = 0x58 (58 04 nn dd cc bb)\r
- elif meta_type == TIME_SIGNATURE:\r
- nn, dd, cc, bb = toBytes(data)\r
- stream.time_signature(nn, dd, cc, bb)\r
- \r
- # KEY_SIGNATURE = 0x59 (59 02 sf mi)\r
- elif meta_type == KEY_SIGNATURE:\r
- sf, mi = toBytes(data)\r
- stream.key_signature(sf, mi)\r
- \r
- # SPECIFIC = 0x7F (Sequencer specific event)\r
- elif meta_type == SPECIFIC:\r
- meta_data = toBytes(data)\r
- stream.sequencer_specific(meta_data)\r
- \r
- # Handles any undefined meta events\r
- else: # undefined meta type\r
- meta_data = toBytes(data)\r
- stream.meta_event(meta_type, meta_data)\r
-\r
-\r
-\r
-\r
-\r
-if __name__ == '__main__':\r
-\r
-\r
- from MidiToText import MidiToText\r
- \r
- outstream = MidiToText()\r
- dispatcher = EventDispatcher(outstream)\r
- dispatcher.channel_messages(NOTE_ON, 0x00, '\x40\x40')
\ No newline at end of file
projects / minwii.git / blobdiff