Source Code for Module echonest.remix.support.midi.EventDispatcher

  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') 
288