enne2
/
R36SHack
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
SDL2/GStreamer DLNA browser for R36S by Matteo Benedetto
You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
40 Commits
1 Branch
0 Tags
5.0 MiB
 
 
 
Tree: fe39312cfa
Branches Tags
${ item.name }
Create tag ${ searchTerm }
Create branch ${ searchTerm }
from 'fe39312cfa'
${ noResults }
R36SHack/src/r36s_dlna_browser/ui/sdl_app.py

502 lines
19 KiB
Raw Blame History

"""SDL2 window, event loop, rendering and input handling."""
from __future__ import annotations
import ctypes
import logging
import os
import queue
import time
from enum import Enum, auto
from typing import Any
import sdl2
try:
import sdl2.sdlttf as ttf
except ImportError:
ttf = None # type: ignore[assignment]
try:
import sdl2.sdlimage as sdl_img
except ImportError:
sdl_img = None # type: ignore[assignment]
from r36s_dlna_browser.dlna.browser_state import BrowserState
from r36s_dlna_browser.dlna.models import ItemType, MediaItem
from r36s_dlna_browser.player.backend import PlayerBackend
from r36s_dlna_browser.platform.controls import Action, map_key
from r36s_dlna_browser.ui import theme
from r36s_dlna_browser.ui import screens
log = logging.getLogger(__name__)
class Screen(Enum):
SERVERS = auto()
BROWSE = auto()
PLAYBACK = auto()
ERROR = auto()
class SDLApp:
"""Owns the SDL2 window and drives the main loop."""
def __init__(
self,
ui_queue: queue.Queue[Any],
cmd_queue: queue.Queue[Any],
browser_state: BrowserState,
player: PlayerBackend,
) -> None:
self._ui_q = ui_queue
self._cmd_q = cmd_queue
self._state = browser_state
self._player = player
self._screen = Screen.SERVERS
self._running = False
self._window = None
self._renderer = None
self._font = None
self._playback_font = None
self._icons: dict = {}
self._layout = theme.DEFAULT_LAYOUT
self._hud_cache: screens.HUDTextCache | None = None
self._playback_clear_once = False
self._needs_redraw = True
self._last_playback_draw_at = 0.0
self._last_playback_snapshot: tuple[Any, ...] | None = None
self._last_playback_interaction_at = 0.0
# ── lifecycle ────────────────────────────────────────────────
def run(self) -> None:
self._init_sdl()
self._running = True
try:
while self._running:
self._poll_events()
self._process_ui_queue()
self._refresh_playback_hud_visibility()
if self._should_draw():
self._draw()
sdl2.SDL_Delay(theme.FRAME_DELAY_MS)
finally:
self._cleanup()
def _init_sdl(self) -> None:
sdl2.SDL_Init(sdl2.SDL_INIT_VIDEO | sdl2.SDL_INIT_GAMECONTROLLER)
if ttf:
ttf.TTF_Init()
window_w, window_h = self._preferred_window_size()
self._layout = theme.get_layout(window_w, window_h)
self._window = sdl2.SDL_CreateWindow(
b"R36S DLNA Browser",
sdl2.SDL_WINDOWPOS_CENTERED,
sdl2.SDL_WINDOWPOS_CENTERED,
window_w,
window_h,
sdl2.SDL_WINDOW_SHOWN,
)
self._renderer = sdl2.SDL_CreateRenderer(
self._window, -1,
sdl2.SDL_RENDERER_ACCELERATED | sdl2.SDL_RENDERER_PRESENTVSYNC,
)
# Open first available gamecontroller
for i in range(sdl2.SDL_NumJoysticks()):
if sdl2.SDL_IsGameController(i):
sdl2.SDL_GameControllerOpen(i)
break
self._reload_fonts()
if not self._font:
log.error("Could not load any font – UI text will be missing.")
self._hud_cache = screens.HUDTextCache()
self._icons = self._load_icons()
self._player.attach_window(self._window)
self._sync_player_viewport()
def _load_icons(self) -> dict:
icons: dict = {}
if not sdl_img:
return icons
sdl_img.IMG_Init(sdl_img.IMG_INIT_PNG)
mapping = [
("folder", ItemType.CONTAINER),
("audio", ItemType.AUDIO),
("video", ItemType.VIDEO),
("image", ItemType.IMAGE),
("playing", "playing"),
("server", "server"),
("hud-play", "hud-play"),
("hud-pause", "hud-pause"),
("hud-stop", "hud-stop"),
("hud-seek", "hud-seek"),
("hud-volume", "hud-volume"),
("hud-display", "hud-display"),
]
for fname, key in mapping:
path = theme.ICONS_DIR / f"{fname}.png"
if path.exists():
surf = sdl_img.IMG_Load(str(path).encode())
if surf:
tex = sdl2.SDL_CreateTextureFromSurface(self._renderer, surf)
sdl2.SDL_FreeSurface(surf)
if tex:
icons[key] = tex
log.debug("Loaded icon: %s", fname)
log.info("Loaded %d icon(s).", len(icons))
return icons
def _load_font(self, size: int):
if not ttf:
log.warning("SDL2_ttf not available – text rendering disabled.")
return None
for path in theme.FONT_SEARCH_PATHS:
expanded = os.path.expanduser(path)
if os.path.isfile(expanded):
f = ttf.TTF_OpenFont(expanded.encode(), size)
if f:
log.info("Loaded font: %s", expanded)
return f
return None
def _reload_fonts(self) -> None:
if self._font and ttf:
ttf.TTF_CloseFont(self._font)
if self._playback_font and ttf:
ttf.TTF_CloseFont(self._playback_font)
self._font = self._load_font(self._layout.font_size)
self._playback_font = self._load_font(self._layout.playback_font_size)
if self._hud_cache is not None:
self._hud_cache.invalidate()
def _preferred_window_size(self) -> tuple[int, int]:
mode = sdl2.SDL_DisplayMode()
if sdl2.SDL_GetCurrentDisplayMode(0, ctypes.byref(mode)) == 0:
width = int(mode.w)
height = int(mode.h)
if 320 <= width <= 960 and 320 <= height <= 960:
return width, height
return theme.SCREEN_W, theme.SCREEN_H
def _update_layout(self, width: int, height: int) -> None:
prev_font_size = self._layout.font_size
prev_playback_font_size = self._layout.playback_font_size
self._layout = theme.get_layout(width, height)
if (
self._layout.font_size != prev_font_size
or self._layout.playback_font_size != prev_playback_font_size
):
self._reload_fonts()
def _cleanup(self) -> None:
if self._hud_cache is not None:
self._hud_cache.invalidate()
self._hud_cache = None
for tex in self._icons.values():
sdl2.SDL_DestroyTexture(tex)
self._icons = {}
if sdl_img:
sdl_img.IMG_Quit()
if self._font and ttf:
ttf.TTF_CloseFont(self._font)
if self._playback_font and ttf:
ttf.TTF_CloseFont(self._playback_font)
if self._renderer:
sdl2.SDL_DestroyRenderer(self._renderer)
if self._window:
sdl2.SDL_DestroyWindow(self._window)
if ttf:
ttf.TTF_Quit()
sdl2.SDL_Quit()
# ── events ───────────────────────────────────────────────────
def _poll_events(self) -> None:
event = sdl2.SDL_Event()
while sdl2.SDL_PollEvent(ctypes.byref(event)):
if event.type == sdl2.SDL_QUIT:
self._running = False
return
if event.type == sdl2.SDL_WINDOWEVENT and event.window.event in (
sdl2.SDL_WINDOWEVENT_MOVED,
sdl2.SDL_WINDOWEVENT_SIZE_CHANGED,
sdl2.SDL_WINDOWEVENT_RESIZED,
):
if self._window and event.window.windowID == sdl2.SDL_GetWindowID(self._window):
self._sync_player_viewport()
self._mark_dirty()
action = map_key(event)
if action is not None:
self._handle_action(action)
def _process_ui_queue(self) -> None:
while True:
try:
msg = self._ui_q.get_nowait()
except queue.Empty:
break
self._handle_ui_message(msg)
def _handle_ui_message(self, msg: tuple) -> None:
kind = msg[0]
if kind == "servers_updated":
self._state.loading = False
self._screen = Screen.SERVERS
elif kind == "items_updated":
self._state.loading = False
self._screen = Screen.BROWSE
elif kind == "playback_started":
self._state.playback_title = msg[1] if len(msg) > 1 else ""
self._state.playback_paused = False
self._show_playback_hud(force=True)
self._playback_clear_once = True
self._screen = Screen.PLAYBACK
elif kind == "playback_paused":
self._state.playback_paused = True
self._show_playback_hud(force=True)
elif kind == "playback_resumed":
self._state.playback_paused = False
self._show_playback_hud(force=True)
elif kind == "playback_stopped":
self._state.playback_paused = False
self._screen = Screen.BROWSE if self._state.in_browse_mode else Screen.SERVERS
elif kind == "error":
self._state.error = msg[1] if len(msg) > 1 else "Unknown error"
self._state.loading = False
self._screen = Screen.ERROR
self._mark_dirty()
# ── input dispatch ───────────────────────────────────────────
def _handle_action(self, action: Action) -> None:
if action == Action.QUIT:
self._running = False
return
if self._screen == Screen.ERROR:
if action in (Action.CONFIRM, Action.BACK):
self._state.error = ""
self._screen = Screen.BROWSE if self._state.in_browse_mode else Screen.SERVERS
self._mark_dirty()
return
if self._screen == Screen.PLAYBACK:
if action == Action.BACK:
self._cmd_q.put(("stop",))
elif action == Action.CONFIRM:
self._cmd_q.put(("toggle_pause",))
elif action == Action.UP:
self._cmd_q.put(("volume", 5))
elif action == Action.DOWN:
self._cmd_q.put(("volume", -5))
elif action == Action.PAGE_UP:
self._cmd_q.put(("seek", -10))
elif action == Action.PAGE_DOWN:
self._cmd_q.put(("seek", 10))
elif action == Action.HUD_MODE:
self._cycle_playback_hud_mode()
self._mark_dirty()
self._show_playback_hud(force=action != Action.HUD_MODE)
return
if self._screen == Screen.SERVERS:
self._handle_server_action(action)
elif self._screen == Screen.BROWSE:
self._handle_browse_action(action)
def _handle_server_action(self, action: Action) -> None:
n = len(self._state.servers)
if action == Action.UP and n:
self._state.server_cursor = max(0, self._state.server_cursor - 1)
self._adjust_server_scroll()
self._mark_dirty()
elif action == Action.DOWN and n:
self._state.server_cursor = min(n - 1, self._state.server_cursor + 1)
self._adjust_server_scroll()
self._mark_dirty()
elif action == Action.CONFIRM:
srv = self._state.selected_server
if srv:
obj_id = self._state.enter_server(srv)
self._state.loading = True
self._cmd_q.put(("browse", srv.location, obj_id))
else:
# No server selected – refresh
self._state.loading = True
self._cmd_q.put(("discover",))
self._mark_dirty()
elif action == Action.BACK:
# Refresh server list
self._state.loading = True
self._cmd_q.put(("discover",))
self._mark_dirty()
def _handle_browse_action(self, action: Action) -> None:
items = self._state.current_items
n = len(items)
if action == Action.UP and n:
self._state.cursor -= 1
self._adjust_browse_scroll()
self._mark_dirty()
elif action == Action.DOWN and n:
self._state.cursor += 1
self._adjust_browse_scroll()
self._mark_dirty()
elif action == Action.PAGE_UP and n:
self._state.cursor = max(0, self._state.cursor - self._layout.page_size)
self._adjust_browse_scroll()
self._mark_dirty()
elif action == Action.PAGE_DOWN and n:
self._state.cursor = min(n - 1, self._state.cursor + self._layout.page_size)
self._adjust_browse_scroll()
self._mark_dirty()
elif action == Action.CONFIRM:
item = self._state.selected_item()
if item:
if item.is_container:
self._state.loading = True
lv = self._state.current_level
loc = lv.server_location if lv else ""
self._cmd_q.put(("browse", loc, item.object_id))
elif item.resource_url:
self._cmd_q.put(("play", item.resource_url, item.title))
self._mark_dirty()
elif action == Action.BACK:
if not self._state.go_back():
self._screen = Screen.SERVERS
else:
self._screen = Screen.BROWSE
self._mark_dirty()
def _adjust_server_scroll(self) -> None:
vis = self._layout.visible_items
cur = self._state.server_cursor
off = self._state.server_scroll_offset
if cur < off:
self._state.server_scroll_offset = cur
elif cur >= off + vis:
self._state.server_scroll_offset = cur - vis + 1
def _adjust_browse_scroll(self) -> None:
vis = self._layout.visible_items
cur = self._state.cursor
off = self._state.scroll_offset
if cur < off:
self._state.scroll_offset = cur
elif cur >= off + vis:
self._state.scroll_offset = cur - vis + 1
# ── drawing ──────────────────────────────────────────────────
def _draw(self) -> None:
r = self._renderer
if self._screen == Screen.PLAYBACK:
sdl2.SDL_SetRenderDrawColor(r, *theme.PLAYBACK_BG_COLOR)
else:
sdl2.SDL_SetRenderDrawColor(r, *theme.BG_COLOR)
sdl2.SDL_RenderClear(r)
self._playback_clear_once = False
f = self._font
if self._screen == Screen.SERVERS:
screens.draw_server_list(r, f, self._state, self._layout, self._icons)
elif self._screen == Screen.BROWSE:
screens.draw_browse_list(r, f, self._state, self._layout, self._icons)
elif self._screen == Screen.PLAYBACK:
self._player.render(r)
screens.draw_playback(r, self._playback_font or f, self._state, self._layout, self._icons, cache=self._hud_cache)
elif self._screen == Screen.ERROR:
screens.draw_error(r, f, self._state, self._layout)
sdl2.SDL_RenderPresent(r)
self._needs_redraw = False
if self._screen == Screen.PLAYBACK:
self._last_playback_draw_at = time.monotonic()
self._last_playback_snapshot = self._playback_snapshot()
def _mark_dirty(self) -> None:
self._needs_redraw = True
def _playback_snapshot(self) -> tuple[Any, ...]:
return (
self._state.playback_title,
self._state.playback_paused,
round(self._state.playback_position, 1),
round(self._state.playback_duration, 1),
self._state.playback_volume,
self._state.playback_buffer_percent,
self._state.playback_resolution,
self._state.playback_backend,
self._state.playback_hud_visible,
self._state.playback_hud_mode,
)
def _refresh_playback_hud_visibility(self) -> None:
if self._screen != Screen.PLAYBACK:
return
if self._state.playback_hud_mode == theme.PLAYBACK_HUD_HIDDEN:
return
if self._state.playback_hud_mode == theme.PLAYBACK_HUD_PINNED or self._state.playback_paused:
if not self._state.playback_hud_visible:
self._state.playback_hud_visible = True
self._mark_dirty()
return
visible = (time.monotonic() - self._last_playback_interaction_at) < (theme.PLAYBACK_HUD_AUTOHIDE_MS / 1000.0)
if visible != self._state.playback_hud_visible:
self._state.playback_hud_visible = visible
self._mark_dirty()
def _show_playback_hud(self, force: bool = False) -> None:
self._last_playback_interaction_at = time.monotonic()
if force or self._state.playback_hud_mode != theme.PLAYBACK_HUD_HIDDEN:
if not self._state.playback_hud_visible:
self._state.playback_hud_visible = True
self._mark_dirty()
def _cycle_playback_hud_mode(self) -> None:
next_mode = {
theme.PLAYBACK_HUD_AUTO: theme.PLAYBACK_HUD_PINNED,
theme.PLAYBACK_HUD_PINNED: theme.PLAYBACK_HUD_HIDDEN,
theme.PLAYBACK_HUD_HIDDEN: theme.PLAYBACK_HUD_AUTO,
}
self._state.playback_hud_mode = next_mode[self._state.playback_hud_mode]
self._state.playback_hud_visible = self._state.playback_hud_mode != theme.PLAYBACK_HUD_HIDDEN
self._last_playback_interaction_at = time.monotonic()
def _should_draw(self) -> bool:
if self._needs_redraw:
return True
if self._screen != Screen.PLAYBACK:
return False
if self._player.has_new_frame():
return True
now = time.monotonic()
if now - self._last_playback_draw_at < (theme.PLAYBACK_HUD_REFRESH_MS / 1000.0):
return False
return self._playback_snapshot() != self._last_playback_snapshot
def _sync_player_viewport(self) -> None:
width = ctypes.c_int()
height = ctypes.c_int()
if self._window:
sdl2.SDL_GetWindowSize(self._window, ctypes.byref(width), ctypes.byref(height))
if width.value <= 0 or height.value <= 0:
width.value = theme.SCREEN_W
height.value = theme.SCREEN_H
self._update_layout(width.value, height.value)
self._player.set_viewport(
width.value,
height.value,
self._layout.playback_hud_top,
height.value - self._layout.playback_video_bottom,
self._layout.playback_video_left,
max(0, width.value - self._layout.playback_video_right),
)