- Refactored the MidiEvent dataclass to handle its own conversions to and from mido messages, eliminating repetitive if/elif blocks in sequencer.py

- Added `_has_pitch_at_beat()` helper to PatternBuilder and refactored the Euclidean, Bresenham, and Ghost Fill algorithms to use it, removing duplicated logic.
- Deleted the private _msg_to_midi_event helper in the Sequencer class since it was superseded by the cleaner factory method on MidiEvent

Author: Simon Holliday

subsequence/pattern_algorithmic.py

@@ -45,6 +45,7 @@ def note (
 			duration: float,
 		) -> "PatternAlgorithmicMixin": ...
 		def _resolve_pitch (self, pitch: typing.Union[int, str]) -> int: ...
+		def _has_pitch_at_beat (self, pitch: typing.Union[int, str], beat: float) -> bool: ...
 
 	def _place_rhythm_sequence (
 		self,
@@ -64,7 +65,6 @@ def _place_rhythm_sequence (
 		across the pattern length. Zeros and dropout-gated steps are skipped.
 		"""
 
-		midi_pitch = self._resolve_pitch(pitch)
 		step_duration = self._pattern.length / len(sequence)
 
 		for i, hit_value in enumerate(sequence):
@@ -75,11 +75,8 @@ def _place_rhythm_sequence (
 			if dropout > 0 and rng.random() < dropout:
 				continue
 
-			if no_overlap:
-				pulse = int(i * step_duration * subsequence.constants.MIDI_QUARTER_NOTE + 0.5)
-				if pulse in self._pattern.steps:
-					if any(n.pitch == midi_pitch for n in self._pattern.steps[pulse].notes):
-						continue
+			if no_overlap and self._has_pitch_at_beat(pitch, i * step_duration):
+				continue
 
 			self.note(pitch=pitch, beat=i * step_duration, velocity=velocity, duration=duration)
 
@@ -265,12 +262,8 @@ def bresenham_poly (
 
 			pitch = voice_names[voice_idx]
 
-			if no_overlap:
-				midi_pitch = self._resolve_pitch(pitch)
-				pulse = int(step_idx * step_duration * subsequence.constants.MIDI_QUARTER_NOTE + 0.5)
-				if pulse in self._pattern.steps:
-					if any(n.pitch == midi_pitch for n in self._pattern.steps[pulse].notes):
-						continue
+			if no_overlap and self._has_pitch_at_beat(pitch, step_idx * step_duration):
+				continue
 
 			if isinstance(velocity, dict):
 				vel = velocity.get(pitch, subsequence.constants.velocity.DEFAULT_VELOCITY)
@@ -483,7 +476,6 @@ def ghost_fill (
 		if max_weight <= 0:
 			return self
 
-		midi_pitch = self._resolve_pitch(pitch)
 		step_duration = self._pattern.length / grid
 
 		for i in range(grid):
@@ -492,11 +484,8 @@ def ghost_fill (
 			if rng.random() >= prob:
 				continue
 
-			if no_overlap:
-				pulse = int(round(i * step_duration * subsequence.constants.MIDI_QUARTER_NOTE))
-				if pulse in self._pattern.steps:
-					if any(n.pitch == midi_pitch for n in self._pattern.steps[pulse].notes):
-						continue
+			if no_overlap and self._has_pitch_at_beat(pitch, i * step_duration):
+				continue
 
 			if callable(velocity):
 				vel = int(velocity(i))

subsequence/pattern_builder.py

@@ -149,6 +149,15 @@ def grid (self) -> int:
 		"""Number of grid slots in this pattern (e.g. 16 for a 4-beat sixteenth-note pattern)."""
 		return self._default_grid
 
+	def _has_pitch_at_beat (self, pitch: typing.Union[int, str], beat: float) -> bool:
+		"""Helper to check if a pitch is already sounding at a specific beat."""
+		midi_pitch = self._resolve_pitch(pitch)
+		pulse = int(beat * subsequence.constants.MIDI_QUARTER_NOTE + 0.5)
+		if pulse in self._pattern.steps:
+			return any(n.pitch == midi_pitch for n in self._pattern.steps[pulse].notes)
+		return False
+
+
 	@property
 	def c (self) -> typing.Optional[subsequence.conductor.Conductor]:
 

subsequence/sequencer.py

@@ -60,6 +60,86 @@ class MidiEvent:
 	device: int = dataclasses.field(compare=False, default=0)
 
 
+	def to_mido (self) -> typing.Optional[typing.Union[mido.Message, mido.MetaMessage]]:
+
+		"""Convert this event to a mido.Message, or None if it's an internal type (like OSC)."""
+
+		if self.message_type in ('note_on', 'note_off'):
+			return mido.Message(
+				self.message_type,
+				channel = self.channel,
+				note = self.note,
+				velocity = self.velocity
+			)
+
+		if self.message_type == 'control_change':
+			return mido.Message(
+				'control_change',
+				channel = self.channel,
+				control = self.control,
+				value = self.value
+			)
+
+		if self.message_type == 'pitchwheel':
+			return mido.Message(
+				'pitchwheel',
+				channel = self.channel,
+				pitch = self.value
+			)
+
+		if self.message_type == 'program_change':
+			return mido.Message(
+				'program_change',
+				channel = self.channel,
+				program = self.value
+			)
+
+		if self.message_type == 'sysex':
+			return mido.Message(
+				'sysex',
+				data = self.data if self.data is not None else b''
+			)
+
+		return None
+
+
+	@classmethod
+	def from_mido (cls, pulse: int, msg: typing.Union[mido.Message, mido.MetaMessage], device: int = 0) -> "MidiEvent":
+
+		"""Convert a mido.Message to a MidiEvent."""
+
+		if msg.type == 'pitchwheel':
+			return cls(
+				pulse = pulse,
+				message_type = 'pitchwheel',
+				channel = msg.channel,
+				value = msg.pitch,
+				device = device,
+			)
+
+		if msg.type == 'control_change':
+			return cls(
+				pulse = pulse,
+				message_type = 'control_change',
+				channel = msg.channel,
+				control = msg.control,
+				value = msg.value,
+				device = device,
+			)
+
+		return cls(
+			pulse = pulse,
+			message_type = msg.type,
+			channel = getattr(msg, 'channel', 0),
+			value = getattr(msg, 'value', 0),
+			note = getattr(msg, 'note', 0),
+			velocity = getattr(msg, 'velocity', 0),
+			data = getattr(msg, 'data', None),
+			control = getattr(msg, 'control', 0),
+			device = device,
+		)
+
+
 @dataclasses.dataclass
 class ScheduledPattern:
 
@@ -571,36 +651,6 @@ def _on_midi_input (self, message: typing.Any, device_idx: int = 0) -> None:
 					self._forward_buffer.append((self.pulse_count, out_msg))
 
 
-	def _msg_to_midi_event (self, pulse: int, msg: typing.Any) -> MidiEvent:
-
-		"""Convert a mido.Message to a MidiEvent at the given pulse.
-
-		Used to inject queued CC forwards into the event heap.
-		"""
-
-		if msg.type == 'pitchwheel':
-			return MidiEvent(
-				pulse = pulse,
-				message_type = 'pitchwheel',
-				channel = msg.channel,
-				value = msg.pitch,
-			)
-		elif msg.type == 'control_change':
-			return MidiEvent(
-				pulse = pulse,
-				message_type = 'control_change',
-				channel = msg.channel,
-				control = msg.control,
-				value = msg.value,
-			)
-		else:
-			return MidiEvent(
-				pulse = pulse,
-				message_type = msg.type,
-				channel = getattr(msg, 'channel', 0),
-				value = getattr(msg, 'value', 0),
-			)
-
 
 	def _estimate_bpm (self, tick_time: float) -> None:
 
@@ -1280,7 +1330,7 @@ async def _process_pulse (self, pulse: int) -> None:
 			# while the callback thread calls append().
 			while self._forward_buffer:
 				fwd_pulse, fwd_msg = self._forward_buffer.popleft()
-				heapq.heappush(self.event_queue, self._msg_to_midi_event(fwd_pulse, fwd_msg))
+				heapq.heappush(self.event_queue, MidiEvent.from_mido(fwd_pulse, fwd_msg))
 
 			while self.event_queue and self.event_queue[0].pulse <= pulse:
 
@@ -1298,21 +1348,9 @@ async def _process_pulse (self, pulse: int) -> None:
 
 				if self.recording and event.message_type != 'osc':
 
-					if event.message_type in ('note_on', 'note_off'):
-						self._record_event(event.pulse, mido.Message(event.message_type, channel=event.channel, note=event.note, velocity=event.velocity))
-
-					elif event.message_type == 'control_change':
-						self._record_event(event.pulse, mido.Message('control_change', channel=event.channel, control=event.control, value=event.value))
-
-					elif event.message_type == 'pitchwheel':
-						self._record_event(event.pulse, mido.Message('pitchwheel', channel=event.channel, pitch=event.value))
-
-					elif event.message_type == 'program_change':
-						self._record_event(event.pulse, mido.Message('program_change', channel=event.channel, program=event.value))
-
-					elif event.message_type == 'sysex':
-						raw = event.data if event.data is not None else b''
-						self._record_event(event.pulse, mido.Message('sysex', data=raw))
+					mido_msg = event.to_mido()
+					if mido_msg is not None:
+						self._record_event(event.pulse, mido_msg)
 
 
 	async def _stop_all_active_notes (self) -> None:
@@ -1343,49 +1381,14 @@ def _send_midi (self, event: MidiEvent) -> None:
 
 			try:
 
-				if event.message_type in ('note_on', 'note_off'):
-					msg = mido.Message(
-						event.message_type,
-						channel = event.channel,
-						note = event.note,
-						velocity = event.velocity
-					)
-
-				elif event.message_type == 'control_change':
-					msg = mido.Message(
-						'control_change',
-						channel = event.channel,
-						control = event.control,
-						value = event.value
-					)
-
-				elif event.message_type == 'pitchwheel':
-					msg = mido.Message(
-						'pitchwheel',
-						channel = event.channel,
-						pitch = event.value
-					)
-
-				elif event.message_type == 'program_change':
-					msg = mido.Message(
-						'program_change',
-						channel = event.channel,
-						program = event.value
-					)
-
-				elif event.message_type == 'sysex':
-					msg = mido.Message(
-						'sysex',
-						data = event.data if event.data is not None else b''
-					)
-
-				elif event.message_type == 'osc':
+				if event.message_type == 'osc':
 					if self.osc_server is not None:
 						address, args = event.data
 						self.osc_server.send(address, *args)
 					return
 
-				else:
+				msg = event.to_mido()
+				if msg is None:
 					return
 
 				port.send(msg)

tests/test_cc_forward.py

@@ -287,23 +287,21 @@ async def test_queued_drained_in_process_pulse(patch_midi: None) -> None:
 	assert any(m.control == 74 and m.value == 64 for m in spy.sent)
 
 
-def test_msg_to_midi_event_cc(patch_midi: None) -> None:
-	"""_msg_to_midi_event should correctly convert a CC message."""
-	seq = _make_sequencer(conftest.SpyMidiOut())
+def test_midi_event_from_mido_cc(patch_midi: None) -> None:
+	"""from_mido should correctly convert a CC message."""
 	msg = mido.Message('control_change', channel=1, control=74, value=100)
-	event = seq._msg_to_midi_event(10, msg)
+	event = subsequence.sequencer.MidiEvent.from_mido(10, msg)
 	assert event.pulse == 10
 	assert event.message_type == 'control_change'
 	assert event.channel == 1
 	assert event.control == 74
 	assert event.value == 100
 
 
-def test_msg_to_midi_event_pitchwheel(patch_midi: None) -> None:
-	"""_msg_to_midi_event should correctly convert a pitchwheel message."""
-	seq = _make_sequencer(conftest.SpyMidiOut())
+def test_midi_event_from_mido_pitchwheel(patch_midi: None) -> None:
+	"""from_mido should correctly convert a pitchwheel message."""
 	msg = mido.Message('pitchwheel', channel=0, pitch=-4096)
-	event = seq._msg_to_midi_event(5, msg)
+	event = subsequence.sequencer.MidiEvent.from_mido(5, msg)
 	assert event.pulse == 5
 	assert event.message_type == 'pitchwheel'
 	assert event.value == -4096