update to modern python unions, add python3.14 to ci

Author: purarue

.github/workflows/ci.yaml

@@ -11,7 +11,7 @@ jobs:
     runs-on: ubuntu-latest
     strategy:
       matrix:
-        python-version: ["3.10", "3.11", "3.12", "3.13"]
+        python-version: ["3.10", "3.11", "3.12", "3.13", "3.14"]
 
     steps:
       - uses: actions/checkout@v5

mpv_history_daemon/__main__.py

@@ -30,8 +30,8 @@ def cli():
 def _parse_polling(
     ctx: click.Context,
     param: click.Parameter,
-    value: Union[str, int],
-) -> Union[Literal["disabled"], int]:
+    value: str | int,
+) -> Literal["disabled"] | int:
     if value == "disabled":
         return "disabled"
     if isinstance(value, int):
@@ -78,9 +78,9 @@ def daemon(
     socket_dir: str,
     data_dir: str,
     log_file: str,
-    scan_time: Union[Literal["disabled"], int],
-    write_period: Optional[int],
-    socket_class_qualname: Optional[str],
+    scan_time: Literal["disabled"] | int,
+    write_period: int | None,
+    socket_class_qualname: str | None,
 ) -> None:
     """
     Socket dir is the directory with mpv sockets (/tmp/mpvsockets, probably)
@@ -177,7 +177,7 @@ def parse(data_files: Sequence[str], all_events: bool, debug: bool) -> None:
     help="If files have been modified in this amount of time, don't merge them",
 )
 def merge(
-    data_files: Sequence[str], move: Optional[Path], write_to: Path, mtime_seconds: int
+    data_files: Sequence[str], move: Path | None, write_to: Path, mtime_seconds: int
 ) -> None:
     """
     merges multiple files into a single merged event file

mpv_history_daemon/daemon.py

@@ -102,7 +102,7 @@ def __init__(
         socket: MPV,
         socket_loc: str,
         data_dir: str,
-        write_period: Optional[int] = None,
+        write_period: int | None = None,
     ):
         self.socket = socket
         self.socket_loc = socket_loc
@@ -159,7 +159,7 @@ def write(self) -> None:
         ) as event_f:
             event_f.write(serialized)
 
-    def nevent(self, event_name: str, event_data: Optional[Any] = None) -> None:
+    def nevent(self, event_name: str, event_data: Any | None = None) -> None:
         """add an event"""
         ct = time()
         logger.debug(f"{self.socket_time}|{ct}|{event_name}|{event_data}")
@@ -278,8 +278,8 @@ def __init__(
         data_dir: str,
         *,
         autostart: bool = True,
-        write_period: Optional[int],
-        poll_time: Optional[int] = 10,
+        write_period: int | None,
+        poll_time: int | None = 10,
         socket_data_cls: type[SocketData] = SocketData,
     ):
         self.data_dir: str = data_dir
@@ -299,7 +299,7 @@ def scan_sockets(self) -> None:
         """
         Look for any new sockets at socket_dir, remove any dead ones
         """
-        socket_loc_scoped: Optional[str] = None  # to access this after the try/except
+        socket_loc_scoped: str | None = None  # to access this after the try/except
         try:
             # iterate through all files
             for socket_name in os.listdir(self.socket_dir):
@@ -497,9 +497,9 @@ def run(
     socket_dir: str,
     data_dir: str,
     log_file: str,
-    write_period: Optional[int],
+    write_period: int | None,
     socket_data_cls: type[SocketData],
-    poll_time: Optional[int],
+    poll_time: int | None,
 ) -> None:
     # if the daemon launched before any mpv instances
     if not os.path.exists(socket_dir):

mpv_history_daemon/events.py

@@ -13,8 +13,8 @@
     Any,
     Optional,
     Union,
-    Callable,
 )
+from collections.abc import Callable
 from collections.abc import Iterator, Sequence
 
 from logzero import setup_logger  # type: ignore[import]
@@ -29,7 +29,7 @@
 EventData = Any
 
 
-def parse_datetime_sec(d: Union[str, float, int]) -> datetime:
+def parse_datetime_sec(d: str | float | int) -> datetime:
     return datetime.fromtimestamp(float(d), tz=timezone.utc)
 
 
@@ -41,7 +41,7 @@ class Action(NamedTuple):
     since_started: float
     action: str
     # this can be None if its a livestream, those dont have a percent-pos
-    percentage: Optional[float]
+    percentage: float | None
 
 
 class Media(NamedTuple):
@@ -50,9 +50,9 @@ class Media(NamedTuple):
     start_time: datetime  # when the media started playing
     end_time: datetime  # when the media was closed/finished
     pause_duration: float  # how long the media was paused for (typically 0)
-    media_duration: Optional[float]  # length of the media
+    media_duration: float | None  # length of the media
     # title of the media (if URL, could be <title>...</title> from ytdl
-    media_title: Optional[str]
+    media_title: str | None
     # additional metadata on what % I was through the media while pausing/playing/seeking
     actions: list[Action]
     metadata: dict[str, Any]  # metadata from the file, if it exists
@@ -147,7 +147,7 @@ def _read_event_stream(
             continue
         if d["end_time"] < d["start_time"]:
             logger.warning(f"End time is less than start time! {d}")
-        fdur: Optional[float] = None
+        fdur: float | None = None
         if "duration" in d:
             fdur = float(d["duration"])
         start_time = parse_datetime_sec(float(d["start_time"]))
@@ -223,7 +223,7 @@ def _reconstruct_event_stream(
     # exec "$mpv_path" "${mpv_options[@]}"
     #
     # get when mpv launched from the filename
-    start_time: Optional[float] = None
+    start_time: float | None = None
     try:
         start_time = float(int(PurePath(filename).stem) / 1e9)
     except ValueError as ve:
@@ -242,7 +242,7 @@ def _reconstruct_event_stream(
     # used to help determine state
     is_playing = True  # assume playing at beginning
     pause_duration = 0.0  # pause duration for this entry
-    pause_start_time: Optional[float] = None  # if the entry is paused, when it started
+    pause_start_time: float | None = None  # if the entry is paused, when it started
     actions: dict[float, tuple[str, float]] = {}
 
     # a heuristic to determine if this is an old file, is-paused can be useful

mpv_history_daemon/utils.py

@@ -21,7 +21,7 @@ class MusicMetadata(NamedTuple):
 def music_parse_metadata_from_blob(
     data: dict[str, Any],
     strip_whitespace: bool = False,
-) -> Optional[MusicMetadata]:
+) -> MusicMetadata | None:
     if "title" not in data or "album" not in data or "artist" not in data:
         return None
     title = data["title"]
@@ -54,13 +54,13 @@ class MediaAllowed:
     def __init__(
         self,
         *,
-        allow_prefixes: Optional[list[str]] = None,
-        ignore_prefixes: Optional[list[str]] = None,
-        allow_extensions: Optional[list[str]] = None,
-        ignore_extensions: Optional[list[str]] = None,
+        allow_prefixes: list[str] | None = None,
+        ignore_prefixes: list[str] | None = None,
+        allow_extensions: list[str] | None = None,
+        ignore_extensions: list[str] | None = None,
         allow_stream: bool = False,
         strict: bool = True,
-        logger: Optional[logging.Logger] = None,
+        logger: logging.Logger | None = None,
     ):
         self.allow_prefixes = allow_prefixes if allow_prefixes else []
         self.ignored_prefixes = ignore_prefixes if ignore_prefixes else []