fix vlc is_playing

.gitignore

@@ -1,12 +1,7 @@
-# Python
 **/*.egg-info
 **/__pycache__/
 .venv/
 
-# Tooling
-.mypy_cache/
-.ruff_cache/
-
 # Testing
 config.json
 scripts/

microkodi/cec_handler.py

@@ -0,0 +1,10 @@
+import cec
+
+class CECHandler:
+    def __init__(self):
+        cec.init()
+
+    def power_on_if_needed(self):
+        tc = cec.Device(cec.CECDEVICE_TV)
+        if not tc.is_on():
+            tc.power_on()

microkodi/config.py

@@ -5,7 +5,6 @@ from typing import Any
 
 from microkodi.helpers import recursive_dict_merge
 
-
 @dataclass
 class Config:
     host: str
@@ -42,10 +41,6 @@ class Config:
     # Enables the use of CEC
     cec: bool
     
-    # Webhook to trigger to turn on power to the TV
-    tv_power_webhook: str | None
-
-
 def load_config(config_path: Path | None) -> Config:
     if config_path is None:
         config_data = {}
@@ -73,5 +68,4 @@ def load_config(config_path: Path | None) -> Config:
         ),
         options=config_data.get("options", {}),
         cec=config_data.get("cec", False),
-        tv_power_webhook=config_data.get("tv_power_webhook"),
     )

microkodi/helpers.py

@@ -1,24 +1,19 @@
 from dataclasses import dataclass
 from typing import TypeVar, ParamSpec, Callable
+from functools import wraps
 
 P = ParamSpec("P")
 T = TypeVar("T")
-
-
 def after(func: Callable[[], None]) -> Callable[P, T]:
     """Runs @func after the decorated function exits."""
-
     def decorator(f: Callable[P, T]) -> Callable[P, T]:
         def inner(*args: P.args, **kwargs: P.kwargs) -> T:
             ret = f(*args, **kwargs)
             func()
             return ret
-
         return inner
-
     return decorator
 
-
 @dataclass
 class KodiTime:
     hours: int
@@ -28,7 +23,6 @@ class KodiTime:
     def to_seconds(self) -> int:
         return self.hours * 3600 + self.minutes * 59 + self.seconds
 
-
 def seconds_to_kodi_format(seconds: int) -> KodiTime:
     """Convert seconds into hours, minutes, and seconds as Kodi wants that."""
     hours = seconds // 3600
@@ -42,6 +36,12 @@ def seconds_to_kodi_format(seconds: int) -> KodiTime:
     )
 
 
+def find(l: list[T], pred: Callable[[T], bool]) -> T | None:
+    for i in l:
+        if pred(i):
+            return i
+    return None
+
 def recursive_dict_merge(base: dict, update: dict) -> dict:
     unique_keys = set([*base.keys(), *update.keys()])
     out = {}

microkodi/jsonrpc.py

@@ -1,18 +1,16 @@
 import json
-from typing import Any
+from typing import Any, Callable
+from dataclasses import dataclass
 from urllib.parse import urlparse
 import logging
 import base64
 import sys
-import importlib
 
 from http.server import BaseHTTPRequestHandler
 
 from microkodi.repository import I
 from microkodi.programs import Program, PlayerInfo
 from microkodi.helpers import KodiTime, seconds_to_kodi_format, after
-from microkodi.config import Config
-
 
 def jsonrpc_response(id: int, payload: dict[str, Any]) -> dict[str, Any]:
     return {
@@ -21,7 +19,6 @@ def jsonrpc_response(id: int, payload: dict[str, Any]) -> dict[str, Any]:
         "id": id,
     }
 
-
 class JsonRpcObject:
     logger: logging.Logger
 
@@ -74,8 +71,9 @@ class ApplicationRpcObject:
                 return "LOL"
             else:
                 return "Unknown"
-
-        return {prop: _get_property(prop) for prop in params["properties"]}
+        return {
+            prop: _get_property(prop) for prop in params["properties"]
+        }
 
 
 class PlayerRpcObject(JsonRpcObject):
