byos_fastapi/trmnl_server/services/state.py

"""Shared device, rotation, and state management utilities."""

from __future__ import annotations

from hashlib import sha1
from os.path import abspath
from time import time
from io import BytesIO
from threading import RLock
from typing import Any, Dict, List, Optional, Tuple
from urllib.parse import quote

from fastapi import Request

from .. import config, models, utils
from ..plugins.base import PluginOutput

logger = config.logger

DEFAULT_DEVICE_ID = 'default'
STATE_LOCK = RLock()
start_time = time()

_IMAGE_TOKEN_TTL_SECONDS = 600.0

_server_base_url = ''


class RotationUnavailableError(RuntimeError):
    """Raised when no rotation entries are available for a device."""
    pass


def set_server_base_url(base_url: str) -> None:
    """Persist the server base URL so freshly created device states have a default."""
    global _server_base_url
    _server_base_url = base_url.rstrip('/') if base_url else ''


def get_server_base_url() -> str:
    return _server_base_url or f"{config.SERVER_SCHEME}://{utils.get_ip_address()}:{config.SERVER_PORT}"


def update_device_preview(
    device_id: str,
    device_state: Optional[Dict[str, Any]],
    entry_index: Optional[int]
) -> Tuple[Optional[str], Optional[str]]:
    if entry_index is None or entry_index < 0:
        return None, None
    target_state = device_state or get_device_state(device_id)
    with STATE_LOCK:
        previous_index = target_state.get('current_preview_entry_index')
        target_state['current_preview_entry_index'] = entry_index
        existing_token = target_state.get('current_preview_token')
        if previous_index == entry_index and existing_token:
            token = str(existing_token)
        else:
            sequence = int(target_state.get('token_sequence', 0))
            token = sha1(f"{device_id}-{entry_index}-{sequence}".encode('utf-8')).hexdigest()[:16]
            target_state['current_preview_token'] = token
        target_state['current_preview_url'] = f"/preview/{quote(device_id)}?token={token}"
        return target_state['current_preview_url'], token


def _client_metrics_store() -> Dict[str, Dict[str, Any]]:
    return global_state.setdefault('client_metrics', {})


def get_client_metrics(device_id: str) -> Dict[str, Any]:
    with STATE_LOCK:
        store = _client_metrics_store()
        metrics = store.get(device_id)
        if metrics is None:
            metrics = {
                'refresh_rate': config.REFRESH_TIME,
                'battery_voltage': config.BATTERY_MAX_VOLTAGE,
                'rssi': -100,
                'last_contact': 0.0
            }
            store[device_id] = metrics
        return metrics


def update_client_metrics(
    device_id: str,
    *,
    refresh_rate: Optional[int] = None,
    battery_voltage: Optional[float] = None,
    rssi: Optional[int] = None
) -> None:
    with STATE_LOCK:
        metrics = get_client_metrics(device_id)
        if refresh_rate is not None:
            metrics['refresh_rate'] = refresh_rate
        if battery_voltage is not None:
            metrics['battery_voltage'] = battery_voltage
        if rssi is not None:
            metrics['rssi'] = rssi
        metrics['last_contact'] = time()


def get_all_client_metrics() -> Dict[str, Dict[str, Any]]:
    with STATE_LOCK:
        return {device_id: dict(metrics) for device_id, metrics in _client_metrics_store().items()}


def _device_profile_cache() -> Dict[str, Dict[str, Any]]:
    return global_state.setdefault('device_profiles', {})


def ensure_device_profile(device_id: str) -> Dict[str, Any]:
    with STATE_LOCK:
        cache = _device_profile_cache()
        profile = cache.get(device_id)
    if profile is None:
        profile = models.ensure_device_profile(device_id)
        with STATE_LOCK:
            cache = _device_profile_cache()
            cache[device_id] = profile
    return profile


def refresh_device_profile(device_id: str) -> Dict[str, Any]:
    profile = models.ensure_device_profile(device_id)
    with STATE_LOCK:
        cache = _device_profile_cache()
        cache[device_id] = profile
    return profile


def update_device_profile(
    device_id: str,
    *,
    friendly_name: Optional[str] = None,
    refresh_interval: Optional[int] = None,
    time_zone: Optional[str] = None
) -> Dict[str, Any]:
    profile = models.update_device_profile(
        device_id,
        friendly_name=friendly_name,
        refresh_interval=refresh_interval,
        time_zone=time_zone
    )
    with STATE_LOCK:
        cache = _device_profile_cache()
        cache[device_id] = profile
    return profile


def get_refresh_interval(device_id: str) -> int:
    profile = ensure_device_profile(device_id)
    refresh_interval = profile.get('refresh_interval')
    if isinstance(refresh_interval, int) and refresh_interval > 0:
        return refresh_interval
    metrics = get_client_metrics(device_id)
    return int(metrics.get('refresh_rate', config.REFRESH_TIME))


def _build_device_state() -> Dict[str, Any]:
    return {
        'bmp_send_switch': True,
        'current_preview_url': None,
        'current_preview_token': None,
        'current_preview_entry_index': None,
        'playlist_ids': [],
        'playlist_indexes': [],
        'playlist_media': [],
        'current_playlist_index': -1,
        'current_entry_index': -1,
        'current_entry_media': 'auto',
        'request_count': 0,
        'last_entry_hash': None,
        'last_entry_plugin': None,
        'pending_entry_index': None,
        'pending_entry_media': None,
        'supports_grayscale': False,
        'grayscale_send_switch': True,
        'token_sequence': 0
    }


