thomas
1f110866f5
CI / backend (pull_request) Canceled after 0s
CI / shell (pull_request) Canceled after 0s
CI / frontend (pull_request) Canceled after 0s
CI / arm64-smoke (pull_request) Canceled after 0s
CI / backend (push) Canceled after 0s
CI / shell (push) Canceled after 0s
CI / frontend (push) Canceled after 0s
CI / arm64-smoke (push) Canceled after 0s
88 lines
2.9 KiB
Python
88 lines
2.9 KiB
Python
from __future__ import annotations
|
|
|
|
from dataclasses import dataclass, field
|
|
from datetime import UTC, datetime
|
|
from time import monotonic
|
|
from typing import Any
|
|
|
|
|
|
def list_input_devices() -> list[str]:
|
|
import mido
|
|
|
|
return list(mido.get_input_names())
|
|
|
|
|
|
def match_input_device(device_name: str, devices: list[str] | None = None) -> str | None:
|
|
available = devices or list_input_devices()
|
|
requested = device_name.casefold()
|
|
for name in available:
|
|
if requested in name.casefold():
|
|
return name
|
|
return None
|
|
|
|
|
|
def normalize_message(message: Any, bridge_id: str, device_name: str) -> dict[str, object] | None:
|
|
message_type = getattr(message, "type", None)
|
|
if message_type not in {"note_on", "note_off", "control_change", "program_change"}:
|
|
return None
|
|
number = getattr(message, "note", None)
|
|
if message_type in {"control_change", "program_change"}:
|
|
number = getattr(message, "control", None) if message_type == "control_change" else getattr(message, "program", None)
|
|
if number is None:
|
|
return None
|
|
value = getattr(message, "velocity", None)
|
|
if message_type == "control_change":
|
|
value = getattr(message, "value", 0)
|
|
if message_type == "program_change":
|
|
value = getattr(message, "value", 127)
|
|
if value is None:
|
|
value = 0
|
|
return {
|
|
"type": "midi_event",
|
|
"protocol_version": 1,
|
|
"bridge_id": bridge_id,
|
|
"device": device_name,
|
|
"timestamp": datetime.now(UTC).isoformat(),
|
|
"message": {
|
|
"type": message_type,
|
|
"channel": int(getattr(message, "channel", 0)),
|
|
"number": int(number),
|
|
"value": int(value),
|
|
},
|
|
}
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class MidiEventFilter:
|
|
channel_filter: int | None
|
|
control_change_interval_ms: int
|
|
suppress_duplicate_values: bool
|
|
_last_sent_at: dict[tuple[str, int, int], float] = field(default_factory=dict, init=False)
|
|
_last_values: dict[tuple[str, int, int], int] = field(default_factory=dict, init=False)
|
|
|
|
def should_forward(self, payload: dict[str, object]) -> bool:
|
|
message = payload["message"]
|
|
if not isinstance(message, dict):
|
|
return False
|
|
message_type = str(message["type"])
|
|
channel = int(message["channel"])
|
|
number = int(message["number"])
|
|
value = int(message["value"])
|
|
|
|
if self.channel_filter is not None and channel != self.channel_filter:
|
|
return False
|
|
|
|
key = (message_type, channel, number)
|
|
if self.suppress_duplicate_values and self._last_values.get(key) == value:
|
|
return False
|
|
|
|
if message_type == "control_change":
|
|
now_ms = monotonic() * 1000
|
|
last_sent = self._last_sent_at.get(key, 0.0)
|
|
if now_ms - last_sent < self.control_change_interval_ms:
|
|
return False
|
|
self._last_sent_at[key] = now_ms
|
|
|
|
self._last_values[key] = value
|
|
return True
|