@@ -131,9 +129,7 @@ class PlayerRpcObject(JsonRpcObject):
             self.logger.warn("Trying to toggle player that is not running")
             return "OK"
         if params["play"] != "toggle":
-            self.logger.warn(
-                "Trying to call PlayPause with unknown play: '%s'", params["play"]
-            )
+            self.logger.warn("Trying to call PlayPause with unknown play: '%s'", params["play"])
             return "Err"
 
         playing = self._active_program.is_playing()
@@ -157,7 +153,8 @@ class PlayerRpcObject(JsonRpcObject):
     def open(self, params: dict[str, Any]) -> Any:
         # Turn on the TV
         config: Config = I.get("Config")
-        I.get("TVHandler").turn_on_tv()
+        if config.cec:
+            I.get("CECHandler").power_on_if_needed()
 
         I.get("DataBridge").set_loading(True)
         url = urlparse(params["item"]["file"])
@@ -171,30 +168,28 @@ class PlayerRpcObject(JsonRpcObject):
         # Find out what player class to use
         scheme_configuration = config.players.get(url.scheme)
         if scheme_configuration is None:
-            I.get("DataBridge").notification.emit(
-                f"No player available for {url.scheme}"
-            )
+            I.get("DataBridge").notification.emit(f"No player available for {url.scheme}")
             self.logger.warn("Client requested unknown scheme: '%s'", url.scheme)
-            return {"error": "invalid protocol"}
+            return {
+                "error": "invalid protocol"
+            }
         player_class_name = scheme_configuration.get(url.netloc)
         if player_class_name is None:
             player_class_name = scheme_configuration.get("*")
         if player_class_name is None:
-            I.get("DataBridge").notification.emit(
-                f"No player available for {url.netloc}"
-            )
+            I.get("DataBridge").notification.emit(f"No player available for {url.netloc}")
             self.logger.warn("No player was picked for url '%s'", url)
-            return {"error": "invalid protocol"}
+            return {
+                "error": "invalid protocol"
+            }
 
         # Try to import the class
         *module_parts, class_name = player_class_name.split(".")
         module_name = ".".join(module_parts)
         program_cls = None
         if module_name not in sys.modules:
-            self.logger.debug(
-                "Trying to import %s to get class %s", module_name, class_name
-            )
-            spec = importlib.util.find_spec(module_name)
+            self.logger.debug("Trying to import %s to get class %s", module_name, class_name)
+            spec = importlib.util.find_spec(module)
             module = importlib.util.module_from_spec(spec)
             sys.modules[module_name] = module
             spec.loader.exec_module(module)
@@ -203,7 +198,9 @@ class PlayerRpcObject(JsonRpcObject):
         if program_cls is None:
             I.get("DataBridge").notification.emit("Could not start player")
             self.logger.warn("Class %s not found in module %s", class_name, module_name)
-            return {"error": "invalid protocol"}
+            return {
+                "error": "invalid protocol"
+            }
  
         if self._active_program is not None:
             if isinstance(self._active_program, program_cls):
@@ -228,7 +225,9 @@ class PlayerRpcObject(JsonRpcObject):
             return name if self.is_active else ""
 
         player_info = self._active_program.get_player_info()