global_state: Dict[str, Any] = {
    'rotation_master': {
        'bmp_entries': [],
        'png_entries': [],
        'hashes': [],
        'meta': [],
        'selected_ids': [],
        'version': 0,
        'has_persistent_playlist': False
    },
    'devices': {},
    'device_playlists': {},
    'device_profiles': {},
    'client_metrics': {},
    'server': {
        'uptime': 0,
        'cpu_load': 0,
        'current_time': 0
    },
    'client': {
        'battery_voltage': 0,
        'battery_voltage_max': 0
    },
    'plugins': {}
}


def _image_token_store() -> Dict[str, Dict[str, Any]]:
    return global_state.setdefault('image_tokens', {})


def register_image_token(device_id: str, token: str) -> None:
    normalized_device = (device_id or '').strip() or DEFAULT_DEVICE_ID
    if normalized_device == DEFAULT_DEVICE_ID:
        return
    token_value = (token or '').strip()
    if not token_value:
        return
    now = time()
    cutoff = now - _IMAGE_TOKEN_TTL_SECONDS
    with STATE_LOCK:
        store = _image_token_store()
        store[token_value] = {'device_id': normalized_device, 'ts': now}
        for key, payload in list(store.items()):
            if not isinstance(payload, dict):
                store.pop(key, None)
                continue
            ts = payload.get('ts')
            if not isinstance(ts, (int, float)) or ts < cutoff:
                store.pop(key, None)


def resolve_device_id_from_token(token: Optional[str]) -> Optional[str]:
    token_value = (token or '').strip()
    if not token_value:
        return None
    now = time()
    cutoff = now - _IMAGE_TOKEN_TTL_SECONDS
    with STATE_LOCK:
        store = _image_token_store()
        payload = store.get(token_value)
        if not isinstance(payload, dict):
            return None
        ts = payload.get('ts')
        if not isinstance(ts, (int, float)) or ts < cutoff:
            store.pop(token_value, None)
            return None
        device_id = payload.get('device_id')
        if isinstance(device_id, str) and device_id.strip() and device_id.strip() != DEFAULT_DEVICE_ID:
            return device_id.strip()
    return None


def _read_image_bytes(image_path: str) -> bytes:
    with open(image_path, 'rb') as file:
        return file.read()


def _digest_bytes(data: bytes) -> str:
    return sha1(data).hexdigest()


def _rotation_entry_id(plugin_name: str, bmp_path: str, png_path: str) -> str:
    plugin = (plugin_name or '').strip()
    bmp_url = utils.path_to_web_url(bmp_path)
    png_url = utils.path_to_web_url(png_path)
    return f"{plugin}:{bmp_url}|{png_url}"


def _collision_safe_rotation_id(preferred: str, disallowed: set[str]) -> str:
    if preferred not in disallowed:
        return preferred
    for idx in range(1, 100):
        candidate = f"{preferred}#{idx}"
        if candidate not in disallowed:
            return candidate
    return f"{preferred}#{len(disallowed) + 1}"


def rotation_master() -> Dict[str, Any]:
    """Return the shared rotation master structure."""
    return global_state['rotation_master']


def _selected_matches_all(master: Dict[str, Any]) -> bool:
    selected_ids = master.get('selected_ids') or []
    meta = master.get('meta') or []
    ids = [entry.get('id') for entry in meta if entry.get('id')]
    if len(selected_ids) != len(ids):
        return False
    return all(lhs == rhs for lhs, rhs in zip(selected_ids, ids))


def _prune_missing_selected_ids(master: Dict[str, Any]) -> bool:
    selected_ids = master.get('selected_ids') or []
    if not selected_ids:
        return False
    meta = master.get('meta') or []
    valid_ids = {entry.get('id') for entry in meta if entry.get('id')}
    filtered_ids: List[str] = []
    for entry in selected_ids:
        base_id, _ = _parse_playlist_entry_id(entry)
        if base_id in valid_ids:
            filtered_ids.append(entry)
    if len(filtered_ids) == len(selected_ids):
        return False
    master['selected_ids'] = filtered_ids
    return True


def persist_default_playlist(selected_ids: List[str]) -> None:
    try:
        models.save_rotation_playlist(selected_ids)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to persist default playlist: %s", exc)


def persist_device_playlist(device_id: str, selected_ids: List[str]) -> None:
    try:
        models.save_rotation_playlist(selected_ids, device_id=device_id)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to persist playlist for %s: %s", device_id, exc)


def cache_device_playlist(device_id: str, selected_ids: Optional[List[str]]) -> None:
    with STATE_LOCK:
        playlists = global_state.setdefault('device_playlists', {})
        if selected_ids is None:
            playlists.pop(device_id, None)
        else:
            playlists[device_id] = selected_ids


def _named_playlists_store() -> Dict[str, List[str]]:
    return global_state.setdefault('named_playlists', {})


def cache_named_playlist(name: str, selected_ids: Optional[List[str]]) -> None:
    normalized = (name or '').strip()
    if not normalized or normalized == DEFAULT_DEVICE_ID:
        return
    with STATE_LOCK:
        store = _named_playlists_store()
        if selected_ids is None:
            store.pop(normalized, None)
        else:
            store[normalized] = list(selected_ids)


