diff --git a/lib/streaming_providers/base/epg/epg_manager.py b/lib/streaming_providers/base/epg/epg_manager.py index 9eb8d04..679756a 100644 --- a/lib/streaming_providers/base/epg/epg_manager.py +++ b/lib/streaming_providers/base/epg/epg_manager.py @@ -311,4 +311,4 @@ class EPGManager: Returns: True if mapping exists """ - return self.mapping.has_mapping(provider_name, channel_id) \ No newline at end of file + return self.mapping.has_mapping(provider_name, channel_id) diff --git a/lib/streaming_providers/base/models/epg_models.py b/lib/streaming_providers/base/models/epg_models.py index e1b4176..b1d079c 100644 --- a/lib/streaming_providers/base/models/epg_models.py +++ b/lib/streaming_providers/base/models/epg_models.py @@ -2,12 +2,36 @@ # streaming_providers/base/models/epg_models.py """ EPG Models - Data classes for Electronic Program Guide entries -Based on Kodi PVR EPG Tag specification +Based on Kodi PVR EPG Tag specification (ETSI EN 300 468 DVB-SI standard) """ from dataclasses import dataclass from typing import Optional, List from datetime import datetime +from enum import IntEnum + +# EPG Constants from Kodi PVR specification +EPG_TAG_INVALID_UID = 0 +"""Special broadcast ID value to indicate invalid/unset EPG event UID.""" + +EPG_TAG_INVALID_SERIES_EPISODE = -1 +"""Special value for series/episode/part numbers to indicate not applicable.""" + +EPG_TIMEFRAME_UNLIMITED = -1 +"""Special timeframe value to indicate no time restrictions.""" + +EPG_STRING_TOKEN_SEPARATOR = "," +"""Separator for multiple values in string fields (cast, directors, writers).""" + + +class EPGEventState(IntEnum): + """ + EPG event states for event lifecycle callbacks. + Used with EpgEventStateChange() callback in C++ PVR client. + """ + CREATED = 0 # Event created + UPDATED = 1 # Event updated + DELETED = 2 # Event deleted @dataclass @@ -119,8 +143,18 @@ class EPGEntry: flags: Optional[int] = None """ Bit field flags for EPG entry properties. - Common flags: IS_SERIES=0x01, IS_NEW=0x02, IS_PREMIERE=0x04, - IS_FINALE=0x08, IS_LIVE=0x10 + Combine flags using bitwise OR: EPGFlags.IS_SERIES | EPGFlags.IS_NEW + + Available flags: + - EPGFlags.UNDEFINED (0x00): Nothing special + - EPGFlags.IS_SERIES (0x01): Part of a series + - EPGFlags.IS_NEW (0x02): New episode/content + - EPGFlags.IS_PREMIERE (0x04): Premiere episode + - EPGFlags.IS_FINALE (0x08): Finale episode + - EPGFlags.IS_LIVE (0x10): Live broadcast + + Example: + flags = EPGFlags.IS_SERIES | EPGFlags.IS_NEW """ def to_dict(self) -> dict: @@ -223,10 +257,121 @@ class EPGEntry: # and doesn't start after range ends return self.end > start_time and self.start < end_time + def has_flag(self, flag: int) -> bool: + """ + Check if a specific flag is set. + + Args: + flag: Flag to check (use EPGFlags constants) + + Returns: + True if flag is set + + Example: + if entry.has_flag(EPGFlags.IS_LIVE): + print("Live broadcast") + """ + if self.flags is None: + return False + return EPGFlags.has_flag(self.flags, flag) + + def set_flag(self, flag: int) -> None: + """ + Set a specific flag (adds to existing flags). + + Args: + flag: Flag to set (use EPGFlags constants) + + Example: + entry.set_flag(EPGFlags.IS_NEW) + """ + if self.flags is None: + self.flags = flag + else: + self.flags |= flag + + def clear_flag(self, flag: int) -> None: + """ + Clear a specific flag (removes from existing flags). + + Args: + flag: Flag to clear (use EPGFlags constants) + + Example: + entry.clear_flag(EPGFlags.IS_NEW) + """ + if self.flags is not None: + self.flags &= ~flag + + @property + def is_series(self) -> bool: + """Check if this entry is part of a series.""" + return self.has_flag(EPGFlags.IS_SERIES) + + @property + def is_new(self) -> bool: + """Check if this entry is flagged as new.""" + return self.has_flag(EPGFlags.IS_NEW) + + @property + def is_premiere(self) -> bool: + """Check if this entry is a premiere.""" + return self.has_flag(EPGFlags.IS_PREMIERE) + + @property + def is_finale(self) -> bool: + """Check if this entry is a finale.""" + return self.has_flag(EPGFlags.IS_FINALE) + + @property + def is_live(self) -> bool: + """Check if this entry is a live broadcast.""" + return self.has_flag(EPGFlags.IS_LIVE) + + @staticmethod + def join_string_list(items: List[str]) -> str: + """ + Join list of strings using EPG_STRING_TOKEN_SEPARATOR. + Useful for cast, directors, writers fields when converting from lists. + + Args: + items: List of strings to join + + Returns: + Joined string using EPG separator + + Example: + cast_str = EPGEntry.join_string_list(["Actor 1", "Actor 2", "Actor 3"]) + # Returns: "Actor 1,Actor 2,Actor 3" + """ + return EPG_STRING_TOKEN_SEPARATOR.join(items) + + @staticmethod + def split_string_list(text: str) -> List[str]: + """ + Split string using EPG_STRING_TOKEN_SEPARATOR. + Useful for parsing cast, directors, writers fields. + + Args: + text: String to split + + Returns: + List of strings + + Example: + cast = EPGEntry.split_string_list("Actor 1,Actor 2,Actor 3") + # Returns: ["Actor 1", "Actor 2", "Actor 3"] + """ + if not text: + return [] + return [item.strip() for item in text.split(EPG_STRING_TOKEN_SEPARATOR) if item.strip()] + def __post_init__(self): """Validate required fields after initialization.""" - if self.broadcast_id <= 0: - raise ValueError("broadcast_id must be positive") + if self.broadcast_id <= EPG_TAG_INVALID_UID: + raise ValueError( + f"broadcast_id must be greater than EPG_TAG_INVALID_UID ({EPG_TAG_INVALID_UID})" + ) if not self.title or not self.title.strip(): raise ValueError("title is required and cannot be empty") @@ -237,15 +382,116 @@ class EPGEntry: if self.end <= self.start: raise ValueError("end time must be after start time") + # Validate episode numbers if set + if self.season_number is not None and self.season_number < EPG_TAG_INVALID_SERIES_EPISODE: + raise ValueError( + f"season_number must be >= EPG_TAG_INVALID_SERIES_EPISODE ({EPG_TAG_INVALID_SERIES_EPISODE})" + ) + + if self.episode_number is not None and self.episode_number < EPG_TAG_INVALID_SERIES_EPISODE: + raise ValueError( + f"episode_number must be >= EPG_TAG_INVALID_SERIES_EPISODE ({EPG_TAG_INVALID_SERIES_EPISODE})" + ) + + if self.episode_part_number is not None and self.episode_part_number < EPG_TAG_INVALID_SERIES_EPISODE: + raise ValueError( + f"episode_part_number must be >= EPG_TAG_INVALID_SERIES_EPISODE ({EPG_TAG_INVALID_SERIES_EPISODE})" + ) + # Constants matching C++ EPG_TAG_FLAG values class EPGFlags: - """Bit flags for EPG entry properties.""" - IS_SERIES = 0x01 - IS_NEW = 0x02 - IS_PREMIERE = 0x04 - IS_FINALE = 0x08 - IS_LIVE = 0x10 + """ + Bit field flags for EPG entry properties. + Based on Kodi's EPG_TAG_FLAG enum. + + These can be combined using bitwise OR operator: + flags = EPGFlags.IS_SERIES | EPGFlags.IS_NEW + + Example: + entry = EPGEntry( + ..., + flags=EPGFlags.IS_SERIES | EPGFlags.IS_PREMIERE + ) + """ + UNDEFINED = 0x00 # 0000 0000 : Nothing special to say about this entry + IS_SERIES = 0x01 # 0000 0001 : This EPG entry is part of a series + IS_NEW = 0x02 # 0000 0010 : This EPG entry will be flagged as new + IS_PREMIERE = 0x04 # 0000 0100 : This EPG entry will be flagged as a premiere + IS_FINALE = 0x08 # 0000 1000 : This EPG entry will be flagged as a finale + IS_LIVE = 0x10 # 0001 0000 : This EPG entry will be flagged as live + + @staticmethod + def has_flag(flags: int, flag: int) -> bool: + """ + Check if a specific flag is set. + + Args: + flags: Combined flags value + flag: Flag to check for + + Returns: + True if flag is set + + Example: + if EPGFlags.has_flag(entry.flags, EPGFlags.IS_LIVE): + print("This is a live broadcast") + """ + return (flags & flag) == flag + + @staticmethod + def combine(*flags: int) -> int: + """ + Combine multiple flags using bitwise OR. + + Args: + *flags: Variable number of flag values + + Returns: + Combined flags value + + Example: + combined = EPGFlags.combine( + EPGFlags.IS_SERIES, + EPGFlags.IS_NEW, + EPGFlags.IS_PREMIERE + ) + """ + result = 0 + for flag in flags: + result |= flag + return result + + @staticmethod + def get_flag_names(flags: int) -> List[str]: + """ + Get list of flag names that are set. + + Args: + flags: Combined flags value + + Returns: + List of flag names + + Example: + flags = EPGFlags.IS_SERIES | EPGFlags.IS_NEW + names = EPGFlags.get_flag_names(flags) + # Returns: ["IS_SERIES", "IS_NEW"] + """ + flag_map = { + EPGFlags.IS_SERIES: "IS_SERIES", + EPGFlags.IS_NEW: "IS_NEW", + EPGFlags.IS_PREMIERE: "IS_PREMIERE", + EPGFlags.IS_FINALE: "IS_FINALE", + EPGFlags.IS_LIVE: "IS_LIVE", + } + + result = [] + for flag_value, flag_name in flag_map.items(): + if flags & flag_value: + result.append(flag_name) + + return result if result else ["UNDEFINED"] # Genre type constants based on ETSI EN 300 468 V1.14.1 (DVB-SI EIT content descriptor) @@ -407,4 +653,23 @@ class EPGGenreSubtype: # Legacy alias for backwards compatibility with existing code -PVREPGTag = EPGEntry \ No newline at end of file +PVREPGTag = EPGEntry + +# Export all public symbols +__all__ = [ + # Main classes + 'EPGEntry', + 'PVREPGTag', # Legacy alias + + # Constants + 'EPG_TAG_INVALID_UID', + 'EPG_TAG_INVALID_SERIES_EPISODE', + 'EPG_TIMEFRAME_UNLIMITED', + 'EPG_STRING_TOKEN_SEPARATOR', + + # Enums and flags + 'EPGEventState', + 'EPGFlags', + 'EPGGenre', + 'EPGGenreSubtype', +] \ No newline at end of file