-        item = {prop: _get_property(prop, player_info) for prop in params["properties"]}
+        item = {
+            prop: _get_property(prop, player_info) for prop in params["properties"]
+        }
         return {
             "item": {
                 **item,
@@ -237,19 +236,13 @@ class PlayerRpcObject(JsonRpcObject):
         }
 
     def get_properties(self, params: dict[str, Any]) -> Any:
-        def _get_property(
-            name: str, player_info: PlayerInfo
-        ) -> str | bool | dict[str, Any]:
+        def _get_property(name: str, player_info: PlayerInfo) -> str | bool | dict[str, Any]:
             if name == "currentsubtitle":
                 return None
             elif name == "partymode":
                 return False
             elif name == "percentage":
-                return (
-                    player_info.position / player_info.runtime
-                    if player_info.runtime > 0
-                    else 0.0
-                )
+                return player_info.position / player_info.runtime if player_info.runtime > 0 else 0.0
             elif name == "playlistid":
                 return 0
             elif name == "position":
@@ -289,10 +282,10 @@ class PlayerRpcObject(JsonRpcObject):
             else:
                 self.logger.error("Unknown param '%s'", name)
                 return "Unknown"
-
         player_info = self._active_program.get_player_info()
-        return {prop: _get_property(prop, player_info) for prop in params["properties"]}
-
+        return {
+            prop: _get_property(prop, player_info) for prop in params["properties"]
+        }
 
 class PlaylistRpcObject(JsonRpcObject):
     def handle(self, method: str, params: dict[str, Any]) -> Any:
@@ -305,9 +298,7 @@ class PlaylistRpcObject(JsonRpcObject):
 
     def clear(self, params: dict[str, Any]) -> Any:
         playlistid: int = params["playlistid"]
-        self.logger.warn(
-            "Asked to empty playlist %d but we're just pretending to do so", playlistid
-        )
+        self.logger.warn("Asked to empty playlist %d but we're just pretending to do so", playlistid)
         return "OK"
 
     def get_playlists(self, params: dict[str, Any]) -> Any:
@@ -316,7 +307,6 @@ class PlaylistRpcObject(JsonRpcObject):
     def get_items(self, params: dict[str, Any]) -> Any:
         return "ERR"
 
-
 class GlobalMethodHandler:
     objects: dict[str, JsonRpcObject]
 
@@ -341,6 +331,7 @@ class JsonRpcHandler(BaseHTTPRequestHandler):
         logger = logging.getLogger("JsonRpcHandler")
         logger.debug("GET %s", self.path)
 
+
     def do_POST(self):
         logger = logging.getLogger("JsonRpcHandler")
         if self.path != "/jsonrpc":
@@ -371,24 +362,18 @@ class JsonRpcHandler(BaseHTTPRequestHandler):
         if authenticated:
             auth: str | None = self.headers.get("Authorization")
             if auth is None:
-                logger.warn(
-                    "Client provided no Authorization header. Method: %s", method_name
-                )
+                logger.warn("Client provided no Authorization header. Method: %s", method)
                 self.send_error(401)
                 return
 
             auth_type, auth_payload = auth.split(" ")
             if auth_type != "Basic":
-                logger.warn(
-                    "Client provided no Basic authorization. Method: %s", method_name
-                )
+                logger.warn("Client provided no Basic authorization. Method: %s", method)
                 self.send_error(401)
                 return
             credentials = base64.b64decode(auth_payload).decode("utf-8")
             if credentials != "kodi:1234":
-                logger.warn(
-                    "Rejecting request due to wrong credentials. Method: %s", method_name
-                )
+                logger.warn("Rejecting request due to wrong credentials. Method: %s", method)
                 self.send_error(403)
                 return
 

microkodi/main.py

@@ -5,6 +5,7 @@ import threading
 import argparse
 from pathlib import Path
 import importlib.util
+import time
 
 from PySide6.QtGui import QGuiApplication
 from PySide6.QtQml import QQmlApplicationEngine
@@ -12,7 +13,7 @@ from PySide6.QtQml import QQmlApplicationEngine
 from microkodi.jsonrpc import JsonRpcHandler, GlobalMethodHandler
 from microkodi.ui.bridge import DataBridge
 from microkodi.config import Config, load_config
-from microkodi.cec_handler import TVHandler
+from microkodi.cec_handler import CECHandler
 from microkodi.repository import I
 
 
@@ -24,7 +25,9 @@ def run_kodi_server():
     I.register("GlobalMethodHandler", method_handler)
 
     # Setup CEC
-    I.register("TVHandler", TVHandler())
+    if config.cec:
+        I.register("CECHandler", CECHandler())
+        logger.info("Enabling CEC support")
 
     # Load extra plugins
     if config.scripts:
@@ -50,25 +53,22 @@ def run_kodi_server():
             init_method = getattr(module, "init")
             init_method()
 
-    httpd = HTTPServer(
-        (
-            config.host,
-            config.port,
-        ),
-        JsonRpcHandler,
-    )
+
+
+    server = HTTPServer
+    httpd = HTTPServer((config.host, config.port,), JsonRpcHandler)
     logger.info("Starting server on %s:%i", config.host, config.port)
     httpd.serve_forever()
     logger.info("Shutting down server")
 
-
-def main():
+if __name__ == "__main__":
     parser = argparse.ArgumentParser()
     parser.add_argument("--config", "-c", type=Path, help="Location of the config file")
     parser.add_argument("--debug", action="store_true", default=False)
     options = parser.parse_args()
 
     logging.basicConfig(level=logging.DEBUG if options.debug else logging.INFO)
+    logger = logging.getLogger("ui")
 
     # Load the config
     config = load_config(options.config)
@@ -100,7 +100,3 @@ def main():
     exit_code = app.exec()
     del engine
     sys.exit(exit_code)
-
-
-if __name__ == "__main__":
-    main()

microkodi/process.py

@@ -2,7 +2,6 @@ from threading import Thread
 import subprocess
 from typing import Callable
 
-
 def nonblocking_run(args: list[str], when_done: Callable[[], None]) -> subprocess.Popen:
     def _run(process: subprocess.Popen):
         process.wait() 

microkodi/programs/__init__.py

@@ -4,7 +4,6 @@ import logging
 
 from microkodi.helpers import KodiTime
 
-
 @dataclass
 class PlayerInfo:
     runtime: int
@@ -12,7 +11,6 @@ class PlayerInfo:
     title: str
     playing: bool
 
-
 class Program:
     logger: logging.Logger
 

microkodi/programs/mpv.py

@@ -1,7 +1,8 @@
 import subprocess
 from typing import Any, Callable
-from urllib.parse import ParseResult, urlunparse
+from urllib.parse import ParseResult, urlunparse, urldefrag, unquote
 import logging
+from dataclasses import dataclass
 import os
 import json
 import socket
@@ -11,7 +12,6 @@ from microkodi.config import Config
 from microkodi.programs import PlayerInfo, Program
 from microkodi.repository import I
 
-
 class MpvConfig:
     # Mapping of file extensions to a potential transformer function
     _ext_map: dict[str, Callable[[ParseResult], tuple[list[str], str]]]
@@ -19,23 +19,14 @@ class MpvConfig:
     def __init__(self):
         self._ext_map = {}
 
-    def register_ext_transformer(
-        self, ext: str, transformer: Callable[[ParseResult], tuple[list[str], str]]
-    ):
-        logging.getLogger("MpvConfig").debug(
-            "Registering transformer for .%s URLs", ext
-        )
+    def register_ext_transformer(self, ext: str, transformer: Callable[[ParseResult], tuple[list[str], str]]):
+        logging.getLogger("MpvConfig").debug("Registering transformer for .%s URLs", ext)
         self._ext_map[ext] = transformer
 
-    def get_ext_transformer(
-        self, ext: str
-    ) -> Callable[[ParseResult], tuple[list[str], str]] | None:
+    def get_ext_transformer(self, ext: str) -> Callable[[ParseResult], tuple[list[str], str]] | None:
         return self._ext_map.get(ext)
-
-
 I.register("MpvConfig", MpvConfig())
 
-
 class MpvProgram(Program):
     _process: subprocess.Popen
 
@@ -54,14 +45,9 @@ class MpvProgram(Program):
 
     def __mpv_command(self, args: list[str]) -> dict[str, Any]:
         ipc_logger = logging.getLogger("mpv-ipc")
-        command = (
-            json.dumps(
-                {
+        command = json.dumps({
             "command": args,
-                }
-            )
-            + "\n"
-        )
+        }) + "\n"
         ipc_logger.debug("-> %s", command)
         with socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) as client:
             try:
@@ -75,6 +61,7 @@ class MpvProgram(Program):
             ipc_logger.debug("<- %s", resp_raw)
             return json.loads(resp_raw)
 
+
     def __read_prop(self, prop: str, default: Any) -> Any:
         return self.__mpv_command(["get_property", prop]).get("data", default)
 

microkodi/programs/vlc.py

@@ -15,7 +15,6 @@ from microkodi.helpers import KodiTime
 
 EVENT_PLAYER_EXIT = "post-player-exit"
 
-
 class VlcConfig:
     # Mapping of file extensions to a potential transformer function
     _ext_map: dict[str, Callable[[ParseResult], tuple[list[str], str]]]
@@ -30,30 +29,18 @@ class VlcConfig:
         self._event_listeners = {}
         self._domain_map = {}
 
-    def register_ext_transformer(
-        self, ext: str, transformer: Callable[[ParseResult], tuple[list[str], str]]
-    ):
-        logging.getLogger("VlcConfig").debug(
-            "Registering transformer for .%s URLs", ext
-        )
+    def register_ext_transformer(self, ext: str, transformer: Callable[[ParseResult], tuple[list[str], str]]):
+        logging.getLogger("VlcConfig").debug("Registering transformer for .%s URLs", ext)
         self._ext_map[ext] = transformer
 