def _device_playlist_binding_store() -> Dict[str, str]:
    return global_state.setdefault('device_playlist_bindings', {})


def cache_device_playlist_binding(device_id: str, playlist_name: Optional[str]) -> None:
    normalized_device = (device_id or '').strip() or DEFAULT_DEVICE_ID
    normalized_name = (playlist_name or '').strip() or DEFAULT_DEVICE_ID
    if normalized_device == DEFAULT_DEVICE_ID:
        return
    with STATE_LOCK:
        store = _device_playlist_binding_store()
        if normalized_name == DEFAULT_DEVICE_ID:
            store.pop(normalized_device, None)
        else:
            store[normalized_device] = normalized_name


def get_device_playlist_binding_name(device_id: str) -> Optional[str]:
    normalized_device = (device_id or '').strip() or DEFAULT_DEVICE_ID
    if normalized_device == DEFAULT_DEVICE_ID:
        return None
    with STATE_LOCK:
        store = _device_playlist_binding_store()
        cached = store.get(normalized_device)
        if cached:
            return cached
    bound = models.get_device_playlist_binding(normalized_device)
    cache_device_playlist_binding(normalized_device, bound)
    if bound and bound.strip() and bound.strip() != DEFAULT_DEVICE_ID:
        return bound.strip()
    return None


def get_named_playlist_selection(name: str) -> Optional[List[str]]:
    normalized = (name or '').strip()
    if not normalized or normalized == DEFAULT_DEVICE_ID:
        return None
    with STATE_LOCK:
        store = _named_playlists_store()
        cached = store.get(normalized)
        if cached is not None:
            return list(cached)
    selected = models.get_rotation_playlist(device_id=None, name=normalized)
    cache_named_playlist(normalized, selected)
    if selected is None:
        return None
    return list(selected)


def _ensure_device_playlist_cached(device_id: str) -> Optional[List[str]]:
    with STATE_LOCK:
        playlists = global_state.setdefault('device_playlists', {})
        if device_id in playlists:
            return playlists[device_id]
    selected_ids = models.get_rotation_playlist(device_id=device_id)
    cache_device_playlist(device_id, selected_ids)
    return selected_ids


def initialize_rotation_playlists_from_storage() -> None:
    default_playlist = models.get_rotation_playlist()
    if default_playlist is not None:
        with STATE_LOCK:
            master = rotation_master()
            master['selected_ids'] = default_playlist
            master['has_persistent_playlist'] = True
    for device_id, selected_ids in models.list_device_playlists():
        cache_device_playlist(device_id, selected_ids)

    for name, selected_ids in models.list_named_rotation_playlists():
        cache_named_playlist(name, selected_ids)

    for device_id, playlist_name in models.list_device_playlist_bindings():
        cache_device_playlist_binding(device_id, playlist_name)


def _replace_selected_hash(master: Dict[str, Any], previous_hash: Optional[str], new_hash: str) -> None:
    if not previous_hash:
        return
    selected_ids = master.get('selected_ids')
    if not selected_ids:
        return
    master['selected_ids'] = [new_hash if entry == previous_hash else entry for entry in selected_ids]


def _selected_ids_for_device(device_id: Optional[str]) -> List[str]:
    if device_id:
        bound_name = get_device_playlist_binding_name(device_id)
        if bound_name:
            bound_playlist = get_named_playlist_selection(bound_name)
            if bound_playlist:
                return list(bound_playlist)
        cached = _ensure_device_playlist_cached(device_id)
        if cached is not None:
            return list(cached)
    master = rotation_master()
    selected_ids = master.get('selected_ids') or []
    if selected_ids:
        return list(selected_ids)
    meta = master.get('meta') or []
    return [entry.get('id') for entry in meta if entry.get('id')]


def get_playlist_selection(device_id: Optional[str] = None) -> List[str]:
    return _selected_ids_for_device(device_id)


def known_device_ids(include_default: bool = True) -> List[str]:
    with STATE_LOCK:
        ids = set(global_state.get('devices', {}).keys())
        ids.update((global_state.get('device_playlists') or {}).keys())
        ids.update((global_state.get('device_playlist_bindings') or {}).keys())
        ids.update((global_state.get('client_metrics') or {}).keys())
        ids.update((global_state.get('device_profiles') or {}).keys())
    if include_default:
        ids.add(DEFAULT_DEVICE_ID)
    return sorted(ids)


def list_playlist_targets(include_default: bool = True) -> List[Dict[str, Any]]:
    assignments: List[Dict[str, Any]] = []
    if include_default:
        default_playlist = _selected_ids_for_device(None)
        assignments.append({
            'device_id': DEFAULT_DEVICE_ID,
            'friendly_name': 'Default playlist',
            'playlist': list(default_playlist),
            'count': len(default_playlist)
        })
    with STATE_LOCK:
        overrides = list((global_state.get('device_playlists') or {}).items())
    for device_id, playlist_ids in sorted(overrides, key=lambda item: item[0]):
        current_playlist = list(playlist_ids or [])
        profile = ensure_device_profile(device_id)
        assignments.append({
            'device_id': device_id,
            'friendly_name': profile.get('friendly_name') or device_id,
            'playlist': current_playlist,
            'count': len(current_playlist)
        })
    return assignments


