236 lines
7.4 KiB
Python
236 lines
7.4 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):
|
|
"""Scrub bar + time labels + LIVE toggle.
|
|
|
|
The scrub bar shows segment blocks; labels and LIVE button sit below.
|
|
"""
|
|
|
|
def __init__(self, timeline, **kwargs):
|
|
super().__init__(orientation=Gtk.Orientation.VERTICAL, spacing=2, **kwargs)
|
|
self._timeline = timeline
|
|
|
|
self.set_margin_start(4)
|
|
self.set_margin_end(4)
|
|
self.set_margin_top(2)
|
|
self.set_margin_bottom(4)
|
|
|
|
# Scrub bar (segment blocks)
|
|
from cht.ui.scrub_bar import ScrubBar
|
|
self._scrub_bar = ScrubBar()
|
|
self.append(self._scrub_bar)
|
|
|
|
# Bottom row: time label + duration + LIVE button
|
|
bottom = Gtk.Box(orientation=Gtk.Orientation.HORIZONTAL, spacing=4)
|
|
|
|
self._time_label = Gtk.Label(label="00:00")
|
|
self._time_label.set_width_chars(6)
|
|
bottom.append(self._time_label)
|
|
|
|
spacer = Gtk.Box()
|
|
spacer.set_hexpand(True)
|
|
bottom.append(spacer)
|
|
|
|
self._duration_label = Gtk.Label(label="00:00 / 00:00")
|
|
self._duration_label.set_width_chars(14)
|
|
bottom.append(self._duration_label)
|
|
|
|
self._live_btn = Gtk.Button(label="LIVE")
|
|
self._live_btn.connect("clicked", self._on_live_clicked)
|
|
bottom.append(self._live_btn)
|
|
|
|
self.append(bottom)
|
|
|
|
timeline.connect("changed", self._on_changed)
|
|
GLib.timeout_add(1000, self._tick_total)
|
|
|
|
@property
|
|
def scrub_bar(self):
|
|
"""Access the ScrubBar widget for signal connections."""
|
|
return self._scrub_bar
|
|
|
|
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_changed(self, timeline):
|
|
s = timeline.state
|
|
|
|
if s.live:
|
|
self._live_btn.add_css_class("suggested-action")
|
|
else:
|
|
self._live_btn.remove_css_class("suggested-action")
|
|
|
|
self._scrub_bar.set_duration(s.duration)
|
|
self._scrub_bar.set_cursor(s.cursor)
|
|
self._scrub_bar.set_scene_markers(s.scene_markers)
|
|
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}"
|