-    def get_ext_transformer(
-        self, ext: str
-    ) -> Callable[[ParseResult], tuple[list[str], str]] | None:
+    def get_ext_transformer(self, ext: str) -> Callable[[ParseResult], tuple[list[str], str]] | None:
         return self._ext_map.get(ext)
 
-    def register_domain_transformer(
-        self, domain: str, transformer: Callable[[ParseResult], tuple[list[str], str]]
-    ):
-        logging.getLogger("VlcConfig").debug(
-            "Registering transformer for the %s domain", domain
-        )
+    def register_domain_transformer(self, domain: str, transformer: Callable[[ParseResult], tuple[list[str], str]]):
+        logging.getLogger("VlcConfig").debug("Registering transformer for the %s domain", domain)
         self._domain_map[domain] = transformer
 
-    def get_domain_transformer(
-        self, domain: str
-    ) -> Callable[[ParseResult], tuple[list[str], str]] | None:
+    def get_domain_transformer(self, domain: str) -> Callable[[ParseResult], tuple[list[str], str]] | None:
         return self._domain_map.get(domain)
 
     def register_event_listener(self, event: str, callback: Callable[[], None]):
@@ -67,11 +54,8 @@ class VlcConfig:
         for listener in self._event_listeners.get(event, []):
             logger.debug("Running listener %s for event %s", listener.__name__, event)
             listener()
-
-
 I.register("VlcConfig", VlcConfig())
 
-
 class VlcProgram(Program):
     _process: subprocess.Popen
 
@@ -155,12 +139,10 @@ class VlcProgram(Program):
             I.get("DataBridge").notification.emit("VLC exited with an error")
         self._process = None
 
-    def __vlc_command(
-        self, command: str, params: dict[str, Any] | None = None
-    ) -> str | None:
+    def __vlc_command(self, command: str, params: dict[str, Any] | None = None) -> str | None:
         try:
             req = requests.get(
-                "http://127.0.0.1:9090/requests/status.xml",
+                f"http://127.0.0.1:9090/requests/status.xml",
                 auth=("", self._vlc_password),
                 params={
                     "command": command,
@@ -184,7 +166,7 @@ class VlcProgram(Program):
             "seek",
             {
                 "val": timestamp.to_seconds(),
-            },
+            }
         )
 
     def exit(self):
@@ -210,9 +192,7 @@ class VlcProgram(Program):
             root = dom.getElementsByTagName("root")[0]
             time = int(root.getElementsByTagName("time")[0].childNodes[0].data)
             length = int(root.getElementsByTagName("length")[0].childNodes[0].data)
-            playing = (
-                root.getElementsByTagName("state")[0].childNodes[0].data == "playing"
-            )
+            playing = root.getElementsByTagName("state")[0].childNodes[0].data == "playing"
         except Exception as ex:
             self.logger.warning("Failed to parse vlc API response: %s", ex)
             self.logger.debug("Response body: %s", req.text)

microkodi/repository.py

@@ -1,6 +1,5 @@
 from typing import Any
 
-
 class _Repository:
     _data: dict[str, Any]
 
@@ -13,5 +12,4 @@ class _Repository:
     def get(self, key: str) -> Any | None:
         return self._data.get(key)
 
-
-I = _Repository() # noqa:E741
+I = _Repository()

microkodi/tv.py