def set_named_playlist(name: str, playlist_ids: List[str]) -> None:
    normalized = (name or '').strip()
    if not normalized or normalized == DEFAULT_DEVICE_ID:
        raise ValueError('invalid playlist name')
    if not playlist_ids:
        raise ValueError('playlist must contain at least one active entry')
    validate_playlist_ids(playlist_ids)
    models.save_rotation_playlist(list(playlist_ids), device_id=None, name=normalized)
    cache_named_playlist(normalized, list(playlist_ids))


def delete_named_playlist(name: str) -> None:
    normalized = (name or '').strip()
    if not normalized or normalized == DEFAULT_DEVICE_ID:
        raise ValueError('default playlist cannot be deleted')

    with STATE_LOCK:
        bindings = dict(_device_playlist_binding_store())
    for device_id, playlist_name in bindings.items():
        if playlist_name == normalized:
            models.delete_device_playlist_binding(device_id)
            cache_device_playlist_binding(device_id, None)
            with STATE_LOCK:
                device_state = get_device_state(device_id)
                device_state['request_count'] = 0
                device_state['current_playlist_index'] = -1
                device_state['current_entry_index'] = -1
                device_state['last_entry_hash'] = None
                device_state['last_entry_plugin'] = None
            persist_device_state(device_id, 0, -1, [], None, None)

    models.delete_named_rotation_playlist(normalized)
    cache_named_playlist(normalized, None)


def set_device_playlist_binding(device_id: str, playlist_name: Optional[str]) -> None:
    normalized_device = (device_id or '').strip() or DEFAULT_DEVICE_ID
    if normalized_device == DEFAULT_DEVICE_ID:
        raise ValueError('invalid device id')

    normalized_name = (playlist_name or '').strip() or DEFAULT_DEVICE_ID
    # Binding is the primary assignment mechanism; clear any legacy per-device overrides
    # so the binding is reflected immediately in rotation selection.
    cache_device_playlist(normalized_device, None)
    try:
        models.delete_rotation_playlist(device_id=normalized_device)
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to delete legacy rotation playlist for %s: %s", normalized_device, exc)

    if normalized_name != DEFAULT_DEVICE_ID:
        selection = get_named_playlist_selection(normalized_name)
        if not selection:
            raise ValueError('unknown playlist')
        models.set_device_playlist_binding(normalized_device, normalized_name)
        cache_device_playlist_binding(normalized_device, normalized_name)
    else:
        models.delete_device_playlist_binding(normalized_device)
        cache_device_playlist_binding(normalized_device, None)

    with STATE_LOCK:
        device_state = get_device_state(normalized_device)
        device_state['request_count'] = 0
        device_state['current_playlist_index'] = -1
        device_state['current_entry_index'] = -1
        device_state['pending_entry_index'] = None
        device_state['last_entry_hash'] = None
        device_state['last_entry_plugin'] = None
    persist_device_state(normalized_device, 0, -1, [], None, None)


def _playlist_index_map(master: Optional[Dict[str, Any]] = None) -> Dict[str, int]:
    if master is None:
        master = rotation_master()
    meta = master.get('meta') or []
    return {entry.get('id'): idx for idx, entry in enumerate(meta) if entry.get('id')}


def _parse_playlist_entry_id(entry_id: str) -> Tuple[str, Optional[str]]:
    """Parse playlist entry IDs with optional media directives.

    Supported suffixes: '@bmp'/'@mono' forces 1-bit BMP delivery; '@png'/'@gray'
    forces grayscale PNG delivery; '@auto' uses the default device behavior.

    Returns (base_id, media) where media is 'bmp', 'png', or None.
    """
    raw = (entry_id or '').strip()
    if not raw or '@' not in raw:
        return raw, None

    base, suffix = raw.rsplit('@', 1)
    suffix_norm = suffix.strip().lower()
    if suffix_norm in ('bmp', 'mono', '1bit', '1-bit'):
        return base, 'bmp'
    if suffix_norm in ('png', 'gray', 'grayscale'):
        return base, 'png'
    if suffix_norm in ('auto',):
        return base, None
    return raw, None


def _resolved_playlist_for_device(device_id: Optional[str]) -> Tuple[List[str], List[int]]:
    master = rotation_master()
    playlist_ids = _selected_ids_for_device(device_id)
    if not playlist_ids:
        playlist_ids = [entry.get('id') for entry in master.get('meta') or [] if entry.get('id')]
    mapping = _playlist_index_map(master)
    resolved_ids: List[str] = []
    resolved_indexes: List[int] = []
    resolved_media: List[str] = []
    for entry_id in playlist_ids:
        base_id, media = _parse_playlist_entry_id(entry_id)
        index = mapping.get(base_id)
        if index is None:
            continue
        resolved_ids.append(entry_id)
        resolved_indexes.append(index)
        resolved_media.append(media or 'auto')
    return resolved_ids, resolved_indexes, resolved_media


def _update_playlist_snapshot(device_id: str, device_state: Dict[str, Any]) -> Tuple[List[str], List[int]]:
    resolved_ids, resolved_indexes, resolved_media = _resolved_playlist_for_device(device_id)
    device_state['playlist_ids'] = list(resolved_ids)
    device_state['playlist_indexes'] = list(resolved_indexes)
    device_state['playlist_media'] = list(resolved_media)
    if resolved_indexes:
        max_index = len(resolved_indexes) - 1
        if device_state.get('current_playlist_index', -1) > max_index:
            device_state['current_playlist_index'] = -1
    else:
        device_state['current_playlist_index'] = -1
        device_state['current_entry_index'] = -1
        device_state['current_entry_media'] = 'auto'
    return device_state['playlist_ids'], device_state['playlist_indexes']


