cleanup code, dont use a static for devices finders

requirements.txt

@@ -25,7 +25,7 @@ requests==2.25.1
 six==1.15.0
 TgCrypto==1.2.2
 typing-extensions==3.7.4.3
-urllib3==1.26.2
+urllib3==1.26.3
 voluptuous==0.12.1
 wrapt==1.12.1
 yarl==1.6.3

smart_tv_telegram/__init__.py

@@ -1,16 +1,16 @@
 from .config import Config
 from .mtproto import Mtproto
+from .devices_collection import DeviceFinderCollection
 from .http import Http, OnStreamClosed
 from .bot import Bot
 
-
 __version__ = "1.3.1"
 __version_info__ = ("1", "3", "1")
 __author__ = "https://github.com/andrew-ld"
 
-
 __all__ = [
     "Config",
+    "DeviceFinderCollection",
     "Mtproto",
     "Http",
     "OnStreamClosed",

smart_tv_telegram/__main__.py

@@ -5,8 +5,9 @@
 import os.path
 import typing
 
-from smart_tv_telegram import Http, Mtproto, Config, Bot
-from smart_tv_telegram.devices import FINDERS
+from smart_tv_telegram import Http, Mtproto, Config, Bot, DeviceFinderCollection
+from smart_tv_telegram.devices import UpnpDeviceFinder, ChromecastDeviceFinder, VlcDeviceFinder, \
+    WebDeviceFinder, XbmcDeviceFinder
 
 
 def open_config(parser: argparse.ArgumentParser, arg: str) -> typing.Optional[Config]:
@@ -28,24 +29,23 @@ def open_config(parser: argparse.ArgumentParser, arg: str) -> typing.Optional[Co
     return None
 
 
-async def async_main(config: Config):
-    finders = [f() for f in FINDERS if f.is_enabled(config)]
+async def async_main(config: Config, devices: DeviceFinderCollection):
     mtproto = Mtproto(config)
-    http = Http(mtproto, config, finders)
-    bot = Bot(mtproto, config, http, finders)
+    http = Http(mtproto, config, devices)
+    bot = Bot(mtproto, config, http, devices)
     http.set_on_stream_closed_handler(bot.get_on_stream_closed())
     bot.prepare()
 
     await mtproto.start()
     await http.start()
 
 
-def main(config: Config):
+def main(config: Config, devices: DeviceFinderCollection):
     loop = asyncio.get_event_loop()
-    loop.run_until_complete(async_main(config))
+    loop.run_until_complete(async_main(config, devices))
 
 
-def arg_parser():
+def arg_parser(devices: DeviceFinderCollection):
     parser = argparse.ArgumentParser()
     parser.add_argument("-c", "--config", type=lambda x: open_config(parser, x), default="config.ini")
     parser.add_argument("-v", "--verbosity", type=int, choices=[0, 1, 2], default=0)
@@ -61,8 +61,15 @@ def arg_parser():
     elif args.verbosity == 2:
         logging.basicConfig(level=logging.DEBUG)
 
-    main(args.config)
+    main(args.config, devices)
 
 
 if __name__ == "__main__":
-    arg_parser()
+    _devices = DeviceFinderCollection()
+    _devices.register_finder(UpnpDeviceFinder())
+    _devices.register_finder(ChromecastDeviceFinder())
+    _devices.register_finder(VlcDeviceFinder())
+    _devices.register_finder(WebDeviceFinder())
+    _devices.register_finder(XbmcDeviceFinder())
+
+    arg_parser(_devices)

smart_tv_telegram/bot.py

@@ -12,15 +12,14 @@
 from pyrogram.types import ReplyKeyboardRemove, Message, KeyboardButton, ReplyKeyboardMarkup, CallbackQuery, \
     InlineKeyboardMarkup, InlineKeyboardButton
 
-from . import Config, Mtproto, Http, OnStreamClosed
-from .devices import Device, DeviceFinder, DevicePlayerFunction
+from . import Config, Mtproto, Http, OnStreamClosed, DeviceFinderCollection
+from .devices import Device, DevicePlayerFunction
 from .tools import build_uri, pyrogram_filename, secret_token
 
 __all__ = [
     "Bot"
 ]
 
-
 _REMOVE_KEYBOARD = ReplyKeyboardRemove()
 _CANCEL_BUTTON = "^Cancel"
 
@@ -92,10 +91,10 @@ class Bot:
     _state_machine: TelegramStateMachine
     _mtproto: Mtproto
     _http: Http
-    _finders: typing.List[DeviceFinder]
+    _finders: DeviceFinderCollection
     _functions: typing.Dict[int, typing.Dict[int, DevicePlayerFunction]]
 
-    def __init__(self, mtproto: Mtproto, config: Config, http: Http, finders: typing.List[DeviceFinder]):
+    def __init__(self, mtproto: Mtproto, config: Config, http: Http, finders: DeviceFinderCollection):
         self._config = config
         self._mtproto = mtproto
         self._http = http
@@ -223,7 +222,7 @@ async def _select_device(self, _: Client, message: Message):
     async def _new_document(self, _: Client, message: Message):
         devices = []
 
-        for finder in self._finders:
+        for finder in self._finders.get_finders(self._config):
             with async_timeout.timeout(self._config.device_request_timeout + 1):
                 devices.extend(await finder.find(self._config))
 

smart_tv_telegram/config.py

@@ -2,7 +2,6 @@
 import configparser
 import typing
 
-
 __all__ = [
     "Config"
 ]

smart_tv_telegram/devices/__init__.py

@@ -1,22 +1,10 @@
-import typing
-
 from .device import Device, DeviceFinder, RoutersDefType, RequestHandler, DevicePlayerFunction
 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,
-    WebDeviceFinder
-]
-
-
 __all__ = [
     "Device",
     "DeviceFinder",
@@ -28,7 +16,6 @@
     "XbmcDeviceFinder",
     "VlcDevice",
     "VlcDeviceFinder",
-    "FINDERS",
     "RoutersDefType",
     "RequestHandler",
     "WebDeviceFinder",

smart_tv_telegram/devices/chromecast_device.py

@@ -7,7 +7,6 @@
 from .. import Config
 from ..tools import run_method_in_executor
 
-
 __all__ = [
     "ChromecastDevice",
     "ChromecastDeviceFinder"

smart_tv_telegram/devices/device.py

@@ -19,7 +19,6 @@ async def handle(self, request: Request) -> Response:
 
 RoutersDefType = typing.List[RequestHandler]
 
-
 __all__ = [
     "Device",
     "DeviceFinder",

smart_tv_telegram/devices/upnp_device.py

@@ -11,13 +11,11 @@
 from .. import Config, Mtproto
 from ..tools import ascii_only
 
-
 __all__ = [
     "UpnpDevice",
     "UpnpDeviceFinder"
 ]
 
-
 _AVTRANSPORT_SCHEMA = "urn:schemas-upnp-org:service:AVTransport:1"
 
 _DLL_METADATA = """

smart_tv_telegram/devices/vlc_device.py

@@ -6,13 +6,11 @@
 from . import DeviceFinder, Device, RoutersDefType, DevicePlayerFunction
 from .. import Config
 
-
 __all__ = [
     "VlcDevice",
     "VlcDeviceFinder"
 ]
 
-
 _LOGGER = logging.getLogger(__name__)
 _ENCODING = "utf8"
 _EOF = b"\n\r"

smart_tv_telegram/devices/web_device.py

@@ -8,7 +8,6 @@
 from smart_tv_telegram.devices import DeviceFinder, RoutersDefType, Device, RequestHandler, DevicePlayerFunction
 from smart_tv_telegram.tools import secret_token, AsyncDebounce
 
-
 __all__ = [
     "WebDeviceFinder",
     "WebDevice"

smart_tv_telegram/devices/xbmc_device.py

@@ -11,13 +11,11 @@
 from . import Device, DeviceFinder, RoutersDefType, DevicePlayerFunction
 from .. import Config
 
-
 __all__ = [
     "XbmcDevice",
     "XbmcDeviceFinder"
 ]
 
-
 _LOGGER = logging.getLogger(__name__)
 _ARGTYPE = typing.Union[typing.AnyStr, int, bool]
 

smart_tv_telegram/devices_collection.py

@@ -0,0 +1,17 @@
+import typing
+
+from smart_tv_telegram import Config
+from smart_tv_telegram.devices import DeviceFinder
+
+
+class DeviceFinderCollection:
+    _finders: typing.List[DeviceFinder]
+
+    def __init__(self):
+        self._finders = list()
+
+    def register_finder(self, finder: DeviceFinder):
+        self._finders.append(finder)
+
+    def get_finders(self, config: Config) -> typing.List[DeviceFinder]:
+        return [finder for finder in self._finders if finder.is_enabled(config)]

smart_tv_telegram/http.py

@@ -10,11 +10,9 @@
 from pyrogram.raw.types import MessageMediaDocument, Document
 from pyrogram.utils import get_peer_id
 
-from . import Config, Mtproto
-from .devices import DeviceFinder
+from . import Config, Mtproto, DeviceFinderCollection
 from .tools import parse_http_range, mtproto_filename, serialize_token, AsyncDebounce
 
-
 __all__ = [
     "Http",
     "OnStreamClosed"
@@ -30,15 +28,15 @@ async def handle(self, remains: float, chat_id: int, message_id: int, local_toke
 class Http:
     _mtproto: Mtproto
     _config: Config
-    _finders: typing.List[DeviceFinder]
+    _finders: DeviceFinderCollection
     _on_stream_closed: typing.Optional[OnStreamClosed] = None
 
     _tokens: typing.Set[int]
     _downloaded_blocks: typing.Dict[int, typing.Set[int]]
     _stream_debounce: typing.Dict[int, AsyncDebounce]
     _stream_transports: typing.Dict[int, typing.Set[asyncio.Transport]]
 
-    def __init__(self, mtproto: Mtproto, config: Config, finders: typing.List[DeviceFinder]):
+    def __init__(self, mtproto: Mtproto, config: Config, finders: DeviceFinderCollection):
         self._mtproto = mtproto
         self._config = config
         self._finders = finders
@@ -59,7 +57,7 @@ async def start(self):
         app.add_routes([web.put("/stream/{message_id}/{token}", self._upnp_discovery_handler)])
         app.add_routes([web.get("/healthcheck", self._health_check_handler)])
 
-        for finder in self._finders:
+        for finder in self._finders.get_finders(self._config):
             routers = await finder.get_routers(self._config)
 
             for handler in routers:

smart_tv_telegram/mtproto.py

@@ -19,7 +19,6 @@
 
 from . import Config
 
-
 __all__ = [
     "Mtproto"
 ]
@@ -61,11 +60,11 @@ async def get_message(self, message_id: int) -> Message:
 
     async def health_check(self):
         if not all(
-            session.is_connected.is_set()
-            for session in (
-                *self._client.media_sessions.values(),
-                self._client.session
-            )
+                session.is_connected.is_set()
+                for session in (
+                        *self._client.media_sessions.values(),
+                        self._client.session
+                )
         ):
             raise ConnectionError()
 

smart_tv_telegram/static/index.html

@@ -9,7 +9,7 @@
             const inputPassword = document.getElementById("password");
             const buttonPassword = document.getElementById("password_submit")
 
-            xhttp.onloadend = function() {
+            xhttp.onloadend = function () {
                 if (this.status === 200) {
                     inputPassword.hidden = true;
                     buttonPassword.hidden = true;
@@ -27,7 +27,7 @@
             const player = document.getElementById("player")
             const xhttp = new XMLHttpRequest();
 
-            xhttp.onloadend = function() {
+            xhttp.onloadend = function () {
                 if (this.status === 200) {
                     if (player.hidden) {
                         player.appendChild(source);
@@ -44,7 +44,9 @@
                     location.reload();
                 }
 
-                setTimeout(function () { pollVideo(token); }, 1000);
+                setTimeout(function () {
+                    pollVideo(token);
+                }, 1000);
             };
 
             xhttp.open("GET", "/web/api/poll/" + token, true);

smart_tv_telegram/tools.py

@@ -10,7 +10,6 @@
 
 from . import Config
 
-
 __all__ = [
     "mtproto_filename",
     "build_uri",
@@ -23,7 +22,6 @@
     "AsyncDebounce"
 ]
 
-
 _NAMED_MEDIA_TYPES = ("document", "video", "audio", "video_note", "animation")
 _RANGE_REGEX = re.compile(r"bytes=([0-9]+)-([0-9]+)?")
 _EXECUTOR = concurrent.futures.ThreadPoolExecutor()
@@ -91,8 +89,8 @@ def pyrogram_filename(message: BoxedMessage) -> str:
 
 def mtproto_filename(message: TlMessage) -> str:
     if not (
-        isinstance(message.media, MessageMediaDocument) and
-        isinstance(message.media.document, Document)
+            isinstance(message.media, MessageMediaDocument) and
+            isinstance(message.media.document, Document)
     ):
         raise TypeError()
 
@@ -117,6 +115,7 @@ def ascii_only(haystack: str) -> str:
 def run_method_in_executor(func):
     async def wraps(*args):
         return await _LOOP.run_in_executor(_EXECUTOR, func, *args)
+
     return wraps