web play api

config.ini.example

@@ -14,6 +14,10 @@ request_gone_timeout=60
 listen_host=192.168.1.2
 listen_port=8350
 
+[web_ui]
+enabled=0
+password=owo
+
 [discovery]
 upnp_enabled=1
 upnp_scan_timeout=3

smart_tv_telegram/config.py

@@ -26,6 +26,9 @@ class Config:
     _chromecast_enabled: bool
     _chromecast_scan_timeout: int = 0
 
+    _web_ui_enabled: bool
+    _web_ui_password: str = ""
+
     _xbmc_enabled: bool
     _xbmc_devices: typing.List[dict]
 
@@ -57,6 +60,11 @@ def __init__(self, path: str):
         if self._upnp_enabled:
             self._upnp_scan_timeout = int(config["discovery"]["upnp_scan_timeout"])
 
+        self._web_ui_enabled = bool(int(config["web_ui"]["enabled"]))
+
+        if self._web_ui_enabled:
+            self._web_ui_password = config["web_ui"]["password"]
+
         self._chromecast_enabled = bool(int(config["discovery"]["chromecast_enabled"]))
 
         self._device_request_timeout = int(config["discovery"]["device_request_timeout"])
@@ -101,6 +109,14 @@ def __init__(self, path: str):
         if not all(isinstance(x, int) for x in self._admins):
             raise ValueError("admins list should contain only integers")
 
+    @property
+    def web_ui_enabled(self) -> bool:
+        return self._web_ui_enabled
+
+    @property
+    def web_ui_password(self) -> str:
+        return self._web_ui_password
+
     @property
     def request_gone_timeout(self) -> int:
         return self._request_gone_timeout

smart_tv_telegram/devices/__init__.py

@@ -1,17 +1,19 @@
 import typing
 
-from .device import Device, DeviceFinder, RoutersDefType
+from .device import Device, DeviceFinder, RoutersDefType, RequestHandler
 from .upnp_device import UpnpDevice, UpnpDeviceFinder
 from .chromecast_device import ChromecastDevice, ChromecastDeviceFinder
 from .vlc_device import VlcDeviceFinder, VlcDevice
+from .web_device import WebDeviceFinder, WebDevice
 from .xbmc_device import XbmcDevice, XbmcDeviceFinder
 
 
 FINDERS: typing.List[typing.Type[DeviceFinder]] = [
     UpnpDeviceFinder,
     ChromecastDeviceFinder,
     XbmcDeviceFinder,
-    VlcDeviceFinder
+    VlcDeviceFinder,
+    WebDeviceFinder
 ]
 
 
@@ -27,5 +29,8 @@
     "VlcDevice",
     "VlcDeviceFinder",
     "FINDERS",
-    "RoutersDefType"
+    "RoutersDefType",
+    "RequestHandler",
+    "WebDeviceFinder",
+    "WebDevice"
 ]

smart_tv_telegram/devices/chromecast_device.py

@@ -6,6 +6,7 @@
 from .. import Config
 from ..tools import run_method_in_executor
 