def _select_next_playlist_entry(device_id: str, device_state: Dict[str, Any]) -> Tuple[int, str]:
    with STATE_LOCK:
        _, playlist_indexes = _update_playlist_snapshot(device_id, device_state)
        if not playlist_indexes:
            raise RotationUnavailableError(f'No rotation entries are available for {device_id}')
        request_count = int(device_state.get('request_count') or 0)
        position = request_count % len(playlist_indexes)
        entry_index = playlist_indexes[position]
        playlist_media = device_state.get('playlist_media') or []
        media = str(playlist_media[position]) if position < len(playlist_media) else 'auto'
        if media not in ('bmp', 'png'):
            media = 'auto'
        master = rotation_master()
        meta_list = master.get('meta') or []
        if entry_index < 0 or entry_index >= len(meta_list):
            raise RotationUnavailableError(f'Playlist entry {entry_index} is invalid for {device_id}')
        entry = meta_list[entry_index]
        entry_id = entry.get('id')
        plugin_id = entry.get('plugin')
        device_state['request_count'] = request_count + 1
        device_state['current_playlist_index'] = position
        device_state['current_entry_index'] = entry_index
        device_state['current_entry_media'] = media
        device_state['last_entry_hash'] = entry_id
        device_state['last_entry_plugin'] = plugin_id
        playlist_snapshot = list(playlist_indexes)
        request_snapshot = device_state['request_count']
    persist_device_state(
        device_id,
        request_snapshot,
        position,
        playlist_snapshot,
        entry_id,
        plugin_id
    )
    return entry_index, media


def validate_playlist_ids(playlist_ids: List[str]) -> None:
    mapping = _playlist_index_map()
    unknown: List[str] = []
    for pid in playlist_ids:
        base_id, _ = _parse_playlist_entry_id(pid)
        if base_id not in mapping:
            unknown.append(pid)
    if unknown:
        raise ValueError(f'unknown playlist ids: {unknown}')


def set_default_playlist(playlist_ids: List[str]) -> None:
    master = rotation_master()
    mapping = _playlist_index_map(master)
    unknown: List[str] = []
    for pid in playlist_ids:
        base_id, _ = _parse_playlist_entry_id(pid)
        if base_id not in mapping:
            unknown.append(pid)
    if unknown:
        raise ValueError(f'unknown playlist ids: {unknown}')
    with STATE_LOCK:
        master = rotation_master()
        master['selected_ids'] = playlist_ids
        master['has_persistent_playlist'] = False  # Reset flag when user explicitly sets playlist
        master['version'] += 1
        selection_snapshot = list(master['selected_ids'])
    persist_default_playlist(selection_snapshot)


def set_device_playlist(device_id: str, playlist_ids: List[str]) -> None:
    master = rotation_master()
    mapping = _playlist_index_map(master)
    unknown: List[str] = []
    resolved_indexes: List[int] = []
    for pid in playlist_ids:
        base_id, _ = _parse_playlist_entry_id(pid)
        idx = mapping.get(base_id)
        if idx is None:
            unknown.append(pid)
        else:
            resolved_indexes.append(idx)
    if unknown:
        raise ValueError(f'unknown playlist ids: {unknown}')

    playlist_indexes = resolved_indexes
    with STATE_LOCK:
        device_state = get_device_state(device_id)
        device_state['playlist_ids'] = list(playlist_ids)
        device_state['playlist_indexes'] = list(playlist_indexes)
        device_state['playlist_media'] = []
        device_state['request_count'] = 0
        device_state['current_playlist_index'] = -1
        device_state['current_entry_index'] = -1
        device_state['current_entry_media'] = 'auto'
        device_state['last_entry_hash'] = None
        device_state['last_entry_plugin'] = None
        device_state['pending_entry_media'] = None
        cache_device_playlist(device_id, list(playlist_ids))
    persist_device_playlist(device_id, list(playlist_ids))
    persist_device_state(device_id, 0, -1, playlist_indexes, None, None)


def clear_device_playlist(device_id: str) -> None:
    with STATE_LOCK:
        playlists = global_state.setdefault('device_playlists', {})
        playlists.pop(device_id, None)
        devices = global_state.setdefault('devices', {})
        device_state = devices.setdefault(device_id, _build_device_state())
        device_state['playlist_ids'] = []
        device_state['playlist_indexes'] = []
        device_state['request_count'] = 0
        device_state['current_playlist_index'] = -1
        device_state['current_entry_index'] = -1
        device_state['last_entry_hash'] = None
        device_state['last_entry_plugin'] = None

    models.delete_rotation_playlist(device_id=device_id)
    persist_device_state(device_id, 0, -1, [], None, None)


def persist_device_state(
    device_id: str,
    request_count: int,
    playlist_position: int,
    playlist_indexes: List[int],
    current_entry_id: Optional[str],
    current_plugin_id: Optional[str]
) -> None:
    try:
        snapshot = [str(idx) for idx in playlist_indexes]
        models.save_device_state(
            device_id=device_id,
            rotation_version=request_count,
            rotation_index=playlist_position,
            rotation_hash_order=snapshot,
            last_entry_hash=current_entry_id,
            current_plugin_id=current_plugin_id
        )
    except Exception as exc:  # noqa: BLE001
        logger.warning("Failed to persist device state for %s: %s", device_id, exc)


