from __future__ import annotations import hashlib import hmac import logging import secrets from collections import deque from dataclasses import dataclass from datetime import UTC, datetime, timedelta from fnmatch import fnmatch from time import monotonic from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from app.core.database import SessionLocal from app.dmx.engine import DmxEngine from app.dmx.frame import FrameLayer from app.effects.service import EffectService from app.models.entities import MidiBridge, MidiBridgeToken, MidiMapping, Scene from app.scenes.service import SceneService logger = logging.getLogger(__name__) MIDI_PROTOCOL_VERSION = 1 MIDI_SCOPES = ["midi:connect", "midi:events", "midi:heartbeat"] MIDI_ACTIONS = { "activate_scene", "deactivate_scene", "toggle_scene", "flash_scene", "blackout_on", "blackout_off", "blackout_toggle", "set_master_dimmer", "set_scene_intensity", "start_effect", "stop_effect", } MIDI_MODES = {"trigger", "toggle", "hold", "flash", "continuous"} SCENE_ACTIONS = { "activate_scene", "deactivate_scene", "toggle_scene", "flash_scene", "set_scene_intensity", } EFFECT_ACTIONS = {"start_effect", "stop_effect"} def utc_now() -> datetime: return datetime.now(UTC) @dataclass(slots=True) class TokenCacheEntry: id: int label: str token_hash: str bridge_id: str | None scopes: list[str] created_at: datetime revoked_at: datetime | None last_used_at: datetime | None @dataclass(slots=True) class LearnState: active: bool = False allow_passthrough: bool = False expires_at: datetime | None = None captured_event: dict[str, object] | None = None class MidiService: def __init__( self, engine: DmxEngine, scenes: SceneService, effects: EffectService, session_factory: async_sessionmaker[AsyncSession] = SessionLocal, ) -> None: self.engine = engine self.scenes = scenes self.effects = effects self._session_factory = session_factory self._mappings: list[dict[str, object]] = [] self._tokens: list[TokenCacheEntry] = [] self._bridges: dict[str, dict[str, object]] = {} self._bridge_event_windows: dict[str, deque[float]] = {} self._bridge_persist_deadline: dict[str, float] = {} self._toggle_state: dict[int, bool] = {} self._hold_state: dict[int, bool] = {} self._learn_state = LearnState() async def startup(self) -> None: await self.reload() async def reload(self) -> None: await self._reload_mappings() await self._reload_tokens() await self._reload_bridges() async def list_mappings(self) -> dict[str, object]: return {"items": [dict(item) for item in self._mappings]} async def get_mapping(self, mapping_id: int) -> dict[str, object]: mapping = self._mapping_by_id(mapping_id) if mapping is None: raise LookupError("MIDI-mapping ikke fundet") return dict(mapping) async def create_mapping(self, payload: dict[str, object]) -> dict[str, object]: record = MidiMapping( name=str(payload["name"]).strip(), enabled=bool(payload.get("enabled", True)), bridge_id=self._normalize_optional_text(payload.get("bridge_id")), device_name=self._normalize_optional_text(payload.get("device_name")), message_type=str(payload["message_type"]).strip(), channel=self._optional_int(payload.get("channel")), number=int(payload["number"]), action=str(payload["action"]).strip(), target_type=str(payload.get("target_type") or "scene").strip(), target_id=self._normalize_optional_text(payload.get("target_id")), mode=str(payload.get("mode") or "trigger").strip(), minimum_value=int(payload.get("minimum_value", 0)), maximum_value=int(payload.get("maximum_value", 127)), updated_at=utc_now(), ) self._validate_mapping_record(record) async with self._session_factory() as session: session.add(record) await session.commit() await session.refresh(record) await self._reload_mappings() return self._serialize_mapping(record, None) async def update_mapping(self, mapping_id: int, payload: dict[str, object]) -> dict[str, object]: async with self._session_factory() as session: record = await session.get(MidiMapping, mapping_id) if record is None: raise LookupError("MIDI-mapping ikke fundet") record.name = str(payload["name"]).strip() record.enabled = bool(payload.get("enabled", True)) record.bridge_id = self._normalize_optional_text(payload.get("bridge_id")) record.device_name = self._normalize_optional_text(payload.get("device_name")) record.message_type = str(payload["message_type"]).strip() record.channel = self._optional_int(payload.get("channel")) record.number = int(payload["number"]) record.action = str(payload["action"]).strip() record.target_type = str(payload.get("target_type") or "scene").strip() record.target_id = self._normalize_optional_text(payload.get("target_id")) record.mode = str(payload.get("mode") or "trigger").strip() record.minimum_value = int(payload.get("minimum_value", 0)) record.maximum_value = int(payload.get("maximum_value", 127)) record.updated_at = utc_now() self._validate_mapping_record(record) await session.commit() await session.refresh(record) await self._reload_mappings() return self._mapping_by_id(mapping_id) or self._serialize_mapping(record, None) async def delete_mapping(self, mapping_id: int) -> dict[str, object]: async with self._session_factory() as session: record = await session.get(MidiMapping, mapping_id) if record is None: raise LookupError("MIDI-mapping ikke fundet") await session.delete(record) await session.commit() self._toggle_state.pop(mapping_id, None) self._hold_state.pop(mapping_id, None) await self._reload_mappings() return {"deleted": mapping_id} async def create_token(self, payload: dict[str, object]) -> dict[str, object]: raw_token = f"tuxmidi_{secrets.token_urlsafe(32)}" record = MidiBridgeToken( label=str(payload["label"]).strip(), token_hash=self._hash_token(raw_token), bridge_id=self._normalize_optional_text(payload.get("bridge_id")), scopes=self._normalize_scopes(payload.get("scopes")), created_at=utc_now(), ) async with self._session_factory() as session: session.add(record) await session.commit() await session.refresh(record) await self._reload_tokens() serialized = self._serialize_token(record) serialized["token"] = raw_token return serialized async def list_tokens(self) -> dict[str, object]: return {"items": [self._serialize_token_entry(entry) for entry in self._tokens]} async def revoke_token(self, token_id: int) -> dict[str, object]: async with self._session_factory() as session: record = await session.get(MidiBridgeToken, token_id) if record is None: raise LookupError("MIDI-token ikke fundet") record.revoked_at = utc_now() await session.commit() await self._reload_tokens() return {"revoked": token_id} async def list_bridges(self) -> dict[str, object]: self._expire_bridge_presence() items = sorted(self._bridges.values(), key=lambda bridge: str(bridge["bridge_id"]).casefold()) return {"items": [dict(item) for item in items]} async def get_learn_state(self) -> dict[str, object]: self._expire_learn_state() return self._serialize_learn_state() async def start_learn(self, timeout_seconds: int, allow_passthrough: bool) -> dict[str, object]: self._learn_state = LearnState( active=True, allow_passthrough=allow_passthrough, expires_at=utc_now() + timedelta(seconds=timeout_seconds), captured_event=None, ) return self._serialize_learn_state() async def cancel_learn(self) -> dict[str, object]: self._learn_state.active = False self._learn_state.expires_at = None return self._serialize_learn_state() async def receive_heartbeat(self, payload: dict[str, object], remote_ip: str | None) -> dict[str, object]: if int(payload.get("protocol_version", 0)) != MIDI_PROTOCOL_VERSION: raise ValueError("Ugyldig MIDI protocol_version") bridge_id = str(payload["bridge_id"]).strip() bridge = self._touch_bridge( bridge_id=bridge_id, device_name=self._normalize_optional_text(payload.get("device")), remote_ip=remote_ip, protocol_version=int(payload.get("protocol_version", MIDI_PROTOCOL_VERSION)), is_event=False, event=None, error=None, ) await self._persist_bridge(bridge_id, force=True) return { "accepted": True, "bridge_id": bridge_id, "online": bool(bridge["online"]), "protocol_version": MIDI_PROTOCOL_VERSION, } async def receive_event(self, payload: dict[str, object], remote_ip: str | None) -> dict[str, object]: if int(payload.get("protocol_version", 0)) != MIDI_PROTOCOL_VERSION: raise ValueError("Ugyldig MIDI protocol_version") bridge_id = str(payload["bridge_id"]).strip() event = self._normalize_event(payload) bridge = self._touch_bridge( bridge_id=bridge_id, device_name=str(payload.get("device") or "").strip(), remote_ip=remote_ip, protocol_version=int(payload.get("protocol_version", MIDI_PROTOCOL_VERSION)), is_event=True, event=event, error=None, ) if not self._check_bridge_rate_limit(bridge_id): bridge["last_error"] = "For mange MIDI-events fra bridgen." await self._persist_bridge(bridge_id, force=True) return {"accepted": False, "reason": "rate-limited"} if self._learn_state.active and not self._learn_state.allow_passthrough: self._capture_learn_event(bridge, event) await self._persist_bridge(bridge_id) return {"accepted": True, "captured_for_learning": True, "matched": 0} matches = [mapping for mapping in self._mappings if self._mapping_matches(mapping, bridge, event)] executed = 0 errors: list[str] = [] for mapping in matches: try: if self._learn_state.active and self._learn_state.captured_event is None: self._capture_learn_event(bridge, event) performed = await self._dispatch_mapping(mapping, event) executed += 1 if performed else 0 except Exception as exc: error_message = f"Mapping {mapping['id']} fejlede: {exc}" errors.append(error_message) bridge["last_error"] = error_message logger.warning("MIDI mapping failed: %s", error_message) await self._persist_bridge(bridge_id, force=bool(errors)) return { "accepted": True, "bridge_id": bridge_id, "matched": len(matches), "executed": executed, "errors": errors, } async def test_mapping(self, mapping_id: int, value: int, message_type: str | None = None) -> dict[str, object]: mapping = self._mapping_by_id(mapping_id) if mapping is None: raise LookupError("MIDI-mapping ikke fundet") test_event = { "type": message_type or str(mapping["message_type"]), "channel": int(mapping["channel"] or 0), "number": int(mapping["number"]), "value": value, } await self._dispatch_mapping(mapping, test_event) return { "status": "sent", "mapping_id": mapping_id, "service": mapping["action"], "value": value, } async def verify_token(self, token: str | None, required_scope: str, bridge_id: str | None = None) -> TokenCacheEntry: if not token: raise PermissionError("Mangler Bearer-token") token_hash = self._hash_token(token) for entry in self._tokens: if not hmac.compare_digest(entry.token_hash, token_hash): continue if entry.revoked_at is not None: break if required_scope not in entry.scopes: break if entry.bridge_id and bridge_id and entry.bridge_id != bridge_id: break await self._mark_token_used(entry.id) return entry raise PermissionError("Ugyldig eller tilbagekaldt MIDI-token") async def _dispatch_mapping(self, mapping: dict[str, object], event: dict[str, object]) -> bool: action = str(mapping["action"]) mode = str(mapping["mode"]) target_id = self._normalize_optional_text(mapping.get("target_id")) value = int(event["value"]) normalized = self._normalize_value(value, int(mapping["minimum_value"]), int(mapping["maximum_value"])) message_type = str(event["type"]) if action == "activate_scene" and target_id: if mode == "hold": if message_type == "note_on": await self.scenes.activate(target_id) self._hold_state[int(mapping["id"])] = True return True if message_type == "note_off": await self.scenes.release(target_id) self._hold_state[int(mapping["id"])] = False return True return False if message_type == "note_off": return False await self.scenes.activate(target_id) return True if action == "deactivate_scene" and target_id: if message_type == "note_off" and mode == "hold": return False await self.scenes.release(target_id) return True if action == "toggle_scene" and target_id: if message_type == "note_off": return False if self.engine.layers.get(f"scene:{target_id}") is not None: await self.scenes.release(target_id) self._toggle_state[int(mapping["id"])] = False else: await self.scenes.activate(target_id) self._toggle_state[int(mapping["id"])] = True return True if action == "flash_scene" and target_id: layer_name = f"midi:flash:{mapping['id']}:{target_id}" if message_type == "note_on": layer = await self._build_scene_layer( target_id, layer_name=layer_name, scale=1.0, priority_override=90, ) self.engine.set_layer(layer) self._hold_state[int(mapping["id"])] = True return True if message_type == "note_off": self.engine.remove_layer(layer_name) self._hold_state[int(mapping["id"])] = False return True return False if action == "blackout_on": if message_type == "note_off" and mode != "continuous": return False self.engine.trigger_blackout() return True if action == "blackout_off": if message_type == "note_off" and mode != "continuous": return False self.engine.release_blackout() return True if action == "blackout_toggle": if message_type == "note_off": return False if self.engine.blackout: self.engine.release_blackout() self._toggle_state[int(mapping["id"])] = False else: self.engine.trigger_blackout() self._toggle_state[int(mapping["id"])] = True return True if action == "set_master_dimmer": self.engine.master = max(0, min(255, int(round(normalized * 255)))) return True if action == "set_scene_intensity" and target_id: layer_name = f"midi:intensity:{mapping['id']}:{target_id}" if normalized <= 0: self.engine.remove_layer(layer_name) return True layer = await self._build_scene_layer( target_id, layer_name=layer_name, scale=normalized, priority_override=80, ) self.engine.set_layer(layer) return True if action == "start_effect" and target_id: if mode == "hold": if message_type == "note_on": self.effects.trigger(target_id) self._hold_state[int(mapping["id"])] = True return True if message_type == "note_off": self.effects.stop(target_id) self._hold_state[int(mapping["id"])] = False return True return False if mode == "toggle": if message_type == "note_off": return False if target_id in self.effects.active_effects: self.effects.stop(target_id) self._toggle_state[int(mapping["id"])] = False else: self.effects.trigger(target_id) self._toggle_state[int(mapping["id"])] = True return True if message_type == "note_off": return False self.effects.trigger(target_id) return True if action == "stop_effect" and target_id: if message_type == "note_off" and mode != "continuous": return False self.effects.stop(target_id) return True raise ValueError(f"Ukendt eller ikke-understøttet MIDI-action: {action}") async def _build_scene_layer( self, slug: str, layer_name: str, scale: float, priority_override: int | None = None, ) -> FrameLayer: scene, values_by_universe, precedence_by_universe = await self._resolve_scene_layer_data(slug) scaled_values: dict[int, dict[int, int]] = {} for universe, values in values_by_universe.items(): scaled_values[universe] = { channel: max(0, min(255, int(round(value * scale)))) for channel, value in values.items() } return FrameLayer( name=layer_name, priority=priority_override if priority_override is not None else int(scene.priority), values_by_universe=scaled_values, precedence_map_by_universe=precedence_by_universe, ) async def _resolve_scene_layer_data( self, slug: str, ) -> tuple[Scene, dict[int, dict[int, int]], dict[int, dict[int, str]]]: async with self._session_factory() as session: result = await session.execute(select(Scene).where(Scene.slug == slug).limit(1)) scene = result.scalar_one_or_none() if scene is None: raise LookupError("Scene ikke fundet") values_by_universe, precedence_by_universe = await self.scenes._resolve_layer_data(scene) return scene, values_by_universe, precedence_by_universe async def _reload_mappings(self) -> None: async with self._session_factory() as session: result = await session.execute(select(MidiMapping).order_by(MidiMapping.created_at, MidiMapping.id)) self._mappings = [self._serialize_mapping(record, None) for record in result.scalars().all()] async def _reload_tokens(self) -> None: async with self._session_factory() as session: result = await session.execute(select(MidiBridgeToken).order_by(MidiBridgeToken.created_at, MidiBridgeToken.id)) self._tokens = [ TokenCacheEntry( id=record.id, label=record.label, token_hash=record.token_hash, bridge_id=record.bridge_id, scopes=list(record.scopes or MIDI_SCOPES), created_at=record.created_at, revoked_at=record.revoked_at, last_used_at=record.last_used_at, ) for record in result.scalars().all() ] async def _reload_bridges(self) -> None: async with self._session_factory() as session: result = await session.execute(select(MidiBridge).order_by(MidiBridge.bridge_id)) self._bridges = {record.bridge_id: self._serialize_bridge(record) for record in result.scalars().all()} async def _mark_token_used(self, token_id: int) -> None: now = utc_now() for entry in self._tokens: if entry.id == token_id: entry.last_used_at = now break async with self._session_factory() as session: record = await session.get(MidiBridgeToken, token_id) if record is None: return record.last_used_at = now await session.commit() def _mapping_matches(self, mapping: dict[str, object], bridge: dict[str, object], event: dict[str, object]) -> bool: if not bool(mapping["enabled"]): return False mapping_message_type = str(mapping["message_type"]) event_type = str(event["type"]) if mapping_message_type != event_type: if not ( mapping_message_type == "note_on" and event_type == "note_off" and str(mapping["mode"]) in {"hold", "flash"} ): return False if int(mapping["number"]) != int(event["number"]): return False channel = mapping.get("channel") if channel is not None and int(channel) != int(event["channel"]): return False if not self._wildcard_match(mapping.get("bridge_id"), bridge.get("bridge_id")): return False if not self._wildcard_match(mapping.get("device_name"), bridge.get("device_name")): return False return True def _touch_bridge( self, bridge_id: str, device_name: str | None, remote_ip: str | None, protocol_version: int, is_event: bool, event: dict[str, object] | None, error: str | None, ) -> dict[str, object]: now = utc_now() bridge = self._bridges.get(bridge_id) or { "bridge_id": bridge_id, "device_name": device_name or "", "ip_address": remote_ip, "protocol_version": protocol_version, "online": True, "last_heartbeat_at": None, "last_event_at": None, "last_error": None, "last_event": None, "updated_at": now.isoformat(), "created_at": now.isoformat(), } bridge["device_name"] = device_name or bridge.get("device_name") or "" bridge["ip_address"] = remote_ip or bridge.get("ip_address") bridge["protocol_version"] = protocol_version bridge["online"] = True bridge["last_heartbeat_at"] = now.isoformat() bridge["updated_at"] = now.isoformat() if is_event: bridge["last_event_at"] = now.isoformat() bridge["last_event"] = event if error is not None: bridge["last_error"] = error self._bridges[bridge_id] = bridge return bridge async def _persist_bridge(self, bridge_id: str, force: bool = False) -> None: now = monotonic() previous = self._bridge_persist_deadline.get(bridge_id, 0.0) if not force and now - previous < 2.0: return self._bridge_persist_deadline[bridge_id] = now bridge = self._bridges.get(bridge_id) if bridge is None: return async with self._session_factory() as session: result = await session.execute(select(MidiBridge).where(MidiBridge.bridge_id == bridge_id).limit(1)) record = result.scalar_one_or_none() if record is None: record = MidiBridge( bridge_id=bridge_id, device_name=self._normalize_optional_text(bridge.get("device_name")), ip_address=self._normalize_optional_text(bridge.get("ip_address")), protocol_version=int(bridge.get("protocol_version", MIDI_PROTOCOL_VERSION)), online=bool(bridge.get("online", True)), created_at=utc_now(), updated_at=utc_now(), ) session.add(record) record.device_name = self._normalize_optional_text(bridge.get("device_name")) record.ip_address = self._normalize_optional_text(bridge.get("ip_address")) record.protocol_version = int(bridge.get("protocol_version", MIDI_PROTOCOL_VERSION)) record.online = bool(bridge.get("online", True)) record.last_heartbeat_at = self._parse_dt(bridge.get("last_heartbeat_at")) record.last_event_at = self._parse_dt(bridge.get("last_event_at")) record.last_error = self._normalize_optional_text(bridge.get("last_error")) record.last_event = dict(bridge.get("last_event") or {}) record.updated_at = utc_now() await session.commit() def _check_bridge_rate_limit(self, bridge_id: str) -> bool: window = self._bridge_event_windows.setdefault(bridge_id, deque()) now = monotonic() while window and now - window[0] > 1.0: window.popleft() window.append(now) return len(window) <= 200 def _expire_bridge_presence(self) -> None: threshold = utc_now() - timedelta(seconds=15) for bridge in self._bridges.values(): last_heartbeat = self._parse_dt(bridge.get("last_heartbeat_at")) bridge["online"] = bool(last_heartbeat and last_heartbeat >= threshold) def _expire_learn_state(self) -> None: if not self._learn_state.active or self._learn_state.expires_at is None: return if self._learn_state.expires_at < utc_now(): self._learn_state.active = False def _capture_learn_event(self, bridge: dict[str, object], event: dict[str, object]) -> None: self._learn_state.captured_event = { "bridge_id": bridge.get("bridge_id"), "device": bridge.get("device_name"), "timestamp": utc_now().isoformat(), "message": dict(event), } self._learn_state.active = False def _serialize_learn_state(self) -> dict[str, object]: return { "active": self._learn_state.active, "allow_passthrough": self._learn_state.allow_passthrough, "expires_at": self._learn_state.expires_at.isoformat() if self._learn_state.expires_at else None, "captured_event": self._learn_state.captured_event, } def _serialize_mapping(self, record: MidiMapping, runtime: dict[str, object] | None) -> dict[str, object]: return { "id": record.id, "name": record.name, "enabled": record.enabled, "bridge_id": record.bridge_id, "device_name": record.device_name, "message_type": record.message_type, "channel": record.channel, "number": record.number, "action": record.action, "target_type": record.target_type, "target_id": record.target_id, "mode": record.mode, "minimum_value": record.minimum_value, "maximum_value": record.maximum_value, "created_at": record.created_at.isoformat(), "updated_at": record.updated_at.isoformat(), "active": bool(runtime.get("active")) if runtime else self._mapping_is_active(record), } def _serialize_bridge(self, record: MidiBridge) -> dict[str, object]: return { "bridge_id": record.bridge_id, "device_name": record.device_name or "", "ip_address": record.ip_address, "protocol_version": record.protocol_version, "online": record.online, "last_heartbeat_at": record.last_heartbeat_at.isoformat() if record.last_heartbeat_at else None, "last_event_at": record.last_event_at.isoformat() if record.last_event_at else None, "last_error": record.last_error, "last_event": record.last_event or None, "updated_at": record.updated_at.isoformat(), "created_at": record.created_at.isoformat(), } def _serialize_token(self, record: MidiBridgeToken) -> dict[str, object]: return { "id": record.id, "label": record.label, "bridge_id": record.bridge_id, "scopes": list(record.scopes or MIDI_SCOPES), "created_at": record.created_at.isoformat(), "revoked_at": record.revoked_at.isoformat() if record.revoked_at else None, "last_used_at": record.last_used_at.isoformat() if record.last_used_at else None, "token_preview": self._token_preview(record.token_hash), } def _serialize_token_entry(self, entry: TokenCacheEntry) -> dict[str, object]: return { "id": entry.id, "label": entry.label, "bridge_id": entry.bridge_id, "scopes": list(entry.scopes), "created_at": entry.created_at.isoformat(), "revoked_at": entry.revoked_at.isoformat() if entry.revoked_at else None, "last_used_at": entry.last_used_at.isoformat() if entry.last_used_at else None, "token_preview": self._token_preview(entry.token_hash), } def _normalize_event(self, payload: dict[str, object]) -> dict[str, object]: message = payload.get("message") if not isinstance(message, dict): raise ValueError("MIDI payload mangler message") return { "type": str(message["type"]), "channel": int(message["channel"]), "number": int(message["number"]), "value": int(message["value"]), } def _mapping_is_active(self, record: MidiMapping) -> bool: target_id = record.target_id if record.action in SCENE_ACTIONS and target_id: if record.action == "set_scene_intensity": return self.engine.layers.get(f"midi:intensity:{record.id}:{target_id}") is not None if record.action == "flash_scene": return self.engine.layers.get(f"midi:flash:{record.id}:{target_id}") is not None return self.engine.layers.get(f"scene:{target_id}") is not None if record.action in EFFECT_ACTIONS and target_id: return target_id in self.effects.active_effects if record.action.startswith("blackout"): return self.engine.blackout if record.action == "set_master_dimmer": return self.engine.master > 0 return False def _mapping_by_id(self, mapping_id: int) -> dict[str, object] | None: for mapping in self._mappings: if int(mapping["id"]) == mapping_id: return mapping return None def _validate_mapping_record(self, record: MidiMapping) -> None: if record.action not in MIDI_ACTIONS: raise ValueError("Ukendt MIDI-action") if record.mode not in MIDI_MODES: raise ValueError("Ukendt MIDI-mode") if record.minimum_value > record.maximum_value: raise ValueError("minimum_value må ikke være større end maximum_value") if record.action in SCENE_ACTIONS and not record.target_id: raise ValueError("Scene-actions kræver target_id") if record.action in EFFECT_ACTIONS and not record.target_id: raise ValueError("Effect-actions kræver target_id") if record.action == "set_scene_intensity" and record.mode != "continuous": raise ValueError("set_scene_intensity kræver continuous-mode") if record.action == "set_master_dimmer" and record.mode != "continuous": raise ValueError("set_master_dimmer kræver continuous-mode") if record.action == "flash_scene" and record.mode not in {"hold", "flash"}: raise ValueError("flash_scene kræver hold eller flash-mode") if record.message_type == "control_change" and record.mode in {"hold", "flash"}: return def _normalize_value(self, raw: int, minimum: int, maximum: int) -> float: clamped = max(minimum, min(maximum, raw)) span = max(1, maximum - minimum) return max(0.0, min(1.0, (clamped - minimum) / span)) def _wildcard_match(self, pattern: object, value: object) -> bool: normalized_pattern = self._normalize_optional_text(pattern) if not normalized_pattern or normalized_pattern == "*": return True normalized_value = self._normalize_optional_text(value) if not normalized_value: return False return fnmatch(normalized_value.casefold(), normalized_pattern.casefold()) def _normalize_optional_text(self, value: object) -> str | None: if value is None: return None normalized = str(value).strip() return normalized or None def _optional_int(self, value: object) -> int | None: if value is None or value == "": return None return int(value) def _normalize_scopes(self, payload: object) -> list[str]: if not isinstance(payload, list): return list(MIDI_SCOPES) scopes = [str(scope).strip() for scope in payload if str(scope).strip()] return scopes or list(MIDI_SCOPES) def _hash_token(self, token: str) -> str: return hashlib.sha256(token.encode("utf-8")).hexdigest() def _token_preview(self, token_hash: str) -> str: return f"********{token_hash[-4:]}" def _parse_dt(self, value: object) -> datetime | None: if isinstance(value, datetime): return value if value.tzinfo is not None else value.replace(tzinfo=UTC) if isinstance(value, str) and value: parsed = datetime.fromisoformat(value) return parsed if parsed.tzinfo is not None else parsed.replace(tzinfo=UTC) return None