251 lines
8.2 KiB
Python
251 lines
8.2 KiB
Python
"""
|
|
Timeline: state machine + shared slider for the recording.
|
|
|
|
State is a plain ViewState dataclass — two orthogonal flags replace the old
|
|
4-state enum. A single "changed" GObject signal is emitted on any mutation;
|
|
consumers read timeline.state directly.
|
|
|
|
live=True — watching UDP relay; play/pause and seek are disabled
|
|
live=False — scrub mode; paused controls play/pause
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
|
|
import gi
|
|
gi.require_version("Gtk", "4.0")
|
|
from gi.repository import Gtk, GLib, GObject
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class ViewState:
|
|
live: bool = False # True = watching live UDP relay
|
|
paused: bool = False # scrub mode: is playback paused?
|
|
cursor: float = 0.0 # current position in recording (seconds)
|
|
duration: float = 0.0 # recording duration (seconds)
|
|
scene_markers: list[float] = field(default_factory=list)
|
|
|
|
|
|
class Timeline(GObject.Object):
|
|
"""State machine for recording playback.
|
|
|
|
Single signal: "changed" — emitted on any state mutation.
|
|
Consumers read timeline.state directly (no signal arguments).
|
|
"""
|
|
|
|
__gsignals__ = {
|
|
"changed": (GObject.SignalFlags.RUN_FIRST, None, ()),
|
|
}
|
|
|
|
def __init__(self):
|
|
super().__init__()
|
|
self.state = ViewState()
|
|
log.info("Timeline created")
|
|
|
|
def _emit(self):
|
|
self.emit("changed")
|
|
|
|
# -- Duration / cursor (called from background threads via GLib.idle_add) --
|
|
|
|
def set_duration(self, duration):
|
|
"""Update recording duration. In live mode, cursor follows end."""
|
|
if duration <= self.state.duration:
|
|
return
|
|
self.state.duration = duration
|
|
if self.state.live:
|
|
self.state.cursor = duration
|
|
self._emit()
|
|
|
|
def set_cursor(self, position):
|
|
"""Set cursor position (from slider drag or playback sync)."""
|
|
position = max(0.0, min(position, self.state.duration))
|
|
if abs(position - self.state.cursor) < 0.05:
|
|
return
|
|
self.state.cursor = position
|
|
self._emit()
|
|
|
|
def tick_live(self):
|
|
"""Advance cursor by 1s in live mode (smooth progression between probes)."""
|
|
if not self.state.live:
|
|
return
|
|
self.state.cursor += 1.0
|
|
# Only cap once duration is known — avoids clamping to 0 before first probe
|
|
if self.state.duration > 0:
|
|
self.state.cursor = min(self.state.cursor, self.state.duration)
|
|
self._emit()
|
|
|
|
def add_scene_marker(self, timestamp):
|
|
"""Add a scene change marker (does not emit — no UI update needed)."""
|
|
self.state.scene_markers.append(timestamp)
|
|
self.state.scene_markers.sort()
|
|
|
|
# -- User actions --
|
|
|
|
def go_live(self):
|
|
"""Go to live mode at the recording end."""
|
|
self.state.live = True
|
|
self.state.paused = False
|
|
if self.state.duration > 0:
|
|
self.state.cursor = self.state.duration
|
|
# else: keep current cursor (tick_live will continue from here)
|
|
self._emit()
|
|
|
|
def toggle_live(self, live_player_pos=None):
|
|
"""Toggle between live and scrub mode.
|
|
|
|
When leaving live, cursor stays at the live player's actual position
|
|
if provided, otherwise stays at current cursor.
|
|
"""
|
|
if self.state.live:
|
|
# Enter scrub mode at the live player's current position
|
|
self.state.live = False
|
|
self.state.paused = True
|
|
if live_player_pos is not None and live_player_pos > 0:
|
|
pos = max(0.0, live_player_pos)
|
|
# Only clamp to duration if duration is known
|
|
if self.state.duration > 0:
|
|
pos = min(pos, self.state.duration)
|
|
self.state.cursor = pos
|
|
else:
|
|
self.state.live = True
|
|
self.state.paused = False
|
|
if self.state.duration > 0:
|
|
self.state.cursor = self.state.duration
|
|
# else: keep current tick-based cursor, set_duration will snap later
|
|
self._emit()
|
|
|
|
def play(self):
|
|
if self.state.live:
|
|
return
|
|
self.state.paused = False
|
|
self._emit()
|
|
|
|
def pause(self):
|
|
if self.state.live:
|
|
return
|
|
self.state.paused = True
|
|
self._emit()
|
|
|
|
def seek(self, position):
|
|
"""Seek to position — enters scrub mode, pauses."""
|
|
self.state.live = False
|
|
self.state.paused = True
|
|
position = max(0.0, min(position, self.state.duration))
|
|
if abs(position - self.state.cursor) >= 0.05:
|
|
self.state.cursor = position
|
|
self._emit()
|
|
|
|
def reset(self):
|
|
"""Reset all state (called on disconnect)."""
|
|
self.state = ViewState()
|
|
self._emit()
|
|
|
|
|
|
class TimelineControls(Gtk.Box):
|
|
"""Slider + LIVE toggle. Scrub mode is always paused (seek-only, like a video editor).
|
|
|
|
LIVE button is a toggle — active style when live=True.
|
|
Slider is insensitive in live mode.
|
|
"""
|
|
|
|
def __init__(self, timeline, **kwargs):
|
|
super().__init__(orientation=Gtk.Orientation.HORIZONTAL, spacing=4, **kwargs)
|
|
self._timeline = timeline
|
|
self._dragging = False
|
|
|
|
self.set_margin_start(4)
|
|
self.set_margin_end(4)
|
|
self.set_margin_top(2)
|
|
self.set_margin_bottom(4)
|
|
|
|
# Current time label
|
|
self._time_label = Gtk.Label(label="00:00")
|
|
self._time_label.set_width_chars(6)
|
|
self.append(self._time_label)
|
|
|
|
# Slider — disabled in live mode, scrub-seeks on release
|
|
self._slider = Gtk.Scale(orientation=Gtk.Orientation.HORIZONTAL)
|
|
self._slider.set_hexpand(True)
|
|
self._slider.set_range(0, 1)
|
|
self._slider.set_draw_value(False)
|
|
self._slider.connect("value-changed", self._on_slider_value_changed)
|
|
|
|
press_ctrl = Gtk.GestureClick()
|
|
press_ctrl.connect("pressed", self._on_slider_pressed)
|
|
press_ctrl.connect("released", self._on_slider_released)
|
|
press_ctrl.set_propagation_phase(Gtk.PropagationPhase.CAPTURE)
|
|
self._slider.add_controller(press_ctrl)
|
|
self.append(self._slider)
|
|
|
|
# Duration label
|
|
self._duration_label = Gtk.Label(label="00:00 / 00:00")
|
|
self._duration_label.set_width_chars(14)
|
|
self.append(self._duration_label)
|
|
|
|
# LIVE toggle button
|
|
self._live_btn = Gtk.Button(label="LIVE")
|
|
self._live_btn.connect("clicked", self._on_live_clicked)
|
|
self.append(self._live_btn)
|
|
|
|
timeline.connect("changed", self._on_changed)
|
|
GLib.timeout_add(1000, self._tick_total)
|
|
|
|
def set_live_toggle_callback(self, cb):
|
|
"""Override the LIVE button handler."""
|
|
self._live_toggle_cb = cb
|
|
|
|
def _on_live_clicked(self, btn):
|
|
if hasattr(self, "_live_toggle_cb"):
|
|
self._live_toggle_cb()
|
|
else:
|
|
self._timeline.toggle_live()
|
|
|
|
def _on_slider_value_changed(self, slider):
|
|
if self._dragging:
|
|
self._time_label.set_text(self._fmt_time(slider.get_value()))
|
|
|
|
def _on_slider_pressed(self, gesture, n_press, x, y):
|
|
self._dragging = True
|
|
|
|
def _on_slider_released(self, gesture, n_press, x, y):
|
|
if self._dragging:
|
|
self._dragging = False
|
|
self._timeline.seek(self._slider.get_value())
|
|
|
|
def _on_changed(self, timeline):
|
|
s = timeline.state
|
|
|
|
self._slider.set_sensitive(not s.live)
|
|
|
|
if s.live:
|
|
self._live_btn.add_css_class("suggested-action")
|
|
else:
|
|
self._live_btn.remove_css_class("suggested-action")
|
|
|
|
if not self._dragging:
|
|
self._slider.set_range(0, max(s.duration, 0.1))
|
|
self._slider.set_value(s.cursor)
|
|
|
|
self._time_label.set_text(self._fmt_time(s.cursor))
|
|
self._update_duration_label()
|
|
|
|
def _tick_total(self):
|
|
self._update_duration_label()
|
|
return True
|
|
|
|
def _update_duration_label(self):
|
|
s = self._timeline.state
|
|
self._duration_label.set_text(
|
|
f"{self._fmt_time(s.cursor)} / {self._fmt_time(s.duration)}"
|
|
)
|
|
|
|
@staticmethod
|
|
def _fmt_time(seconds):
|
|
m, s = divmod(int(seconds), 60)
|
|
h, m = divmod(m, 60)
|
|
if h:
|
|
return f"{h}:{m:02d}:{s:02d}"
|
|
return f"{m:02d}:{s:02d}"
|