def schedule_next_rotation_entry(device_id: str, device_state: Dict[str, Any]) -> int:
    entry_idx, media = _select_next_playlist_entry(device_id, device_state)
    with STATE_LOCK:
        device_state['pending_entry_index'] = entry_idx
        device_state['pending_entry_media'] = media
    return entry_idx


def _pop_pending_entry_index(device_state: Dict[str, Any]) -> Optional[int]:
    with STATE_LOCK:
        pending = device_state.get('pending_entry_index')
        device_state['pending_entry_index'] = None
        device_state['pending_entry_media'] = None
        return pending


def _resolve_next_entry(device_id: str, device_state: Dict[str, Any]) -> int:
    pending = _pop_pending_entry_index(device_state)
    if pending is not None:
        return pending
    entry_idx, _ = _select_next_playlist_entry(device_id, device_state)
    return entry_idx


def _rotation_frame_bytes(entry_idx: int, media: str) -> bytes:
    key = 'bmp_entries' if media == 'bmp' else 'png_entries'
    with STATE_LOCK:
        master = rotation_master()
        entries = master.get(key) or []
        if entry_idx < 0 or entry_idx >= len(entries):
            raise RotationUnavailableError(f'Rotation index {entry_idx} is out of range for {media.upper()} frames')
        return entries[entry_idx]


def get_rotation_bmp_bytes(entry_idx: int) -> bytes:
    return _rotation_frame_bytes(entry_idx, 'bmp')


def get_rotation_png_bytes(entry_idx: int) -> bytes:
    return _rotation_frame_bytes(entry_idx, 'png')


def current_frame_entry_index(device_state: Dict[str, Any]) -> Optional[int]:
    with STATE_LOCK:
        pending = device_state.get('pending_entry_index')
        if isinstance(pending, int) and pending >= 0:
            return pending
        current = device_state.get('current_entry_index')
        if isinstance(current, int) and current >= 0:
            return current
    return None


def preview_frame_entry_index(device_state: Dict[str, Any]) -> Optional[int]:
    with STATE_LOCK:
        preview = device_state.get('current_preview_entry_index')
        if isinstance(preview, int) and preview >= 0:
            return preview
    return None


def get_next_bmp_frame(device_id: str, device_state: Dict[str, Any]) -> BytesIO:
    entry_idx = _resolve_next_entry(device_id, device_state)
    bmp_bytes = get_rotation_bmp_bytes(entry_idx)
    return BytesIO(bmp_bytes)


def get_next_png_frame(device_id: str, device_state: Dict[str, Any]) -> BytesIO:
    entry_idx = _resolve_next_entry(device_id, device_state)
    png_bytes = get_rotation_png_bytes(entry_idx)
    return BytesIO(png_bytes)


def _build_rotation_meta(
    plugin_name: str,
    bmp_path: str,
    png_path: str,
    entry_id: str,
    content_hash: str,
    display_name: Optional[str] = None
) -> Dict[str, Any]:
    label = display_name or plugin_name
    return {
        'id': entry_id,
        'hash': content_hash,
        'plugin': plugin_name,
        'label': label,
        'bmp_path': abspath(bmp_path),
        'png_path': abspath(png_path),
        'url_bmp': utils.path_to_web_url(bmp_path),
        'url_png': utils.path_to_web_url(png_path)
    }


def set_primary_rotation_assets(
    plugin_name: str,
    assets: PluginOutput,
    display_name: Optional[str] = None
) -> None:
    bmp_bytes = _read_image_bytes(assets.monochrome_path)
    png_bytes = _read_image_bytes(assets.grayscale_path)
    content_hash = _digest_bytes(b"|".join((bmp_bytes, png_bytes)))
    preferred_id = _rotation_entry_id(plugin_name, assets.monochrome_path, assets.grayscale_path)

    selection_snapshot: Optional[List[str]] = None
    with STATE_LOCK:
        master = rotation_master()
        if _prune_missing_selected_ids(master):
            selection_snapshot = list(master.get('selected_ids') or [])
        bmp_entries = master.setdefault('bmp_entries', [])
        png_entries = master.setdefault('png_entries', [])
        hashes = master.setdefault('hashes', [])
        meta_list = master.setdefault('meta', [])
        had_entries = bool(bmp_entries)
        auto_fill_enabled = _selected_matches_all(master)

        previous_id = None
        if had_entries and meta_list:
            previous_id = meta_list[0].get('id')
        disallowed_ids = {entry.get('id') for entry in meta_list if entry.get('id')}
        if previous_id:
            disallowed_ids.discard(previous_id)
        entry_id = previous_id or preferred_id
        entry_id = _collision_safe_rotation_id(entry_id, disallowed_ids)
        meta_entry = _build_rotation_meta(
            plugin_name,
            assets.monochrome_path,
            assets.grayscale_path,
            entry_id,
            content_hash,
            display_name
        )

        if bmp_entries:
            bmp_entries[0] = bmp_bytes
            png_entries[0] = png_bytes
            hashes[0] = content_hash
            if meta_list:
                meta_list[0] = meta_entry
            else:
                meta_list.append(meta_entry)
        else:
            bmp_entries.append(bmp_bytes)
            png_entries.append(png_bytes)
            hashes.append(content_hash)
            meta_list.append(meta_entry)

        if not had_entries:
            master['version'] += 1

        # Only auto-fill if there's no existing playlist AND no persistent playlist was loaded
        # This prevents overwriting user-defined playlists during plugin refresh
        if not master.get('selected_ids') and not master.get('has_persistent_playlist'):
            master['selected_ids'] = [entry.get('id') for entry in meta_list if entry.get('id')]
            selection_snapshot = list(master['selected_ids'])

    if selection_snapshot is not None:
        persist_default_playlist(selection_snapshot)