+
 __all__ = [
     "ChromecastDevice",
     "ChromecastDeviceFinder"
@@ -15,7 +16,6 @@
 class ChromecastDevice(Device):
     _device: pychromecast.Chromecast
 
-    # noinspection PyMissingConstructor
     def __init__(self, device: typing.Any):
         self._device = device
         self._device.wait()
@@ -33,9 +33,8 @@ def play(self, url: str, title: str):
 
 
 class ChromecastDeviceFinder(DeviceFinder):
-    @staticmethod
     @run_method_in_executor
-    def find(config: Config) -> typing.List[Device]:
+    def find(self, config: Config) -> typing.List[Device]:
         return [
             ChromecastDevice(device)
             for device in pychromecast.get_chromecasts(

smart_tv_telegram/devices/device.py

@@ -7,22 +7,28 @@
 from .. import Config
 
 
-RoutersDefType = typing.List[typing.Tuple[str, typing.Callable[[Request], typing.Awaitable[Response]]]]
+class RequestHandler(abc.ABC):
+    @abc.abstractmethod
+    def get_path(self) -> str:
+        raise NotImplementedError
+
+    @abc.abstractmethod
+    async def handle(self, request: Request) -> Response:
+        raise NotImplementedError
+
+
+RoutersDefType = typing.List[RequestHandler]
 
 
 __all__ = [
     "Device",
     "DeviceFinder",
-    "RoutersDefType"
+    "RoutersDefType",
+    "RequestHandler",
 ]
 
 
 class Device(abc.ABC):
-    # noinspection PyUnusedLocal
-    @abc.abstractmethod
-    def __init__(self, device: typing.Any):
-        raise NotImplementedError
-
     @abc.abstractmethod
     async def stop(self):
         raise NotImplementedError
@@ -40,9 +46,8 @@ def __repr__(self):
 
 
 class DeviceFinder(abc.ABC):
-    @staticmethod
     @abc.abstractmethod
-    async def find(config: Config) -> typing.List[Device]:
+    async def find(self, config: Config) -> typing.List[Device]:
         raise NotImplementedError
 
     @staticmethod

smart_tv_telegram/devices/upnp_device.py

@@ -11,6 +11,7 @@
 from .. import Config
 from ..tools import ascii_only
 
+
 __all__ = [
     "UpnpDevice",
     "UpnpDeviceFinder"
@@ -40,7 +41,6 @@ class UpnpDevice(Device):
     _device: async_upnp_client.UpnpDevice
     _service: async_upnp_client.UpnpService
 
-    # noinspection PyMissingConstructor
     def __init__(self, device: async_upnp_client.UpnpDevice):
         self._device = device
         self._service = self._device.service(_AVTRANSPORT_SCHEMA)
@@ -67,8 +67,7 @@ async def play(self, url: str, title: str):
 
 
 class UpnpDeviceFinder(DeviceFinder):
-    @staticmethod
-    async def find(config: Config) -> typing.List[Device]:
+    async def find(self, config: Config) -> typing.List[Device]:
         devices = []
         requester = AiohttpRequester()
         factory = UpnpFactory(requester)

smart_tv_telegram/devices/vlc_device.py

@@ -6,6 +6,7 @@
 from . import DeviceFinder, Device, RoutersDefType
 from .. import Config
 
+
 __all__ = [
     "VlcDevice",
     "VlcDeviceFinder"
@@ -47,7 +48,6 @@ def password(self) -> typing.Optional[str]:
 class VlcDevice(Device):
     params: VlcDeviceParams
 
-    # noinspection PyMissingConstructor
     def __init__(self, device: VlcDeviceParams):
         self._params = device
 
@@ -90,8 +90,7 @@ async def play(self, url: str, title: str):
 
 
 class VlcDeviceFinder(DeviceFinder):
-    @staticmethod
-    async def find(config: Config) -> typing.List[Device]:
+    async def find(self, config: Config) -> typing.List[Device]:
         return [
             VlcDevice(VlcDeviceParams(params))
             for params in config.vlc_devices

smart_tv_telegram/devices/web_device.py

@@ -0,0 +1,130 @@
+import functools
+import typing
+
+from aiohttp.web_request import Request
+from aiohttp.web_response import Response
+
+from smart_tv_telegram import Config
+from smart_tv_telegram.devices import DeviceFinder, RoutersDefType, Device, RequestHandler
+from smart_tv_telegram.tools import secret_token, AsyncDebounce
+
+
+__all__ = [
+    "WebDeviceFinder",
+    "WebDevice"
+]
+
+
+class WebDevice(Device):
+    _url_to_play: typing.Optional[str] = None
+    _device_name: str
+    _token: int
+
+    def __init__(self, device_name: str, token: int):
+        self._device_name = device_name
+        self._token = token
+
+    async def stop(self):
+        self._url_to_play = None
+
+    async def play(self, url: str, title: str):
+        self._url_to_play = url
+
+    def get_token(self) -> int:
+        return self._token
+
+    def get_device_name(self) -> str:
+        return self._device_name
+
+    def get_url_to_play(self) -> typing.Optional[str]:
+        tmp = self._url_to_play
+        self._url_to_play = None
+        return tmp
+
+
+class WebDeviceApiRequestRegisterDevice(RequestHandler):
+    _config: Config
+    _devices: typing.Dict[WebDevice, AsyncDebounce]
+
+    def __init__(self, config: Config, devices: typing.Dict[WebDevice, AsyncDebounce]):
+        self._config = config
+        self._devices = devices
+
+    def get_path(self) -> str:
+        return "/web/api/register/{password}"
+
+    async def _remove_device(self, device: WebDevice):
+        try:
+            del self._devices[device]
+        except KeyError:
+            pass
+
+    async def handle(self, request: Request) -> Response:
+        password = request.match_info["password"]
+
+        if password != self._config.web_ui_password:
+            return Response(status=403)
+
+        token = secret_token()
+        device = WebDevice(f"web @({request.remote})", token)
+
+        remove = functools.partial(self._remove_device, device)
+        self._devices[device] = debounce = AsyncDebounce(remove, self._config.request_gone_timeout)
+        debounce.update_args()
+
+        return Response(status=200, body=str(token))
+
+
+class WebDeviceApiRequestPoll(RequestHandler):
+    _config: Config
+    _devices: typing.Dict[WebDevice, AsyncDebounce]
+
+    def __init__(self, config: Config, devices: typing.Dict[WebDevice, AsyncDebounce]):
+        self._devices = devices
+        self._config = config
+
+    def get_path(self) -> str:
+        return "/web/api/poll/{token}"
+
+    async def handle(self, request: Request) -> Response:
+        try:
+            token = int(request.match_info["token"])
+        except ValueError:
+            return Response(status=400)
+
+        try:
+            device = next(
+                d
+                for d in self._devices.keys()
+                if d.get_token() == token
+            )
+        except StopIteration:
+            return Response(status=404)
+
+        self._devices[device].update_args()
+        url_to_play = device.get_url_to_play()
+
+        if url_to_play is None:
+            return Response(status=302)
+
+        return Response(status=200, body=url_to_play)
+
+
+class WebDeviceFinder(DeviceFinder):
+    _devices: typing.Dict[WebDevice, AsyncDebounce]
+
+    def __init__(self):
+        self._devices = dict()
+
+    async def find(self, config: Config) -> typing.List[Device]:
+        return list(self._devices.keys())
+
+    @staticmethod
+    def is_enabled(config: Config) -> bool:
+        return config.web_ui_enabled
+
+    async def get_routers(self, config: Config) -> RoutersDefType:
+        return [
+            WebDeviceApiRequestRegisterDevice(config, self._devices),
+            WebDeviceApiRequestPoll(config, self._devices)
+        ]

smart_tv_telegram/devices/xbmc_device.py

@@ -11,6 +11,7 @@
 from . import Device, DeviceFinder, RoutersDefType
 from .. import Config
 
+
 __all__ = [
     "XbmcDevice",
     "XbmcDeviceFinder"
@@ -65,7 +66,6 @@ class XbmcDevice(Device):
     _http_url: str
     _host: str
 
-    # noinspection PyMissingConstructor
     def __init__(self, device: XbmcDeviceParams):
         if device.username:
             self._auth = aiohttp.BasicAuth(device.username, device.password)
@@ -138,8 +138,7 @@ async def play(self, url: str, title: str):
 
 
 class XbmcDeviceFinder(DeviceFinder):
-    @staticmethod
-    async def find(config: Config) -> typing.List[Device]:
+    async def find(self, config: Config) -> typing.List[Device]:
         return [
             XbmcDevice(XbmcDeviceParams(params))
             for params in config.xbmc_devices

smart_tv_telegram/http.py

@@ -47,8 +47,8 @@ async def start(self):
         for finder in self._finders:
             routers = await finder.get_routers(self._config)
 
-            for path, handler in routers:
-                app.add_routes([web.get(path, handler)])
+            for handler in routers:
+                app.add_routes([web.get(handler.get_path(), handler.handle)])
 
         # noinspection PyProtectedMember
         await web._run_app(app, host=self._config.listen_host, port=self._config.listen_port)