@@ -1,45 +0,0 @@
-import logging
-import time
-
-import cec
-import requests
-
-from microkodi.config import Config
-from microkodi.repository import I
-
-
-class TVHandler:
-    logger: logging.Logger
-
-    def __init__(self):
-        self.logger = logging.getLogger("TVHandler")
-
-        config: Config = I.get("Config")
-        if config.cec:
-            cec.init()
-            self.logger.debug("CEC initialised")
-
-    def enable_power_if_configured(self) -> bool:
-        config: Config = I.get("Config")
-        if config.tv_power_webhook is not None:
-            try:
-                requests.put(config.tv_power_webhook)
-                self.logger.debug("Triggered webhook to enable power to the TV")
-            except Exception:
-                self.logger.warn("Failed to enable power to the TV")
-        return config.tv_power_webhook is not None
-
-    def power_on_if_needed(self) -> bool:
-        config: Config = I.get("Config")
-        if config.cec:
-            tc = cec.Device(cec.CECDEVICE_TV)
-            if not tc.is_on():
-                tc.power_on()
-        return config.cec
-
-    def turn_on_tv(self):
-        if self.enable_power_if_configured():
-            self.logger.debug("Waiting 500ms for TV to get power...")
-            time.sleep(500)
-
-        self.power_on_if_needed()

microkodi/ui/bridge.py

@@ -2,7 +2,6 @@ from datetime import datetime
 
 from PySide6.QtCore import QObject, Signal, Slot
 
-
 class DataBridge(QObject):
     """Bridge between the QML app and the Python "host"."""
 

pyproject.toml

@@ -4,22 +4,9 @@ version = "0.1.0"
 dependencies = [
     "pyside6",
     "requests",
+    "yt-dlp",
     "cec"
 ]
-requires-python = ">= 3.11"
 
-[project.optional-dependencies]
-youtube = [
-    "yt-dlp"
-]
-dev = [
-    "mypy",
-    "black",
-    "ruff"
-]
-
-[project.scripts]
-microkodi = "microkodi.main:main"
-
-[tool.setuptools.packages.find]
-exclude = ["contrib"]
+[tools.build]
+packages = ["microkodi"]

scripts/youtube.py

@@ -7,8 +7,10 @@ from microkodi.helpers import recursive_dict_merge
 
 import yt_dlp
 
-DEFAULT_CONFIG = {"format": "bestvideo[width<=1920]+bestaudio", "ytdlp_options": {}}
-
+DEFAULT_CONFIG = {
+    "format": "bestvideo[width<=1920]+bestaudio",
+    "ytdlp_options": {}
+}
 
 def youtube_url_transformer(url: ParseResult) -> tuple[list[str], str]:
     logger = logging.getLogger("Youtube")
@@ -22,26 +24,21 @@ def youtube_url_transformer(url: ParseResult) -> tuple[list[str], str]:
     with yt_dlp.YoutubeDL(opts) as ytdl:
         info = ytdl.extract_info(urlunparse(url), download=False)
 
-    #user_agent = None
+    user_agent = None
     audio_url = None
     video_url = None
     for format in info["requested_formats"]:
         if format["width"] is None:
             audio_url = format["url"]
         else:
-            #user_agent = format["http_headers"]["User-Agent"]
+            user_agent = format["http_headers"]["User-Agent"]
             video_url = format["url"]
 
-    args = (
-        [
-            f"--input-slave={audio_url}",
-        ]
-        if audio_url
-        else None
-    )
+    args = [
+        f'--input-slave={audio_url}',
+    ] if audio_url else None
     return args, urlparse(video_url)
 
-
 def init():
     # Create the config
     config: Config = I.get("Config")
@@ -56,9 +53,4 @@ def init():
 
     # Register the transformers
     I.get("VlcConfig").register_domain_transformer("youtu.be", youtube_url_transformer)
-    I.get("VlcConfig").register_domain_transformer(
-        "youtube.com", youtube_url_transformer
-    )
-    I.get("VlcConfig").register_domain_transformer(
-        "www.youtube.com", youtube_url_transformer
-    )
+    I.get("VlcConfig").register_domain_transformer("www.youtube.com", youtube_url_transformer)