def append_rotation_assets(
    plugin_name: str,
    assets: PluginOutput,
    display_name: Optional[str] = None
) -> None:
    bmp_bytes = _read_image_bytes(assets.monochrome_path)
    png_bytes = _read_image_bytes(assets.grayscale_path)
    content_hash = _digest_bytes(b"|".join((bmp_bytes, png_bytes)))
    preferred_id = _rotation_entry_id(plugin_name, assets.monochrome_path, assets.grayscale_path)
    selection_snapshot: Optional[List[str]] = None
    replaced_entry = False
    skip_append = False
    with STATE_LOCK:
        master = rotation_master()
        if _prune_missing_selected_ids(master):
            selection_snapshot = list(master.get('selected_ids') or [])
        hashes = master.setdefault('hashes', [])
        meta_list = master.setdefault('meta', [])
        bmp_entries = master.setdefault('bmp_entries', [])
        png_entries = master.setdefault('png_entries', [])
        auto_fill_enabled = _selected_matches_all(master)

        replace_index = None
        for idx, meta_entry in enumerate(meta_list):
            if meta_entry.get('plugin') == plugin_name:
                replace_index = idx
                break

        if replace_index is not None:
            previous_id = meta_list[replace_index].get('id') if replace_index < len(meta_list) else None
            disallowed_ids = {entry.get('id') for entry in meta_list if entry.get('id')}
            if previous_id:
                disallowed_ids.discard(previous_id)
            entry_id = previous_id or preferred_id
            entry_id = _collision_safe_rotation_id(entry_id, disallowed_ids)
            bmp_entries[replace_index] = bmp_bytes
            png_entries[replace_index] = png_bytes
            hashes[replace_index] = content_hash
            meta_list[replace_index] = _build_rotation_meta(
                plugin_name,
                assets.monochrome_path,
                assets.grayscale_path,
                entry_id,
                content_hash,
                display_name
            )
            selection_snapshot = list(master.get('selected_ids') or [])
            replaced_entry = True

        if replace_index is None:
            disallowed_ids = {entry.get('id') for entry in meta_list if entry.get('id')}
            entry_id = _collision_safe_rotation_id(preferred_id, disallowed_ids)

            bmp_entries.append(bmp_bytes)
            png_entries.append(png_bytes)
            hashes.append(content_hash)
            meta_list.append(
                _build_rotation_meta(
                    plugin_name,
                    assets.monochrome_path,
                    assets.grayscale_path,
                    entry_id,
                    content_hash,
                    display_name
                )
            )
            master['version'] += 1
            # Only auto-fill if there's no existing playlist AND no persistent playlist was loaded
            # This prevents overwriting user-defined playlists during plugin refresh
            if not master.get('selected_ids') and not master.get('has_persistent_playlist'):
                master['selected_ids'] = [entry.get('id') for entry in meta_list if entry.get('id')]
                selection_snapshot = list(master['selected_ids'])

    if selection_snapshot is not None:
        persist_default_playlist(selection_snapshot)

    if skip_append or replaced_entry:
        return


