Filtrage des événements au niveau avec l'api de logging python.
[minwii.git] / 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')