shawn/raspberry-pi-mixer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
cdf04bd8eb0902f6f23e186cc20238f123d1c296
raspberry-pi-mixer/src/ui/screens/sessions.py
T
shawn cdf04bd8eb
Lint & Validate / lint (push) Has been cancelled
Details
P6-R1: Audio/video streaming pipeline 
- src/streaming/ module: camera detection (USB webcam + Pi Camera Module via
  v4l2/libcamera), GStreamer pipeline builder with V4L2/OMX/x264 h264 encoders,
  platform presets (YouTube, Twitch, Facebook, custom RTMP), streamer lifecycle
  manager with auto-reconnect, scene management, stream statistics
- Keyboard shortcut controller (evdev + pynput fallback) for stream control:
  Ctrl+Shift+S (start/stop), Ctrl+Shift+1-9 (scene switch), Ctrl+Shift+R (reconnect)
- REST API: /api/v1/stream/* endpoints for start/stop/status, scenes, cameras,
  platforms, and hotkey listing — integrated into NetworkServer
- StreamStateProvider bridging layer connects REST routes to Streamer instance
- GStreamer pipeline supports ALSA/JACK/PulseAudio audio capture and
  V4L2/libcamera video capture with H.264 hardware encoding on RPi4B
- Low-latency streaming settings for live performance (zerolatency, no B-frames)
- 89 streaming tests (709 total project tests pass)
2026-05-19 22:13:51 -04:00

636 lines
20 KiB
Python
Raw Blame History

"""Sessions screen — save/load mixer sessions, setlists, and snapshots.
Touch-optimized layout for Raspberry Pi 5"/7" displays.
Provides:
- Session list with load/delete buttons
- Save current session button
- Snapshot capture/recall
- Setlist browser
"""
from __future__ import annotations
from kivy.properties import ObjectProperty, StringProperty, ListProperty
from kivy.uix.boxlayout import BoxLayout
from kivy.uix.button import Button
from kivy.uix.label import Label
from kivy.uix.screenmanager import Screen
from kivy.uix.scrollview import ScrollView
from kivy.uix.textinput import TextInput
from kivy.uix.togglebutton import ToggleButton
from kivy.metrics import dp
from kivy.clock import Clock
from ..theme import Colors, Fonts, Sizes
class SessionCard(BoxLayout):
"""A single session/snapshot entry card."""
session_name = StringProperty("")
session_info = StringProperty("")
is_snapshot = False
def __init__(self, name="", info="", is_snapshot=False, on_load=None,
on_delete=None, on_recall=None, **kwargs):
super().__init__(**kwargs)
self.orientation = "horizontal"
self.size_hint = (1, None)
self.height = dp(52)
self.padding = [dp(8), dp(4)]
self.spacing = dp(6)
self._name = name
self._is_snapshot = is_snapshot
icon = "📸" if is_snapshot else "💾"
# Name + info
info_layout = BoxLayout(orientation="vertical", size_hint=(1, 1))
name_label = Label(
text=f"{icon} {name}",
color=Colors.TEXT,
font_size=Fonts.SMALL,
halign="left",
valign="middle",
size_hint=(1, 1),
text_size=(None, None),
)
name_label.bind(size=name_label.setter("text_size"))
info_label = Label(
text=info,
color=Colors.TEXT_MUTED,
font_size=Fonts.XSMALL,
halign="left",
size_hint=(1, 0.5),
text_size=(None, None),
)
info_label.bind(size=info_label.setter("text_size"))
info_layout.add_widget(name_label)
info_layout.add_widget(info_label)
# Buttons
btn_layout = BoxLayout(
orientation="horizontal",
size_hint=(None, 1),
width=dp(140),
spacing=dp(4),
)
if is_snapshot:
load_btn = Button(
text="Recall",
font_size=Fonts.XSMALL,
size_hint=(1, 1),
background_color=Colors.BG_BUTTON,
color=Colors.TEXT,
)
if on_recall:
load_btn.bind(on_press=lambda x: on_recall(name))
else:
load_btn = Button(
text="Load",
font_size=Fonts.XSMALL,
size_hint=(1, 1),
background_color=Colors.BG_BUTTON,
color=Colors.TEXT,
)
if on_load:
load_btn.bind(on_press=lambda x: on_load(name))
del_btn = Button(
text="✕",
font_size=Fonts.XSMALL,
size_hint=(None, 1),
width=dp(40),
background_color=Colors.MUTE_ON,
color=Colors.TEXT,
)
if on_delete:
del_btn.bind(on_press=lambda x: on_delete(name))
btn_layout.add_widget(load_btn)
btn_layout.add_widget(del_btn)
self.add_widget(info_layout)
self.add_widget(btn_layout)
class SessionsScreen(Screen):
"""Session management screen for the touch UI."""
client = ObjectProperty(None)
def __init__(self, **kwargs):
super().__init__(**kwargs)
self.name = "sessions"
self._sessions_data = []
self._snapshots_data = []
self._current_subtab = "sessions"
root = BoxLayout(orientation="vertical")
# Header
header = BoxLayout(
orientation="horizontal",
size_hint=(1, None),
height=Sizes.HEADER_HEIGHT,
padding=[Sizes.PAD_MD, Sizes.PAD_XS],
)
header.bg_color = Colors.BG_HEADER
title = Label(
text="SESSIONS",
color=Colors.ACCENT,
font_size=Fonts.HEADER,
bold=True,
size_hint=(None, 1),
width=dp(150),
halign="left",
)
title.bind(size=title.setter("text_size"))
# Sub-tab bar
sub_tabs = BoxLayout(
orientation="horizontal",
size_hint=(1, None),
height=dp(40),
spacing=0,
)
sub_tabs.bg_color = Colors.BG_DARK
self._tab_sessions = ToggleButton(
text="Sessions",
font_size=Fonts.SMALL,
group="subtab",
state="down",
background_color=Colors.BG_HEADER,
color=Colors.ACCENT,
)
self._tab_setlists = ToggleButton(
text="Setlists",
font_size=Fonts.SMALL,
group="subtab",
background_color=Colors.BG_HEADER,
color=Colors.TEXT_MUTED,
)
self._tab_snapshots = ToggleButton(
text="Snapshots",
font_size=Fonts.SMALL,
group="subtab",
background_color=Colors.BG_HEADER,
color=Colors.TEXT_MUTED,
)
self._tab_sessions.bind(on_press=lambda x: self._switch_subtab("sessions"))
self._tab_setlists.bind(on_press=lambda x: self._switch_subtab("setlists"))
self._tab_snapshots.bind(on_press=lambda x: self._switch_subtab("snapshots"))
sub_tabs.add_widget(self._tab_sessions)
sub_tabs.add_widget(self._tab_setlists)
sub_tabs.add_widget(self._tab_snapshots)
# Save bar
save_bar = BoxLayout(
orientation="horizontal",
size_hint=(1, None),
height=dp(44),
padding=[dp(8), dp(4)],
spacing=dp(6),
)
self._name_input = TextInput(
text="",
hint_text="Session name...",
font_size=Fonts.SMALL,
size_hint=(1, 1),
background_color=Colors.BG_INPUT,
foreground_color=Colors.TEXT,
multiline=False,
)
self._notes_input = TextInput(
text="",
hint_text="Notes...",
font_size=Fonts.SMALL,
size_hint=(0.6, 1),
background_color=Colors.BG_INPUT,
foreground_color=Colors.TEXT,
multiline=False,
)
save_btn = Button(
text="Save",
font_size=Fonts.SMALL,
size_hint=(None, 1),
width=dp(80),
background_color=Colors.ACCENT,
color=Colors.BG_DARK,
)
save_btn.bind(on_press=self._on_save)
save_bar.add_widget(self._name_input)
save_bar.add_widget(self._notes_input)
save_bar.add_widget(save_btn)
# Snapshot bar (hidden by default)
self._snapshot_bar = BoxLayout(
orientation="horizontal",
size_hint=(1, None),
height=dp(44),
padding=[dp(8), dp(4)],
spacing=dp(6),
)
self._snap_name_input = TextInput(
text="",
hint_text="Snapshot name...",
font_size=Fonts.SMALL,
size_hint=(1, 1),
background_color=Colors.BG_INPUT,
foreground_color=Colors.TEXT,
multiline=False,
)
snap_btn = Button(
text="Capture",
font_size=Fonts.SMALL,
size_hint=(None, 1),
width=dp(80),
background_color=Colors.ACCENT,
color=Colors.BG_DARK,
)
snap_btn.bind(on_press=self._on_capture_snapshot)
next_btn = Button(
text="▶",
font_size=Fonts.SMALL,
size_hint=(None, 1),
width=dp(44),
background_color=Colors.BG_BUTTON,
color=Colors.TEXT,
)
next_btn.bind(on_press=self._on_snapshot_next)
prev_btn = Button(
text="◀",
font_size=Fonts.SMALL,
size_hint=(None, 1),
width=dp(44),
background_color=Colors.BG_BUTTON,
color=Colors.TEXT,
)
prev_btn.bind(on_press=self._on_snapshot_prev)
self._snapshot_bar.add_widget(self._snap_name_input)
self._snapshot_bar.add_widget(snap_btn)
self._snapshot_bar.add_widget(prev_btn)
self._snapshot_bar.add_widget(next_btn)
# Scrollable list area
self._list_scroll = ScrollView(size_hint=(1, 1), do_scroll_x=False)
self._list_layout = BoxLayout(
orientation="vertical",
size_hint=(1, None),
spacing=dp(2),
padding=[dp(4), dp(4)],
)
self._list_layout.bind(minimum_height=self._list_layout.setter("height"))
self._list_scroll.add_widget(self._list_layout)
# Setlist view (hidden by default)
self._setlist_view = BoxLayout(
orientation="vertical",
size_hint=(1, 1),
spacing=dp(4),
)
self._setlist_view.opacity = 0
self._setlist_view.disabled = True
self._setlist_list = BoxLayout(
orientation="vertical",
size_hint=(1, 1),
spacing=dp(2),
padding=[dp(4), dp(4)],
)
setlist_bar = BoxLayout(
orientation="horizontal",
size_hint=(1, None),
height=dp(44),
padding=[dp(8), dp(4)],
spacing=dp(6),
)
self._setlist_name_input = TextInput(
text="",
hint_text="Setlist name...",
font_size=Fonts.SMALL,
size_hint=(1, 1),
background_color=Colors.BG_INPUT,
foreground_color=Colors.TEXT,
multiline=False,
)
new_setlist_btn = Button(
text="Create",
font_size=Fonts.SMALL,
size_hint=(None, 1),
width=dp(80),
background_color=Colors.ACCENT,
color=Colors.BG_DARK,
)
new_setlist_btn.bind(on_press=self._on_create_setlist)
setlist_bar.add_widget(self._setlist_name_input)
setlist_bar.add_widget(new_setlist_btn)
self._setlist_view.add_widget(setlist_bar)
self._setlist_view.add_widget(self._setlist_list)
# Content area
content = BoxLayout(orientation="vertical", size_hint=(1, 1))
content.add_widget(self._list_scroll)
content.add_widget(self._setlist_view)
# Combine
header.add_widget(title)
root.add_widget(header)
root.add_widget(sub_tabs)
root.add_widget(save_bar)
root.add_widget(self._snapshot_bar)
root.add_widget(content)
self.add_widget(root)
# Refresh periodically
Clock.schedule_interval(self._refresh, 10.0)
def on_enter(self, *args):
"""Called when the screen is displayed."""
self._refresh()
def _switch_subtab(self, tab: str):
"""Switch between sessions/setlists/snapshots."""
self._current_subtab = tab
# Update toggle states
self._tab_sessions.state = "down" if tab == "sessions" else "normal"
self._tab_setlists.state = "down" if tab == "setlists" else "normal"
self._tab_snapshots.state = "down" if tab == "snapshots" else "normal"
# Toggle visibility
if tab == "setlists":
self._list_scroll.opacity = 0
self._list_scroll.disabled = True
self._setlist_view.opacity = 1
self._setlist_view.disabled = False
self._snapshot_bar.opacity = 0
self._snapshot_bar.disabled = True
elif tab == "snapshots":
self._list_scroll.opacity = 1
self._list_scroll.disabled = False
self._setlist_view.opacity = 0
self._setlist_view.disabled = True
self._snapshot_bar.opacity = 1
self._snapshot_bar.disabled = False
else:
self._list_scroll.opacity = 1
self._list_scroll.disabled = False
self._setlist_view.opacity = 0
self._setlist_view.disabled = True
self._snapshot_bar.opacity = 0
self._snapshot_bar.disabled = True
self._refresh()
def _refresh(self, *args):
"""Refresh the session list from the REST API."""
if not self.client:
return
if self._current_subtab in ("sessions", "snapshots"):
self._fetch_sessions()
if self._current_subtab == "snapshots":
self._fetch_snapshots()
elif self._current_subtab == "setlists":
self._fetch_setlists()
def _fetch_sessions(self):
"""Fetch sessions list from REST API."""
try:
self.client.rest_get(
"/api/v1/sessions",
on_success=self._on_sessions_loaded,
on_error=lambda e: None,
)
except Exception:
pass
def _fetch_snapshots(self):
"""Fetch snapshots list from REST API."""
try:
self.client.rest_get(
"/api/v1/snapshots",
on_success=self._on_snapshots_loaded,
on_error=lambda e: None,
)
except Exception:
pass
def _fetch_setlists(self):
"""Fetch setlists list from REST API."""
try:
self.client.rest_get(
"/api/v1/setlists",
on_success=self._on_setlists_loaded,
on_error=lambda e: None,
)
except Exception:
pass
def _on_sessions_loaded(self, data: dict):
"""Render session cards."""
sessions = data.get("sessions", [])
self._sessions_data = sessions
self._render_session_list()
def _on_snapshots_loaded(self, data: dict):
"""Render snapshot cards."""
snapshots = data.get("snapshots", [])
self._snapshots_data = snapshots
self._render_session_list()
def _render_session_list(self):
"""Build the session/snapshot card list."""
self._list_layout.clear_widgets()
items = []
for s in (self._sessions_data or []):
items.append({
"type": "session",
"name": s.get("name", ""),
"info": f"{s.get('size_bytes', 0)} bytes",
"is_snapshot": False,
})
for s in (self._snapshots_data or []):
cat = s.get("category", "")
cur = " (active)" if s.get("is_current") else ""
items.append({
"type": "snapshot",
"name": s.get("name", ""),
"info": f"{cat}{cur}" if (cat or cur) else "",
"is_snapshot": True,
})
if not items:
self._list_layout.add_widget(Label(
text="No sessions saved yet.\nUse the Save button above.",
color=Colors.TEXT_MUTED,
font_size=Fonts.SMALL,
size_hint=(1, None),
height=dp(60),
halign="center",
valign="middle",
))
return
for item in items:
card = SessionCard(
name=item["name"],
info=item["info"],
is_snapshot=item["is_snapshot"],
on_load=self._on_load if not item["is_snapshot"] else None,
on_delete=self._on_delete,
on_recall=self._on_recall if item["is_snapshot"] else None,
)
self._list_layout.add_widget(card)
def _on_setlists_loaded(self, data: dict):
"""Render setlist cards."""
self._setlist_list.clear_widgets()
setlists = data.get("setlists", [])
if not setlists:
self._setlist_list.add_widget(Label(
text="No setlists. Create one above.",
color=Colors.TEXT_MUTED,
font_size=Fonts.SMALL,
size_hint=(1, None),
height=dp(40),
halign="center",
))
return
def _on_load(self, name: str):
"""Load a session."""
if not self.client:
return
try:
self.client.rest_post(
f"/api/v1/sessions/{name}/load",
body={},
on_success=lambda d: None,
on_error=lambda e: None,
)
except Exception:
pass
def _on_delete(self, name: str):
"""Delete a session or snapshot."""
if not self.client:
return
prefix = "snapshots" if self._current_subtab == "snapshots" else "sessions"
try:
self.client.rest_delete(
f"/api/v1/{prefix}/{name}",
on_success=lambda d: self._refresh(),
on_error=lambda e: None,
)
except Exception:
pass
def _on_recall(self, name: str):
"""Recall a snapshot."""
if not self.client:
return
try:
self.client.rest_post(
f"/api/v1/snapshots/{name}/recall",
body={},
on_success=lambda d: None,
on_error=lambda e: None,
)
except Exception:
pass
def _on_save(self, instance):
"""Save current state as a session."""
if not self.client:
return
name = self._name_input.text.strip()
if not name:
return
notes = self._notes_input.text.strip()
try:
self.client.rest_post(
"/api/v1/sessions/save",
body={"name": name, "overwrite": True},
on_success=lambda d: (self._name_input.setter("text")(""), self._refresh()),
on_error=lambda e: None,
)
except Exception:
pass
def _on_capture_snapshot(self, instance):
"""Capture current state as a snapshot."""
if not self.client:
return
name = self._snap_name_input.text.strip()
if not name:
return
try:
self.client.rest_post(
"/api/v1/snapshots/capture",
body={"name": name, "overwrite": True},
on_success=lambda d: (self._snap_name_input.setter("text")(""), self._refresh()),
on_error=lambda e: None,
)
except Exception:
pass
def _on_snapshot_next(self, instance):
"""Go to next snapshot."""
if not self.client:
return
try:
self.client.rest_post(
"/api/v1/snapshots/next",
body={},
on_success=lambda d: None,
on_error=lambda e: None,
)
except Exception:
pass
def _on_snapshot_prev(self, instance):
"""Go to previous snapshot."""
if not self.client:
return
try:
self.client.rest_post(
"/api/v1/snapshots/previous",
body={},
on_success=lambda d: None,
on_error=lambda e: None,
)
except Exception:
pass
def _on_create_setlist(self, instance):
"""Create a new setlist."""
if not self.client:
return
name = self._setlist_name_input.text.strip()
if not name:
return
try:
self.client.rest_post(
"/api/v1/setlists/save",
body={"name": name, "entries": [], "description": ""},
on_success=lambda d: self._refresh(),
on_error=lambda e: None,
)
except Exception:
pass
Reference in New Issue View Git Blame Copy Permalink