"""PyScript controller and authoritative Web Audio look-ahead scheduler.""" from pyscript import ffi, web, when, window # ty: ignore[unresolved-import] from sore.audio import AudioEngine from sore.constants import LOOKAHEAD_SECONDS, SCHEDULER_INTERVAL_MS, START_LEAD_SECONDS from sore.metronome import MetronomeLogic, MetronomeState from sore.ui import MetronomeUI class MetronomeApp: """Coordinate classroom controls, musical state, audio, and DOM visuals.""" def __init__(self) -> None: """Initialise the browser application and bind its input methods.""" self.logic = MetronomeLogic() self.state = MetronomeState() self.audio = AudioEngine() self.ui = MetronomeUI() self.main_accents = [False] * 8 self.intermediate_accents = [False] * 8 self.tap_times = [] self.run_bpm = 120 self.run_intermediate_enabled = False self.run_ball_enabled = False self.run_main_sound_enabled = True self.run_intermediate_sound_enabled = False self.run_loop_enabled = False self.count_in_events = [] self.next_event_time = 0.0 self.next_timeline_index = 0 self.current_event_audio_time = None self.loop_finish_start_step = None self.loop_finish_at_step = None self.finish_scheduled = False self.scheduler_interval_id = None self.visual_timers = {} self.event_handlers = [] self.starting = False self.start_request_token = 0 self.shortcut_help_open = False self.fallback_fullscreen_active = False self.bpm_capture_active = False self.bpm_capture_digits = "" self.bpm_hold_delta = 0 self.bpm_hold_timeout_id = None self.bpm_hold_interval_id = None self.eights_hold_delta = 0.0 self.eights_hold_timeout_id = None self.eights_hold_interval_id = None self.cursor_hide_timeout_id = None self.scheduler_proxy = ffi.create_proxy(self._scheduler_tick) self.bpm_hold_delay_proxy = ffi.create_proxy(self._begin_bpm_repeat) self.bpm_hold_repeat_proxy = ffi.create_proxy(self._repeat_bpm_adjustment) self.eights_hold_delay_proxy = ffi.create_proxy(self._begin_eights_repeat) self.eights_hold_repeat_proxy = ffi.create_proxy(self._repeat_eights_adjustment) self.cursor_hide_proxy = ffi.create_proxy(self._hide_cursor) self._bind_events() self.ui.set_loop_enabled(False) self.ui.set_intermediates_visible(False) self.ui.set_intermediate_sound_available(False) self.ui.set_bouncing_ball_visible(False) self._update_idle_counter() self.ui.set_ready() def _bind_events(self) -> None: @when("click", self.ui.start_button) async def start_clicked(_event): await self.start() @when("click", self.ui.stop_button) def stop_clicked(_event): self.stop() @when("click", self.ui.focus_stop_button) def focus_stop_clicked(_event): self.stop() @when("click", self.ui.focus_ball_button) def focus_ball_clicked(_event): self.ui.bouncing_ball_input.checked = not bool(self.ui.bouncing_ball_input.checked) self.on_ball_toggle() @when("click", self.ui.focus_intermediate_button) def focus_intermediates_clicked(_event): self.ui.intermediate_input.checked = not bool(self.ui.intermediate_input.checked) self.on_intermediate_toggle() @when("click", self.ui.focus_final_button) def focus_final_clicked(_event): self.arm_loop_finish() @when("click", self.ui.shortcut_help_button) def shortcut_help_clicked(_event): self.set_shortcut_help_open(not self.shortcut_help_open) @when("click", self.ui.shortcut_help_close_button) def shortcut_help_close_clicked(_event): self.set_shortcut_help_open(False) @when("click", self.ui.features_close_button) def features_close_clicked(_event): self.ui.features_menu.open = False @when("toggle", self.ui.features_menu) def features_toggled(_event): opened = bool(self.ui.features_menu.open) if opened: self.set_shortcut_help_open(False) self.ui.set_features_open(opened) @when("click", self.ui.reset_button) def reset_clicked(_event): self.reset() @when("click", self.ui.tap_button) def tap_clicked(_event): self.tap_tempo() @when("click", self.ui.fullscreen_button) async def fullscreen_clicked(_event): await self.toggle_fullscreen() @when("fullscreenchange", window.document) def fullscreen_changed(_event): self._sync_fullscreen_state() @when("webkitfullscreenchange", window.document) def webkit_fullscreen_changed(_event): self._sync_fullscreen_state() @when("change", self.ui.loop_input) def loop_changed(_event): self.on_loop_toggle() @when("change", self.ui.count_in_input) def count_in_changed(_event): self.on_count_in_toggle() @when("change", self.ui.intermediate_input) def intermediates_changed(_event): self.on_intermediate_toggle() @when("change", self.ui.bouncing_ball_input) def bouncing_ball_changed(_event): self.on_ball_toggle() @when("change", self.ui.main_sound_input) def main_sound_changed(_event): self.on_sound_toggle() @when("change", self.ui.intermediate_sound_input) def intermediate_sound_changed(_event): self.on_sound_toggle() @when("input", self.ui.eights_input) def eights_changed(_event): self._update_idle_counter() @when("pointerdown", self.ui.bpm_decrease_button) def bpm_decrease_pressed(event): event.preventDefault() self._start_bpm_adjustment(-1) @when("pointerdown", self.ui.bpm_increase_button) def bpm_increase_pressed(event): event.preventDefault() self._start_bpm_adjustment(1) @when("click", self.ui.bpm_decrease_button) def bpm_decrease_accessible_clicked(event): if int(getattr(event, "detail", 0)) == 0: self.adjust_bpm(-1) @when("click", self.ui.bpm_increase_button) def bpm_increase_accessible_clicked(event): if int(getattr(event, "detail", 0)) == 0: self.adjust_bpm(1) @when("pointerdown", self.ui.eights_decrease_button) def eights_decrease_pressed(event): event.preventDefault() self._start_eights_adjustment(-0.5) @when("pointerdown", self.ui.eights_increase_button) def eights_increase_pressed(event): event.preventDefault() self._start_eights_adjustment(0.5) @when("click", self.ui.eights_decrease_button) def eights_decrease_accessible_clicked(event): if int(getattr(event, "detail", 0)) == 0: self.adjust_eights(-0.5) @when("click", self.ui.eights_increase_button) def eights_increase_accessible_clicked(event): if int(getattr(event, "detail", 0)) == 0: self.adjust_eights(0.5) @when("pointerup", web.page.body) def adjustment_pointer_released(_event): self._stop_bpm_adjustment() self._stop_eights_adjustment() @when("pointercancel", web.page.body) def adjustment_pointer_cancelled(_event): self._stop_bpm_adjustment() self._stop_eights_adjustment() @when("pointerleave", self.ui.bpm_decrease_button) def bpm_decrease_left(_event): self._stop_bpm_adjustment() @when("pointerleave", self.ui.bpm_increase_button) def bpm_increase_left(_event): self._stop_bpm_adjustment() @when("pointerleave", self.ui.eights_decrease_button) def eights_decrease_left(_event): self._stop_eights_adjustment() @when("pointerleave", self.ui.eights_increase_button) def eights_increase_left(_event): self._stop_eights_adjustment() @when("pointermove", web.page.body) def pointer_moved(_event): if self.state.running: self.ui.set_cursor_hidden(False) self._schedule_cursor_hide() @when("keydown", web.page.body) async def key_pressed(event): await self._on_keydown(event) @when("keyup", web.page.body) def key_released(event): self._on_keyup(event) self.event_handlers.extend( [ start_clicked, stop_clicked, focus_stop_clicked, focus_ball_clicked, focus_intermediates_clicked, focus_final_clicked, shortcut_help_clicked, shortcut_help_close_clicked, features_close_clicked, features_toggled, reset_clicked, tap_clicked, fullscreen_clicked, fullscreen_changed, webkit_fullscreen_changed, loop_changed, count_in_changed, intermediates_changed, bouncing_ball_changed, main_sound_changed, intermediate_sound_changed, eights_changed, bpm_decrease_pressed, bpm_increase_pressed, bpm_decrease_accessible_clicked, bpm_increase_accessible_clicked, eights_decrease_pressed, eights_increase_pressed, eights_decrease_accessible_clicked, eights_increase_accessible_clicked, adjustment_pointer_released, adjustment_pointer_cancelled, bpm_decrease_left, bpm_increase_left, eights_decrease_left, eights_increase_left, pointer_moved, key_pressed, key_released, ] ) for index, beat in enumerate(self.ui.main_beats): handler = self._accent_handler("main", index) self.event_handlers.append(when("click", beat)(handler)) for index, beat in enumerate(self.ui.intermediate_beats): handler = self._accent_handler("inter", index) self.event_handlers.append(when("click", beat)(handler)) def _accent_handler(self, kind: str, index: int): def toggle_accent(_event): self.toggle_accent(kind, index) return toggle_accent @staticmethod def _event_is_from_editor(event) -> bool: target = event.target tag_name = str(getattr(target, "tagName", "")).upper() return tag_name in ("BUTTON", "INPUT", "TEXTAREA", "SELECT", "SUMMARY") or bool( getattr(target, "isContentEditable", False) ) async def _on_keydown(self, event) -> None: key = str(event.key) if key == "Escape": self.set_shortcut_help_open(False) await self.set_fullscreen(False) return if self._event_is_from_editor(event): return if key in ("Meta", "Control"): if not bool(getattr(event, "repeat", False)): self.bpm_capture_active = True self.bpm_capture_digits = "" return modifier_held = bool(event.metaKey) or bool(event.ctrlKey) or self.bpm_capture_active if modifier_held: lower_key = key.lower() if lower_key in ("l", "c", "i", "b"): if not bool(getattr(event, "repeat", False)): self._force_feature_off(lower_key) event.preventDefault() return if len(key) == 1 and "0" <= key <= "9" and len(self.bpm_capture_digits) < 3: if not bool(getattr(event, "repeat", False)): self.bpm_capture_digits += key event.preventDefault() return lower_key = key.lower() handled = True if key == " ": if self.state.running or self.starting: self.stop() else: await self.start() elif lower_key == "r": self.reset() elif lower_key == "f": await self.toggle_fullscreen() elif lower_key == "t": self.tap_tempo() elif lower_key == "l" and bool(event.shiftKey) and self.state.running: self.arm_loop_finish() elif lower_key == "l" and not self.state.running and not self.starting: self.ui.loop_input.checked = not bool(self.ui.loop_input.checked) self.on_loop_toggle() elif lower_key == "c" and not self.state.running and not self.starting: self.ui.count_in_input.checked = not bool(self.ui.count_in_input.checked) self.on_count_in_toggle() elif lower_key == "i": self.ui.intermediate_input.checked = not bool(self.ui.intermediate_input.checked) self.on_intermediate_toggle() elif lower_key == "b": self.ui.bouncing_ball_input.checked = not bool(self.ui.bouncing_ball_input.checked) self.on_ball_toggle() elif key == "ArrowUp": self.adjust_bpm(1) elif key == "ArrowDown": self.adjust_bpm(-1) elif key == "ArrowRight": self.adjust_eights(0.5) elif key == "ArrowLeft": self.adjust_eights(-0.5) elif key == "?": self.set_shortcut_help_open(not self.shortcut_help_open) else: handled = False if handled: event.preventDefault() def _on_keyup(self, event) -> None: key = str(event.key) if key not in ("Meta", "Control") or not self.bpm_capture_active: return digits = self.bpm_capture_digits self.bpm_capture_active = False self.bpm_capture_digits = "" if not digits: return try: self.set_bpm(self.logic.parse_bpm(digits)) self.ui.show_message("") except ValueError as exc: self.ui.show_message(str(exc)) def set_shortcut_help_open(self, opened: bool) -> None: """Open or close the corner shortcut reference. Args: opened: Whether the reference should stay open. """ self.shortcut_help_open = opened self.ui.set_shortcut_help_open(opened) def on_loop_toggle(self) -> None: """Apply loop-control availability and update the idle counter.""" self.ui.set_loop_enabled(bool(self.ui.loop_input.checked)) self._update_idle_counter() def on_count_in_toggle(self) -> None: """Refresh the compact feature summary after changing count-in.""" self.ui.update_features_summary() def on_intermediate_toggle(self) -> None: """Show or hide subdivisions without changing the musical timeline.""" visible = bool(self.ui.intermediate_input.checked) if not visible: self.ui.intermediate_sound_input.checked = False self.run_intermediate_sound_enabled = False self.run_intermediate_enabled = visible self.ui.set_intermediates_visible(visible) self.ui.set_intermediate_sound_available(visible) self.ui.update_features_summary() if self.state.running: if self.state.in_count_in: _kind, index, _phrase = self.count_in_events[self.state.count_in_step] self.ui.show_count_in(index) else: kind, index = self._step_visual(self.state.current_step) if kind == "main" or visible: self.ui.highlight(kind, index) else: self.ui.highlight("main", index) self._update_idle_counter() def on_ball_toggle(self) -> None: """Show or hide the bouncing beat visual without affecting timing.""" visible = bool(self.ui.bouncing_ball_input.checked) self.run_ball_enabled = visible self.ui.set_bouncing_ball_visible(visible) self.ui.update_features_summary() def on_sound_toggle(self) -> None: """Apply valid audio options and refresh the compact feature summary.""" if bool(self.ui.intermediate_sound_input.checked) and not bool( self.ui.intermediate_input.checked ): self.ui.intermediate_sound_input.checked = False self.run_main_sound_enabled = bool(self.ui.main_sound_input.checked) self.run_intermediate_sound_enabled = bool(self.ui.intermediate_sound_input.checked) self.ui.update_features_summary() def _force_feature_off(self, key: str) -> None: if key == "l": if not self.state.running and not self.starting: self.ui.loop_input.checked = False self.on_loop_toggle() elif key == "c": self.ui.count_in_input.checked = False self.on_count_in_toggle() elif key == "i": self.ui.intermediate_input.checked = False self.on_intermediate_toggle() else: self.ui.bouncing_ball_input.checked = False self.on_ball_toggle() def arm_loop_finish(self) -> None: """Queue one complete final group after the current loop group.""" if ( not self.state.running or not self.run_loop_enabled or self.loop_finish_at_step is not None ): return current_step = max(0, self.state.current_step) self.loop_finish_start_step, self.loop_finish_at_step = ( self.logic.queued_loop_ending_bounds(current_step, self.state.in_count_in) ) self.ui.set_loop_finish_armed(True, pending=True) self.ui.set_last_one(False) self.ui.set_counter_message("LOOP") self.ui.set_runtime_status("Finishing this 8 · Last One starts next") if self.current_event_audio_time is None: return self.audio.cancel_future() self._cancel_visual_callbacks() if self.state.in_count_in: current_timeline_index = self.state.count_in_step else: current_timeline_index = len(self.count_in_events) + current_step self.next_timeline_index = current_timeline_index + 1 self.next_event_time = self.current_event_audio_time + self.logic.interval_seconds( self.run_bpm, True ) self.finish_scheduled = False self._scheduler_tick() def set_bpm(self, bpm: int) -> None: """Update the single BPM state used by every input method. Args: bpm: Positive whole-number tempo. """ bounded_bpm = min(999, max(1, bpm)) self.ui.bpm_input.value = str(bounded_bpm) self.run_bpm = bounded_bpm def _start_bpm_adjustment(self, delta: int) -> None: self._stop_bpm_adjustment() self.bpm_hold_delta = delta self.adjust_bpm(delta) self.bpm_hold_timeout_id = window.setTimeout(self.bpm_hold_delay_proxy, 420) def _begin_bpm_repeat(self) -> None: self.bpm_hold_timeout_id = None if self.bpm_hold_delta == 0: return self.bpm_hold_interval_id = window.setInterval(self.bpm_hold_repeat_proxy, 85) def _repeat_bpm_adjustment(self) -> None: if self.bpm_hold_delta != 0: self.adjust_bpm(self.bpm_hold_delta) def _stop_bpm_adjustment(self) -> None: if self.bpm_hold_timeout_id is not None: window.clearTimeout(self.bpm_hold_timeout_id) self.bpm_hold_timeout_id = None if self.bpm_hold_interval_id is not None: window.clearInterval(self.bpm_hold_interval_id) self.bpm_hold_interval_id = None self.bpm_hold_delta = 0 def adjust_bpm(self, delta: int) -> None: """Adjust BPM from the keyboard while tolerating invalid field text. Args: delta: Signed whole-number change. """ try: current = self.logic.parse_bpm(str(self.ui.bpm_input.value)) except ValueError: current = 120 self.set_bpm(current + delta) def adjust_eights(self, delta: float) -> None: """Adjust finite exercise length by a half-eight. Args: delta: Signed 0.5 increment. """ if bool(self.ui.loop_input.checked): return try: current = self.logic.parse_eights(str(self.ui.eights_input.value)) except ValueError: current = 4.0 rounded = round(max(0.5, current + delta) * 2) / 2 self.ui.eights_input.value = self.logic.format_eights(rounded) self._update_idle_counter() def _start_eights_adjustment(self, delta: float) -> None: self._stop_eights_adjustment() self.eights_hold_delta = delta self.adjust_eights(delta) self.eights_hold_timeout_id = window.setTimeout(self.eights_hold_delay_proxy, 420) def _begin_eights_repeat(self) -> None: self.eights_hold_timeout_id = None if self.eights_hold_delta == 0: return self.eights_hold_interval_id = window.setInterval(self.eights_hold_repeat_proxy, 120) def _repeat_eights_adjustment(self) -> None: if self.eights_hold_delta != 0: self.adjust_eights(self.eights_hold_delta) def _stop_eights_adjustment(self) -> None: if self.eights_hold_timeout_id is not None: window.clearTimeout(self.eights_hold_timeout_id) self.eights_hold_timeout_id = None if self.eights_hold_interval_id is not None: window.clearInterval(self.eights_hold_interval_id) self.eights_hold_interval_id = None self.eights_hold_delta = 0.0 def tap_tempo(self) -> None: """Calculate BPM from recent high-resolution browser tap timestamps.""" now = float(window.performance.now()) / 1000.0 self.tap_times, bpm = self.logic.tap_tempo(self.tap_times, now) if bpm is not None: self.set_bpm(bpm) async def start(self) -> None: """Validate configuration and start a fresh audio-clock playback run.""" if self.state.running or self.starting: return try: bpm = self.logic.parse_bpm(str(self.ui.bpm_input.value)) intermediate_enabled = bool(self.ui.intermediate_input.checked) ball_enabled = bool(self.ui.bouncing_ball_input.checked) main_sound_enabled = bool(self.ui.main_sound_input.checked) intermediate_sound_enabled = intermediate_enabled and bool( self.ui.intermediate_sound_input.checked ) loop_enabled = bool(self.ui.loop_input.checked) total_steps = ( 0 if loop_enabled else self.logic.parse_total_steps(str(self.ui.eights_input.value), True) ) except ValueError as exc: self.ui.show_message(str(exc)) return self.starting = True self.start_request_token += 1 request_token = self.start_request_token self.ui.show_message("") self.ui.set_runtime_status("Preparing audio…") try: await self.audio.prepare() except RuntimeError as exc: self.starting = False self.ui.set_runtime_status("Audio unavailable") self.ui.show_message(str(exc)) return if request_token != self.start_request_token: self.starting = False return self.starting = False self.run_bpm = bpm self.run_intermediate_enabled = intermediate_enabled self.run_ball_enabled = ball_enabled self.run_main_sound_enabled = main_sound_enabled self.run_intermediate_sound_enabled = intermediate_sound_enabled self.ui.set_bouncing_ball_visible(ball_enabled) self.run_loop_enabled = loop_enabled self.count_in_events = ( self.logic.count_in_sequence(True) if bool(self.ui.count_in_input.checked) else [] ) generation = self.state.begin(total_steps, bool(self.count_in_events)) self.next_timeline_index = 0 self.next_event_time = self.audio.current_time + START_LEAD_SECONDS self.current_event_audio_time = None self.loop_finish_start_step = None self.loop_finish_at_step = None self.finish_scheduled = False self.audio.cancel_scheduled() self._cancel_visual_callbacks() self.ui.clear_highlight() self.ui.clear_count_in() self.ui.set_progress(0) self.ui.stop_bouncing_ball() self.ui.set_last_one(False) self.ui.set_loop_finish_armed(False) self.set_shortcut_help_open(False) self.ui.set_playing(True) self._schedule_cursor_hide() self.ui.set_runtime_status("Playing · Web Audio clock") self._update_running_counter(0) self._scheduler_tick() if generation == self.state.playback_generation and self.state.running: self.scheduler_interval_id = window.setInterval( self.scheduler_proxy, SCHEDULER_INTERVAL_MS ) def stop(self) -> None: """Stop promptly while preserving the currently highlighted beat.""" self.start_request_token += 1 self.starting = False self.state.stop() self._stop_scheduler() self._cancel_visual_callbacks() self.audio.cancel_scheduled() self.ui.clear_count_in() self.ui.stop_bouncing_ball() self.loop_finish_start_step = None self.loop_finish_at_step = None self.current_event_audio_time = None self.ui.set_last_one(False) self.ui.set_loop_finish_armed(False) self.ui.set_playing(False) self._stop_cursor_hiding() self.ui.set_runtime_status("Stopped · current beat preserved") self._update_idle_counter() def reset(self) -> None: """Reset playback position without clearing teacher configuration.""" self.stop() self.state.reset() self.ui.set_progress(0) self.ui.clear_highlight() self.ui.clear_count_in() self.ui.set_runtime_status("Ready · reset") self._update_idle_counter() def toggle_accent(self, kind: str, index: int) -> None: """Toggle a configured visual accent. Args: kind: ``main`` or ``inter``. index: Zero-based beat index. """ accents = self.main_accents if kind == "main" else self.intermediate_accents accents[index] = not accents[index] self.ui.set_accent(kind, index, accents[index]) if self.state.running and not self.state.in_count_in: active_step = self.state.current_step if active_step >= 0: active_kind, active_index = self._step_visual(active_step) if active_kind == "main" or self.run_intermediate_enabled: self.ui.highlight(active_kind, active_index) else: self.ui.highlight("main", active_index) async def toggle_fullscreen(self) -> None: """Toggle native fullscreen or the in-page expanded-view fallback.""" if self._fullscreen_element() is not None or self.fallback_fullscreen_active: await self.set_fullscreen(False) else: await self.set_fullscreen(True) @staticmethod def _fullscreen_element(): document = window.document return getattr(document, "fullscreenElement", None) or getattr( document, "webkitFullscreenElement", None ) async def _await_browser_promise(self, result) -> None: if result is not None and callable(getattr(result, "then", None)): await result def _sync_fullscreen_state(self) -> None: native_active = self._fullscreen_element() is not None if native_active: self.fallback_fullscreen_active = False self.ui.set_fullscreen_state( native_active or self.fallback_fullscreen_active, self.fallback_fullscreen_active, ) async def set_fullscreen(self, enabled: bool) -> None: """Enter or leave browser fullscreen when available. Args: enabled: Whether fullscreen should be active. """ document = window.document if not enabled: self.fallback_fullscreen_active = False try: exit_fullscreen = getattr(document, "exitFullscreen", None) or getattr( document, "webkitExitFullscreen", None ) if self._fullscreen_element() is not None and exit_fullscreen is not None: await self._await_browser_promise(exit_fullscreen()) except Exception: pass self._sync_fullscreen_state() self.ui.show_message("") return root = document.documentElement request_fullscreen = getattr(root, "requestFullscreen", None) or getattr( root, "webkitRequestFullscreen", None ) if request_fullscreen is not None: try: await self._await_browser_promise(request_fullscreen()) self.fallback_fullscreen_active = False self._sync_fullscreen_state() self.ui.show_message("") return except Exception: pass self.fallback_fullscreen_active = True self.ui.set_fullscreen_state(True, True) self.ui.show_message("") window.scrollTo(0, 0) def _scheduler_tick(self) -> None: if not self.state.running or self.finish_scheduled: return generation = self.state.playback_generation now = self.audio.current_time horizon = now + LOOKAHEAD_SECONDS self.audio.prune_finished_sources() while self.next_event_time < horizon: if generation != self.state.playback_generation or not self.state.running: return timeline_index = self.next_timeline_index if timeline_index < len(self.count_in_events): kind, index, phrase = self.count_in_events[timeline_index] if kind == "main": self.audio.schedule("countin", self.next_event_time) self._schedule_visual( self.next_event_time, generation, "countin", (kind, index, phrase, timeline_index), ) else: main_step = timeline_index - len(self.count_in_events) reached_loop_finish = ( self.loop_finish_at_step is not None and main_step >= self.loop_finish_at_step ) reached_finite_finish = ( not self.run_loop_enabled and main_step >= self.state.total_steps ) if reached_loop_finish or reached_finite_finish: self.finish_scheduled = True self._schedule_visual(self.next_event_time, generation, "finish", ()) self._stop_scheduler() return kind, index = self._step_visual(main_step) if kind == "main" and self.run_main_sound_enabled: sound_type = "accent" if self.main_accents[index] else "normal" self.audio.schedule(sound_type, self.next_event_time) elif ( kind == "inter" and self.run_intermediate_enabled and self.run_intermediate_sound_enabled ): sound_type = ( "intermediate_accent" if self.intermediate_accents[index] else "intermediate" ) self.audio.schedule(sound_type, self.next_event_time) self._schedule_visual( self.next_event_time, generation, "main", (kind, index, main_step), ) self.next_timeline_index += 1 self.next_event_time += self.logic.interval_seconds(self.run_bpm, True) def _step_visual(self, step: int) -> tuple[str, int]: return self.logic.step_to_kind_and_index(step) def _schedule_visual( self, audio_time: float, generation: int, event_type: str, payload: tuple, ) -> None: delay_ms = max(0, round((audio_time - self.audio.current_time) * 1000)) timer_id = None def apply_visual() -> None: if timer_id in self.visual_timers: del self.visual_timers[timer_id] if generation != self.state.playback_generation or not self.state.running: return self.current_event_audio_time = audio_time if event_type == "countin": kind, index, phrase, count_in_step = payload self.state.in_count_in = True self.state.count_in_step = count_in_step if kind == "main": self.ui.show_count_in(index) if self.run_ball_enabled and index == 3: self.ui.prepare_ball( self.logic.interval_seconds(self.run_bpm, False), self.main_accents[0], ) self.ui.set_counter_message(phrase) elif event_type == "main": kind, index, main_step = payload self.state.in_count_in = False self.state.current_step = main_step self.ui.clear_count_in() if kind == "main" or self.run_intermediate_enabled: self.ui.highlight(kind, index) if kind == "main" and self.run_ball_enabled: next_index = (index + 1) % len(self.main_accents) self.ui.bounce_ball( self.logic.interval_seconds(self.run_bpm, False), self.main_accents[index], self.main_accents[next_index], ) self.ui.impact_floor( self.logic.interval_seconds(self.run_bpm, False), self.main_accents[index], ) if not self.run_loop_enabled: percentage = self.logic.progress_percent( main_step, self.state.total_steps, False ) self.ui.set_progress(percentage) self._update_running_counter(main_step) else: self._finish_playback() callback_proxy = ffi.create_proxy(apply_visual) timer_id = window.setTimeout(callback_proxy, delay_ms) self.visual_timers[timer_id] = callback_proxy def _stop_scheduler(self) -> None: if self.scheduler_interval_id is not None: window.clearInterval(self.scheduler_interval_id) self.scheduler_interval_id = None def _schedule_cursor_hide(self) -> None: if self.cursor_hide_timeout_id is not None: window.clearTimeout(self.cursor_hide_timeout_id) self.cursor_hide_timeout_id = window.setTimeout(self.cursor_hide_proxy, 1200) def _hide_cursor(self) -> None: self.cursor_hide_timeout_id = None if self.state.running: self.ui.set_cursor_hidden(True) def _stop_cursor_hiding(self) -> None: if self.cursor_hide_timeout_id is not None: window.clearTimeout(self.cursor_hide_timeout_id) self.cursor_hide_timeout_id = None self.ui.set_cursor_hidden(False) def _cancel_visual_callbacks(self) -> None: for timer_id in self.visual_timers: window.clearTimeout(timer_id) self.visual_timers = {} def _finish_playback(self) -> None: self.state.stop() self._stop_scheduler() self._cancel_visual_callbacks() self.audio.cancel_scheduled() self.ui.clear_count_in() self.ui.stop_bouncing_ball() self.ui.clear_highlight() if not self.run_loop_enabled: self.ui.set_progress(100) self.loop_finish_start_step = None self.loop_finish_at_step = None self.current_event_audio_time = None self.ui.set_last_one(False) self.ui.set_loop_finish_armed(False) self.ui.set_playing(False) self._stop_cursor_hiding() self.ui.set_runtime_status("Complete") self._update_idle_counter() def _update_idle_counter(self) -> None: if self.state.running: return self.ui.set_last_one(False) if bool(self.ui.loop_input.checked): self.ui.set_counter_message("LOOP") return try: eights = self.logic.parse_eights(str(self.ui.eights_input.value)) formatted_eights = self.logic.format_eights(eights) self.ui.set_counter_values("0", formatted_eights) self.ui.show_message("") except ValueError: self.ui.clear_counter() def _update_running_counter(self, current_step: int) -> None: if self.state.in_count_in: return if self.run_loop_enabled: final_group_started = ( self.loop_finish_start_step is not None and current_step >= self.loop_finish_start_step ) if final_group_started: self.ui.set_counter_message("LOOP") self.ui.set_last_one(True) self.ui.set_loop_finish_armed(True, pending=False) self.ui.set_runtime_status("Last one · stopping after this 8") else: self.ui.set_counter_message("LOOP") self.ui.set_last_one(False) return total_eights = self.logic.total_bars(self.state.total_steps, True) final_start = self.logic.final_group_start(self.state.total_steps) final_group = current_step >= final_start if final_group: current_text = self.logic.format_eights(total_eights) else: current_text = str(self.logic.current_bar(current_step, True)) self.ui.set_counter_values(current_text, self.logic.format_eights(total_eights)) self.ui.set_last_one(final_group) APP = MetronomeApp()