def build_rotation_snapshot() -> Dict[str, Any]:
    persist_default: Optional[List[str]] = None
    persist_devices: Dict[str, List[str]] = {}
    persist_named: Dict[str, List[str]] = {}

    with STATE_LOCK:
        master = rotation_master()
        if _prune_missing_selected_ids(master):
            persist_default = list(master.get('selected_ids') or [])

        meta = list(master.get('meta') or [])
        selected_ids = list(master.get('selected_ids') or [])
        version = master.get('version', 0)

        device_playlists = {
            device_id: list(playlist or [])
            for device_id, playlist in (global_state.get('device_playlists') or {}).items()
            if playlist is not None
        }
        named_playlists = {
            name: list(value or [])
            for name, value in (global_state.get('named_playlists') or {}).items()
            if value is not None
        }
        bindings = {
            device_id: playlist_name
            for device_id, playlist_name in (global_state.get('device_playlist_bindings') or {}).items()
            if playlist_name and playlist_name != DEFAULT_DEVICE_ID
        }
    entries: List[Dict[str, Any]] = []
    for entry in meta:
        entries.append({
            'id': entry.get('id'),
            'label': entry.get('label'),
            'plugin': entry.get('plugin'),
            'url_png': entry.get('url_png'),
            'url_bmp': entry.get('url_bmp')
        })

    valid_ids = {e.get('id') for e in entries if e.get('id')}
    entry_fallback = [e.get('id') for e in entries if e.get('id')]

    # Hygiene: prune IDs that no longer exist in the current rotation entries.
    original_default = list(selected_ids)
    selected_ids = [entry_id for entry_id in selected_ids if _parse_playlist_entry_id(entry_id)[0] in valid_ids]
    if selected_ids != original_default:
        persist_default = list(selected_ids)
        with STATE_LOCK:
            master = rotation_master()
            master['selected_ids'] = list(selected_ids)

    for device_id, playlist in list(device_playlists.items()):
        original = list(playlist or [])
        filtered = [entry_id for entry_id in original if _parse_playlist_entry_id(entry_id)[0] in valid_ids]
        if filtered != original:
            device_playlists[device_id] = filtered
            persist_devices[device_id] = filtered
            with STATE_LOCK:
                global_state.setdefault('device_playlists', {})[device_id] = list(filtered)

    for name, playlist in list(named_playlists.items()):
        original = list(playlist or [])
        filtered = [entry_id for entry_id in original if _parse_playlist_entry_id(entry_id)[0] in valid_ids]
        if filtered != original:
            named_playlists[name] = filtered
            persist_named[name] = filtered
            with STATE_LOCK:
                global_state.setdefault('named_playlists', {})[name] = list(filtered)

    default_playlist = selected_ids or entry_fallback
    payload = {
        'version': version,
        'playlists': {
            'default': default_playlist
        },
        'entries': entries
    }
    if device_playlists:
        payload['playlists']['devices'] = device_playlists
    if named_playlists:
        payload['playlists']['named'] = named_playlists
    if bindings:
        payload['playlists']['bindings'] = bindings

    # Persist pruned playlists back to SQLite (outside the state lock).
    if persist_default is not None:
        persist_default_playlist(list(persist_default))
    for device_id, playlist in persist_devices.items():
        persist_device_playlist(device_id, list(playlist))
    for name, playlist in persist_named.items():
        try:
            models.save_rotation_playlist(list(playlist), device_id=None, name=name)
        except Exception as exc:  # noqa: BLE001
            logger.warning("Failed to persist named playlist '%s': %s", name, exc)

    return payload


def get_rotation_entry(entry_hash: Optional[str]) -> Optional[Dict[str, Any]]:
    if not entry_hash:
        return None
    master = rotation_master()
    for entry in master.get('meta') or []:
        if entry.get('hash') == entry_hash or entry.get('id') == entry_hash:
            return entry
    return None


def _decode_persisted_playlist_indexes(device_id: str, stored_indexes: List[Any]) -> List[int]:
    parsed: List[int] = []
    if not stored_indexes:
        return parsed
    mapping: Optional[Dict[str, int]] = None
    for value in stored_indexes:
        if isinstance(value, int):
            parsed.append(value)
            continue
        try:
            parsed.append(int(value))
            continue
        except (TypeError, ValueError):
            entry_id = str(value).strip()
            if not entry_id:
                continue
            if mapping is None:
                mapping = _playlist_index_map()
            resolved_index = mapping.get(entry_id)
            if resolved_index is None:
                logger.info("Dropping legacy rotation entry '%s' for %s", entry_id, device_id)
                continue
            parsed.append(resolved_index)
    return parsed


def get_device_state(device_id: str) -> Dict[str, Any]:
    created = False
    with STATE_LOCK:
        devices = global_state.setdefault('devices', {})
        if device_id not in devices:
            devices[device_id] = _build_device_state()
            created = True
        state = devices[device_id]
    ensure_device_profile(device_id)
    get_client_metrics(device_id)
    if created:
        persisted = models.get_device_state(device_id)
        if persisted:
            with STATE_LOCK:
                state['request_count'] = int(persisted.get('rotation_version') or 0)
                state['current_playlist_index'] = int(persisted.get('rotation_index', -1))
                stored_indexes = persisted.get('rotation_hash_order') or []
                parsed_indexes = _decode_persisted_playlist_indexes(device_id, stored_indexes)
                state['playlist_indexes'] = parsed_indexes
                state['last_entry_hash'] = persisted.get('last_entry_hash')
                state['last_entry_plugin'] = persisted.get('current_plugin_id')
                if 0 <= state['current_playlist_index'] < len(parsed_indexes):
                    state['current_entry_index'] = parsed_indexes[state['current_playlist_index']]
                else:
                    state['current_playlist_index'] = -1
                    state['current_entry_index'] = -1
    if 'grayscale_send_switch' not in state:
        state['grayscale_send_switch'] = True
    return state


def _extract_device_id(request: Optional[Request]) -> str:
    if not request:
        return DEFAULT_DEVICE_ID
    for header in ('x-device-id', 'device-id', 'id'):
        value = request.headers.get(header)
        if value:
            return value.strip()
    query_params = request.query_params
    for key in ('device', 'device_id', 'id', 'deviceId'):
        candidate = query_params.get(key)
        if candidate:
            return candidate.strip()
    return DEFAULT_DEVICE_ID


def request_base_url(request: Optional[Request]) -> str:
    if not request:
        return get_server_base_url()
    base = str(request.base_url)
    if base:
        return base[:-1] if base.endswith('/') else base
    host = request.headers.get('host')
    scheme = request.url.scheme if request.url.scheme else config.SERVER_SCHEME
    if host:
        return f"{scheme}://{host}"
    return get_server_base_url()


def get_device_state_from_request(request: Request) -> Tuple[str, Dict[str, Any]]:
    device_id = _extract_device_id(request)
    return device_id, get_device_state(device_id)