Merge branch 'main' into python-lint-format-test

* Add lint job to build workflow

* Add prettier-plugin-import-sort

* Install prettier plugins before linting

* Use lint script from package.json

* Move linters to separate workflow

* Remove Python and Shell linters

* Remove popd

* Test that prettier properly fails the lint job

.github/workflows/lint.yml

@@ -0,0 +1,27 @@
+name: Lint
+
+on:
+  push:
+
+jobs:
+  lint:
+    name: Run linters
+    runs-on: ubuntu-20.04
+
+    steps:
+      - uses: actions/checkout@v2 # Check out the repository first.
+      - name: Run prettier (JavaScript & TypeScript)
+        run: |
+          pushd frontend
+          npm install
+          npm run lint
+
+      - name: Run black (Python formatting)
+        uses: lgeiger/black-action@v1.0.1
+        with:
+          args: "./backend --experimental-string-processing --config ./backend/pyproject.toml"
+
+      - name: Run ruff (Python linting)
+        uses: jpetrucciani/ruff-check@main
+        with:
+          path: "./backend"

backend/browser.py

@@ -1,27 +1,33 @@
 # Full imports
 import json
+
 # import pprint
 # from pprint import pformat
 
 # Partial imports
-from aiohttp import ClientSession, web
-from asyncio import get_event_loop, sleep
-from concurrent.futures import ProcessPoolExecutor
+from aiohttp import ClientSession
+from asyncio import sleep
 from hashlib import sha256
 from io import BytesIO
 from logging import getLogger
-from os import R_OK, W_OK, path, rename, listdir, access, mkdir
+from os import R_OK, W_OK, path, listdir, access, mkdir
 from shutil import rmtree
 from subprocess import call
 from time import time
 from zipfile import ZipFile
 
 # Local modules
-from helpers import get_ssl_context, get_user, get_user_group, download_remote_binary_to_path
+from helpers import (
+    get_ssl_context,
+    get_user,
+    get_user_group,
+    download_remote_binary_to_path,
+)
 from injector import get_gamepadui_tab
 
 logger = getLogger("Browser")
 
+
 class PluginInstallContext:
     def __init__(self, artifact, name, version, hash) -> None:
         self.artifact = artifact
@@ -29,6 +35,7 @@ class PluginInstallContext:
         self.version = version
         self.hash = hash
 
+
 class PluginBrowser:
     def __init__(self, plugin_path, plugins, loader) -> None:
         self.plugin_path = plugin_path
@@ -43,32 +50,40 @@ class PluginBrowser:
         zip_file = ZipFile(zip)
         zip_file.extractall(self.plugin_path)
         plugin_dir = self.find_plugin_folder(name)
-        code_chown = call(["chown", "-R", get_user()+":"+get_user_group(), plugin_dir])
+        code_chown = call(
+            ["chown", "-R", get_user() + ":" + get_user_group(), plugin_dir]
+        )
         code_chmod = call(["chmod", "-R", "555", plugin_dir])
         if code_chown != 0 or code_chmod != 0:
-            logger.error(f"chown/chmod exited with a non-zero exit code (chown: {code_chown}, chmod: {code_chmod})")
+            logger.error(
+                f"chown/chmod exited with a non-zero exit code (chown: {code_chown},"
+                f" chmod: {code_chmod})"
+            )
             return False
         return True
-    
+
     async def _download_remote_binaries_for_plugin_with_name(self, pluginBasePath):
         rv = False
         try:
-            packageJsonPath = path.join(pluginBasePath, 'package.json')
-            pluginBinPath = path.join(pluginBasePath, 'bin')
+            packageJsonPath = path.join(pluginBasePath, "package.json")
+            pluginBinPath = path.join(pluginBasePath, "bin")
 
             if access(packageJsonPath, R_OK):
                 with open(packageJsonPath, "r", encoding="utf-8") as f:
                     packageJson = json.load(f)
-                    if "remote_binary" in packageJson and len(packageJson["remote_binary"]) > 0:
+                    if (
+                        "remote_binary" in packageJson
+                        and len(packageJson["remote_binary"]) > 0
+                    ):
                         # create bin directory if needed.
-                        rc=call(["chmod", "-R", "777", pluginBasePath])
+                        call(["chmod", "-R", "777", pluginBasePath])
                         if access(pluginBasePath, W_OK):
-                            
+
                             if not path.exists(pluginBinPath):
                                 mkdir(pluginBinPath)
-                            
+
                             if not access(pluginBinPath, W_OK):
-                                rc=call(["chmod", "-R", "777", pluginBinPath])
+                                call(["chmod", "-R", "777", pluginBinPath])
 
                         rv = True
                         for remoteBinary in packageJson["remote_binary"]:
@@ -76,16 +91,29 @@ class PluginBrowser:
                             binName = remoteBinary["name"]
                             binURL = remoteBinary["url"]
                             binHash = remoteBinary["sha256hash"]
-                            if not await download_remote_binary_to_path(binURL, binHash, path.join(pluginBinPath, binName)):
+                            if not await download_remote_binary_to_path(
+                                binURL, binHash, path.join(pluginBinPath, binName)
+                            ):
                                 rv = False
-                                raise Exception(f"Error Downloading Remote Binary {binName}@{binURL} with hash {binHash} to {path.join(pluginBinPath, binName)}")
+                                raise Exception(
+                                    "Error Downloading Remote Binary"
+                                    f" {binName}@{binURL} with hash {binHash} to"
+                                    f" {path.join(pluginBinPath, binName)}"
+                                )
 
-                        code_chown = call(["chown", "-R", get_user()+":"+get_user_group(), self.plugin_path])
-                        rc=call(["chmod", "-R", "555", pluginBasePath])
+                        call(
+                            [
+                                "chown",
+                                "-R",
+                                get_user() + ":" + get_user_group(),
+                                self.plugin_path,
+                            ]
+                        )
+                        call(["chmod", "-R", "555", pluginBasePath])
                     else:
                         rv = True
-                        logger.debug(f"No Remote Binaries to Download")
-                
+                        logger.debug("No Remote Binaries to Download")
+
         except Exception as e:
             rv = False
             logger.debug(str(e))
@@ -95,12 +123,16 @@ class PluginBrowser:
     def find_plugin_folder(self, name):
         for folder in listdir(self.plugin_path):
             try:
-                with open(path.join(self.plugin_path, folder, 'plugin.json'), "r", encoding="utf-8") as f:
+                with open(
+                    path.join(self.plugin_path, folder, "plugin.json"),
+                    "r",
+                    encoding="utf-8",
+                ) as f:
                     plugin = json.load(f)
 
-                if plugin['name'] == name:
+                if plugin["name"] == name:
                     return str(path.join(self.plugin_path, folder))
-            except:
+            except Exception:
                 logger.debug(f"skipping {folder}")
 
     async def uninstall_plugin(self, name):
@@ -127,8 +159,10 @@ class PluginBrowser:
         except FileNotFoundError:
             logger.warning(f"Plugin {name} not installed, skipping uninstallation")
         except Exception as e:
-            logger.error(f"Plugin {name} in {self.find_plugin_folder(name)} was not uninstalled")
-            logger.error(f"Error at %s", exc_info=e)
+            logger.error(
+                f"Plugin {name} in {self.find_plugin_folder(name)} was not uninstalled"
+            )
+            logger.error("Error at %s", exc_info=e)
         if self.loader.watcher:
             self.loader.watcher.disabled = False
 
@@ -140,8 +174,11 @@ class PluginBrowser:
             pluginFolderPath = self.find_plugin_folder(name)
             if pluginFolderPath:
                 isInstalled = True
-        except:
-            logger.error(f"Failed to determine if {name} is already installed, continuing anyway.")
+        except Exception:
+            logger.error(
+                f"Failed to determine if {name} is already installed, continuing"
+                " anyway."
+            )
         logger.info(f"Installing {name} (Version: {version})")
         async with ClientSession() as client:
             logger.debug(f"Fetching {artifact}")
@@ -155,22 +192,26 @@ class PluginBrowser:
                     try:
                         logger.debug("Uninstalling existing plugin...")
                         await self.uninstall_plugin(name)
-                    except:
+                    except Exception:
                         logger.error(f"Plugin {name} could not be uninstalled.")
                 logger.debug("Unzipping...")
                 ret = self._unzip_to_plugin_dir(res_zip, name, hash)
                 if ret:
                     plugin_dir = self.find_plugin_folder(name)
-                    ret = await self._download_remote_binaries_for_plugin_with_name(plugin_dir)
+                    ret = await self._download_remote_binaries_for_plugin_with_name(
+                        plugin_dir
+                    )
                     if ret:
                         logger.info(f"Installed {name} (Version: {version})")
                         if name in self.loader.plugins:
                             self.loader.plugins[name].stop()
                             self.loader.plugins.pop(name, None)
                         await sleep(1)
-                        self.loader.import_plugin(path.join(plugin_dir, "main.py"), plugin_dir)
+                        self.loader.import_plugin(
+                            path.join(plugin_dir, "main.py"), plugin_dir
+                        )
                     else:
-                        logger.fatal(f"Failed Downloading Remote Binaries")
+                        logger.fatal("Failed Downloading Remote Binaries")
                 else:
                     self.log.fatal(f"SHA-256 Mismatch!!!! {name} (Version: {version})")
                 if self.loader.watcher:
@@ -180,14 +221,21 @@ class PluginBrowser:
 
     async def request_plugin_install(self, artifact, name, version, hash):
         request_id = str(time())
-        self.install_requests[request_id] = PluginInstallContext(artifact, name, version, hash)
+        self.install_requests[request_id] = PluginInstallContext(
+            artifact, name, version, hash
+        )
         tab = await get_gamepadui_tab()
         await tab.open_websocket()
-        await tab.evaluate_js(f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}', '{request_id}', '{hash}')")
+        await tab.evaluate_js(
+            f"DeckyPluginLoader.addPluginInstallPrompt('{name}', '{version}',"
+            f" '{request_id}', '{hash}')"
+        )
 
     async def confirm_plugin_install(self, request_id):
         request = self.install_requests.pop(request_id)
-        await self._install(request.artifact, request.name, request.version, request.hash)
+        await self._install(
+            request.artifact, request.name, request.version, request.hash
+        )
 
     def cancel_plugin_install(self, request_id):
         self.install_requests.pop(request_id)

backend/helpers.py

@@ -6,8 +6,6 @@ import subprocess
 import uuid
 import os
 import sys
-from subprocess import check_output
-from time import sleep
 from hashlib import sha256
 from io import BytesIO
 
@@ -24,22 +22,38 @@ ssl_ctx = ssl.create_default_context(cafile=certifi.where())
 assets_regex = re.compile("^/plugins/.*/assets/.*")
 frontend_regex = re.compile("^/frontend/.*")
 
+
 def get_ssl_context():
     return ssl_ctx
 
+
 def get_csrf_token():
     return csrf_token
 
+
 @middleware
 async def csrf_middleware(request, handler):
-    if str(request.method) == "OPTIONS" or request.headers.get('Authentication') == csrf_token or str(request.rel_url) == "/auth/token" or str(request.rel_url).startswith("/plugins/load_main/") or str(request.rel_url).startswith("/static/") or str(request.rel_url).startswith("/legacy/") or str(request.rel_url).startswith("/steam_resource/") or str(request.rel_url).startswith("/frontend/") or assets_regex.match(str(request.rel_url)) or frontend_regex.match(str(request.rel_url)):
+    if (
+        str(request.method) == "OPTIONS"
+        or request.headers.get("Authentication") == csrf_token
+        or str(request.rel_url) == "/auth/token"
+        or str(request.rel_url).startswith("/plugins/load_main/")
+        or str(request.rel_url).startswith("/static/")
+        or str(request.rel_url).startswith("/legacy/")
+        or str(request.rel_url).startswith("/steam_resource/")
+        or str(request.rel_url).startswith("/frontend/")
+        or assets_regex.match(str(request.rel_url))
+        or frontend_regex.match(str(request.rel_url))
+    ):
         return await handler(request)
-    return Response(text='Forbidden', status='403')
+    return Response(text="Forbidden", status="403")
+
 
 # Deprecated
 def set_user():
     pass
 
+
 # Get the user id hosting the plugin loader
 def get_user_id() -> int:
     proc_path = os.path.realpath(sys.argv[0])
@@ -47,60 +61,73 @@ def get_user_id() -> int:
     for pw in pws:
         if proc_path.startswith(os.path.realpath(pw.pw_dir)):
             return pw.pw_uid
-    raise PermissionError("The plugin loader does not seem to be hosted by any known user.")
+    raise PermissionError(
+        "The plugin loader does not seem to be hosted by any known user."
+    )
+
 
 # Get the user hosting the plugin loader
 def get_user() -> str:
     return pwd.getpwuid(get_user_id()).pw_name
 
+
 # Get the effective user id of the running process
 def get_effective_user_id() -> int:
     return os.geteuid()
 
+
 # Get the effective user of the running process
 def get_effective_user() -> str:
     return pwd.getpwuid(get_effective_user_id()).pw_name
 
+
 # Get the effective user group id of the running process
 def get_effective_user_group_id() -> int:
     return os.getegid()
 
+
 # Get the effective user group of the running process
 def get_effective_user_group() -> str:
     return grp.getgrgid(get_effective_user_group_id()).gr_name
 
+
 # Get the user owner of the given file path.
 def get_user_owner(file_path) -> str:
     return pwd.getpwuid(os.stat(file_path).st_uid).pw_name
 
-# Get the user group of the given file path.
-def get_user_group(file_path) -> str:
-    return grp.getgrgid(os.stat(file_path).st_gid).gr_name
 
 # Deprecated
 def set_user_group() -> str:
     return get_user_group()
 
+
 # Get the group id of the user hosting the plugin loader
 def get_user_group_id() -> int:
     return pwd.getpwuid(get_user_id()).pw_gid
 
+
 # Get the group of the user hosting the plugin loader
-def get_user_group() -> str:
-    return grp.getgrgid(get_user_group_id()).gr_name
+def get_user_group(file_path) -> str:
+    if file_path:
+        return grp.getgrgid(os.stat(file_path).st_gid).gr_name
+    else:
+        return grp.getgrgid(get_user_group_id()).gr_name
+
 
 # Get the default home path unless a user is specified
-def get_home_path(username = None) -> str:
-    if username == None:
+def get_home_path(username=None) -> str:
+    if username is None:
         username = get_user()
     return pwd.getpwnam(username).pw_dir
 
+
 # Get the default homebrew path unless a home_path is specified
-def get_homebrew_path(home_path = None) -> str:
-    if home_path == None:
+def get_homebrew_path(home_path=None) -> str:
+    if home_path is None:
         home_path = get_home_path()
     return os.path.join(home_path, "homebrew")
 
+
 # Recursively create path and chown as user
 def mkdir_as_user(path):
     path = os.path.realpath(path)
@@ -113,11 +140,17 @@ def mkdir_as_user(path):
         chown_path = os.path.join(chown_path, p)
         os.chown(chown_path, uid, gid)
 
+
 # Fetches the version of loader
 def get_loader_version() -> str:
-    with open(os.path.join(os.path.dirname(sys.argv[0]), ".loader.version"), "r", encoding="utf-8") as version_file:
+    with open(
+        os.path.join(os.path.dirname(sys.argv[0]), ".loader.version"),
+        "r",
+        encoding="utf-8",
+    ) as version_file:
         return version_file.readline().replace("\n", "")
 
+
 # Download Remote Binaries to local Plugin
 async def download_remote_binary_to_path(url, binHash, path) -> bool:
     rv = False
@@ -130,27 +163,36 @@ async def download_remote_binary_to_path(url, binHash, path) -> bool:
                 remoteHash = sha256(data.getbuffer()).hexdigest()
                 if binHash == remoteHash:
                     data.seek(0)
-                    with open(path, 'wb') as f:
+                    with open(path, "wb") as f:
                         f.write(data.getbuffer())
                         rv = True
                 else:
-                    raise Exception(f"Fatal Error: Hash Mismatch for remote binary {path}@{url}")
+                    raise Exception(
+                        f"Fatal Error: Hash Mismatch for remote binary {path}@{url}"
+                    )
             else:
                 rv = False
-    except:
+    except Exception:
         rv = False
 
     return rv
 
+
 async def is_systemd_unit_active(unit_name: str) -> bool:
-    res = subprocess.run(["systemctl", "is-active", unit_name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+    res = subprocess.run(
+        ["systemctl", "is-active", unit_name],
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
     return res.returncode == 0
 
+
 async def stop_systemd_unit(unit_name: str) -> subprocess.CompletedProcess:
     cmd = ["systemctl", "stop", unit_name]
 
     return subprocess.run(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT)
 
+
 async def start_systemd_unit(unit_name: str) -> subprocess.CompletedProcess:
     cmd = ["systemctl", "start", unit_name]
 

backend/injector.py

@@ -2,10 +2,9 @@
 
 from asyncio import sleep
 from logging import getLogger
-from traceback import format_exc
 from typing import List
 
-from aiohttp import ClientSession, WSMsgType
+from aiohttp import ClientSession
 from aiohttp.client_exceptions import ClientConnectorError, ClientOSError
 from asyncio.exceptions import TimeoutError
 import uuid
@@ -39,9 +38,12 @@ class Tab:
         async for message in self.websocket:
             data = message.json()
             yield data
-        logger.warn(f"The Tab {self.title} socket has been disconnected while listening for messages.")
+        logger.warn(
+            f"The Tab {self.title} socket has been disconnected while listening for"
+            " messages."
+        )
         await self.close_websocket()
-            
+
     async def _send_devtools_cmd(self, dc, receive=True):
         if self.websocket:
             self.cmd_id += 1
@@ -54,19 +56,24 @@ class Tab:
             return None
         raise RuntimeError("Websocket not opened")
 
-    async def evaluate_js(self, js, run_async=False, manage_socket=True, get_result=True):
+    async def evaluate_js(
+        self, js, run_async=False, manage_socket=True, get_result=True
+    ):
         try:
             if manage_socket:
                 await self.open_websocket()
 
-            res = await self._send_devtools_cmd({
-                "method": "Runtime.evaluate",
-                "params": {
-                    "expression": js,
-                    "userGesture": True,
-                    "awaitPromise": run_async
-                }
-            }, get_result)
+            res = await self._send_devtools_cmd(
+                {
+                    "method": "Runtime.evaluate",
+                    "params": {
+                        "expression": js,
+                        "userGesture": True,
+                        "awaitPromise": run_async,
+                    },
+                },
+                get_result,
+            )
 
         finally:
             if manage_socket:
@@ -74,9 +81,17 @@ class Tab:
         return res
 
     async def has_global_var(self, var_name, manage_socket=True):
-        res = await self.evaluate_js(f"window['{var_name}'] !== null && window['{var_name}'] !== undefined", False, manage_socket)
+        res = await self.evaluate_js(
+            f"window['{var_name}'] !== null && window['{var_name}'] !== undefined",
+            False,
+            manage_socket,
+        )
 
-        if not "result" in res or not "result" in res["result"] or not "value" in res["result"]["result"]:
+        if (
+            "result" not in res
+            or "result" not in res["result"]
+            or "value" not in res["result"]["result"]
+        ):
             return False
 
         return res["result"]["result"]["value"]
@@ -86,9 +101,12 @@ class Tab:
             if manage_socket:
                 await self.open_websocket()
 
-            res = await self._send_devtools_cmd({
-                "method": "Page.close",
-            }, False)
+            res = await self._send_devtools_cmd(
+                {
+                    "method": "Page.close",
+                },
+                False,
+            )
 
         finally:
             if manage_socket:
@@ -99,32 +117,42 @@ class Tab:
         """
         Enables page domain notifications.
         """
-        await self._send_devtools_cmd({
-            "method": "Page.enable",
-        }, False)
+        await self._send_devtools_cmd(
+            {
+                "method": "Page.enable",
+            },
+            False,
+        )
 
     async def disable(self):
         """
         Disables page domain notifications.
         """
-        await self._send_devtools_cmd({
-            "method": "Page.disable",
-        }, False)
+        await self._send_devtools_cmd(
+            {
+                "method": "Page.disable",
+            },
+            False,
+        )
 
-    async def refresh(self):
+    async def refresh(self, manage_socket=False):
         try:
             if manage_socket:
                 await self.open_websocket()
 
-            await self._send_devtools_cmd({
-                "method": "Page.reload",
-            }, False)
+            await self._send_devtools_cmd(
+                {
+                    "method": "Page.reload",
+                },
+                False,
+            )
 
         finally:
             if manage_socket:
                 await self.close_websocket()
 
         return
+
     async def reload_and_evaluate(self, js, manage_socket=True):
         """
         Reloads the current tab, with JS to run on load via debugger
@@ -133,64 +161,70 @@ class Tab:
             if manage_socket:
                 await self.open_websocket()
 
-            await self._send_devtools_cmd({
-                "method": "Debugger.enable"
-            }, True)
+            await self._send_devtools_cmd({"method": "Debugger.enable"}, True)
 
-            await self._send_devtools_cmd({
-                "method": "Runtime.evaluate",
-                "params": {
-                    "expression": "location.reload();",
-                    "userGesture": True,
-                    "awaitPromise": False
-                }
-            }, False)
+            await self._send_devtools_cmd(
+                {
+                    "method": "Runtime.evaluate",
+                    "params": {
+                        "expression": "location.reload();",
+                        "userGesture": True,
+                        "awaitPromise": False,
+                    },
+                },
+                False,
+            )
 
-            breakpoint_res = await self._send_devtools_cmd({
-                "method": "Debugger.setInstrumentationBreakpoint",
-                "params": {
-                    "instrumentation": "beforeScriptExecution"
-                }
-            }, True)
+            breakpoint_res = await self._send_devtools_cmd(
+                {
+                    "method": "Debugger.setInstrumentationBreakpoint",
+                    "params": {"instrumentation": "beforeScriptExecution"},
+                },
+                True,
+            )
 
             logger.info(breakpoint_res)
-            
+
             # Page finishes loading when breakpoint hits
 
             for x in range(20):
                 # this works around 1/5 of the time, so just send it 8 times.
                 # the js accounts for being injected multiple times allowing only one instance to run at a time anyway
-                await self._send_devtools_cmd({
-                    "method": "Runtime.evaluate",
-                    "params": {
-                        "expression": js,
-                        "userGesture": True,
-                        "awaitPromise": False
-                    }
-                }, False)
+                await self._send_devtools_cmd(
+                    {
+                        "method": "Runtime.evaluate",
+                        "params": {
+                            "expression": js,
+                            "userGesture": True,
+                            "awaitPromise": False,
+                        },
+                    },
+                    False,
+                )
 
-            await self._send_devtools_cmd({
-                "method": "Debugger.removeBreakpoint",
-                "params": {
-                    "breakpointId": breakpoint_res["result"]["breakpointId"]
-                }
-            }, False)
+            await self._send_devtools_cmd(
+                {
+                    "method": "Debugger.removeBreakpoint",
+                    "params": {
+                        "breakpointId": breakpoint_res["result"]["breakpointId"]
+                    },
+                },
+                False,
+            )
 
             for x in range(4):
-                await self._send_devtools_cmd({
-                    "method": "Debugger.resume"
-                }, False)
+                await self._send_devtools_cmd({"method": "Debugger.resume"}, False)
 
-            await self._send_devtools_cmd({
-                "method": "Debugger.disable"
-            }, True)
+            await self._send_devtools_cmd({"method": "Debugger.disable"}, True)
 
         finally:
             if manage_socket:
                 await self.close_websocket()
         return
 
-    async def add_script_to_evaluate_on_new_document(self, js, add_dom_wrapper=True, manage_socket=True, get_result=True):
+    async def add_script_to_evaluate_on_new_document(
+        self, js, add_dom_wrapper=True, manage_socket=True, get_result=True
+    ):
         """
         How the underlying call functions is not particularly clear from the devtools docs, so stealing puppeteer's description:
 
@@ -225,35 +259,44 @@ class Tab:
         """
         try:
 
-            wrappedjs = """
-            function scriptFunc() {
+            wrappedjs = (
+                """
+            function scriptFunc() {{
                 {js}
-            }
-            if (document.readyState === 'loading') {
-                addEventListener('DOMContentLoaded', () => {
+            }}
+            if (document.readyState === 'loading') {{
+                addEventListener('DOMContentLoaded', () => {{
                 scriptFunc();
-            });
-            } else {
+            }});
+            }} else {{
                 scriptFunc();
-            }
-            """.format(js=js) if add_dom_wrapper else js
+            }}
+            """.format(
+                    js=js
+                )
+                if add_dom_wrapper
+                else js
+            )
 
             if manage_socket:
                 await self.open_websocket()
 
-            res = await self._send_devtools_cmd({
-                "method": "Page.addScriptToEvaluateOnNewDocument",
-                "params": {
-                    "source": wrappedjs
-                }
-            }, get_result)
+            res = await self._send_devtools_cmd(
+                {
+                    "method": "Page.addScriptToEvaluateOnNewDocument",
+                    "params": {"source": wrappedjs},
+                },
+                get_result,
+            )
 
         finally:
             if manage_socket:
                 await self.close_websocket()
         return res
 
-    async def remove_script_to_evaluate_on_new_document(self, script_id, manage_socket=True):
+    async def remove_script_to_evaluate_on_new_document(
+        self, script_id, manage_socket=True
+    ):
         """
         Removes a script from a page that was added with `add_script_to_evaluate_on_new_document`
 
@@ -267,21 +310,28 @@ class Tab:
             if manage_socket:
                 await self.open_websocket()
 
-            res = await self._send_devtools_cmd({
-                "method": "Page.removeScriptToEvaluateOnNewDocument",
-                "params": {
-                    "identifier": script_id
-                }
-            }, False)
+            await self._send_devtools_cmd(
+                {
+                    "method": "Page.removeScriptToEvaluateOnNewDocument",
+                    "params": {"identifier": script_id},
+                },
+                False,
+            )
 
         finally:
             if manage_socket:
                 await self.close_websocket()
 
     async def has_element(self, element_name, manage_socket=True):
-        res = await self.evaluate_js(f"document.getElementById('{element_name}') != null", False, manage_socket)
+        res = await self.evaluate_js(
+            f"document.getElementById('{element_name}') != null", False, manage_socket
+        )
 
-        if not "result" in res or not "result" in res["result"] or not "value" in res["result"]["result"]:
+        if (
+            "result" not in res
+            or "result" not in res["result"]
+            or "value" not in res["result"]["result"]
+        ):
             return False
 
         return res["result"]["result"]["value"]
@@ -298,23 +348,17 @@ class Tab:
                     document.head.append(style);
                     style.textContent = `{style}`;
                 }})()
-                """, False, manage_socket)
+                """,
+                False,
+                manage_socket,
+            )
 
             if "exceptionDetails" in result["result"]:
-                return {
-                    "success": False,
-                    "result": result["result"]
-                }
+                return {"success": False, "result": result["result"]}
 
-            return {
-                "success": True,
-                "result": css_id
-            }
+            return {"success": True, "result": css_id}
         except Exception as e:
-            return {
-                "success": False,
-                "result": e
-            }
+            return {"success": False, "result": e}
 
     async def remove_css(self, css_id, manage_socket=True):
         try:
@@ -326,25 +370,24 @@ class Tab:
                     if (style.nodeName.toLowerCase() == 'style')
                         style.parentNode.removeChild(style);
                 }})()
-                """, False, manage_socket)
+                """,
+                False,
+                manage_socket,
+            )
 
             if "exceptionDetails" in result["result"]:
-                return {
-                    "success": False,
-                    "result": result
-                }
+                return {"success": False, "result": result}
 
-            return {
-                "success": True
-            }
+            return {"success": True}
         except Exception as e:
-            return {
-                "success": False,
-                "result": e
-            }
+            return {"success": False, "result": e}
 
     async def get_steam_resource(self, url):
-        res = await self.evaluate_js(f'(async function test() {{ return await (await fetch("{url}")).text() }})()', True)
+        res = await self.evaluate_js(
+            f'(async function test() {{ return await (await fetch("{url}")).text()'
+            " })()",
+            True,
+        )
         return res["result"]["result"]["value"]
 
     def __repr__(self):
@@ -387,32 +430,45 @@ async def get_tab(tab_name) -> Tab:
         raise ValueError(f"Tab {tab_name} not found")
     return tab
 
+
 async def get_tab_lambda(test) -> Tab:
     tabs = await get_tabs()
     tab = next((i for i in tabs if test(i)), None)
     if not tab:
-        raise ValueError(f"Tab not found by lambda")
+        raise ValueError("Tab not found by lambda")
     return tab
 
+
 def tab_is_gamepadui(t: Tab) -> bool:
-    return "https://steamloopback.host/routes/" in t.url and (t.title == "Steam Shared Context presented by Valve™" or t.title == "Steam" or t.title == "SP")
+    return "https://steamloopback.host/routes/" in t.url and (
+        t.title == "Steam Shared Context presented by Valve™"
+        or t.title == "Steam"
+        or t.title == "SP"
+    )
+
 
 async def get_gamepadui_tab() -> Tab:
     tabs = await get_tabs()
     tab = next((i for i in tabs if tab_is_gamepadui(i)), None)
     if not tab:
-        raise ValueError(f"GamepadUI Tab not found")
+        raise ValueError("GamepadUI Tab not found")
     return tab
 
+
 async def inject_to_tab(tab_name, js, run_async=False):
     tab = await get_tab(tab_name)
 
     return await tab.evaluate_js(js, run_async)
 
+
 async def close_old_tabs():
     tabs = await get_tabs()
     for t in tabs:
-        if not t.title or (t.title != "Steam Shared Context presented by Valve™" and t.title != "Steam" and t.title != "SP"):
+        if not t.title or (
+            t.title != "Steam Shared Context presented by Valve™"
+            and t.title != "Steam"
+            and t.title != "SP"
+        ):
             logger.debug("Closing tab: " + getattr(t, "title", "Untitled"))
             await t.close()
-            await sleep(0.5)
+            await sleep(0.5)

backend/loader.py

@@ -21,7 +21,7 @@ from plugin import PluginWrapper
 
 class FileChangeHandler(RegexMatchingEventHandler):
     def __init__(self, queue, plugin_path) -> None:
-        super().__init__(regexes=[r'^.*?dist\/index\.js$', r'^.*?main\.py$'])
+        super().__init__(regexes=[r"^.*?dist\/index\.js$", r"^.*?main\.py$"])
         self.logger = getLogger("file-watcher")
         self.plugin_path = plugin_path
         self.queue = queue
@@ -32,7 +32,9 @@ class FileChangeHandler(RegexMatchingEventHandler):
             return
         plugin_dir = Path(path.relpath(src_path, self.plugin_path)).parts[0]
         if exists(path.join(self.plugin_path, plugin_dir, "plugin.json")):
-            self.queue.put_nowait((path.join(self.plugin_path, plugin_dir, "main.py"), plugin_dir, True))
+            self.queue.put_nowait(
+                (path.join(self.plugin_path, plugin_dir, "main.py"), plugin_dir, True)
+            )
 
     def on_created(self, event):
         src_path = event.src_path
@@ -62,6 +64,7 @@ class FileChangeHandler(RegexMatchingEventHandler):
         self.logger.debug(f"file modified: {src_path}")
         self.maybe_reload(src_path)
 
+
 class Loader:
     def __init__(self, server_instance, plugin_path, loop, live_reload=False) -> None:
         self.loop = loop
@@ -81,18 +84,30 @@ class Loader:
             self.loop.create_task(self.handle_reloads())
             self.loop.create_task(self.enable_reload_wait())
 
-        server_instance.add_routes([
-            web.get("/frontend/{path:.*}", self.handle_frontend_assets),
-            web.get("/plugins", self.get_plugins),
-            web.get("/plugins/{plugin_name}/frontend_bundle", self.handle_frontend_bundle),
-            web.post("/plugins/{plugin_name}/methods/{method_name}", self.handle_plugin_method_call),
-            web.get("/plugins/{plugin_name}/assets/{path:.*}", self.handle_plugin_frontend_assets),
-
-            # The following is legacy plugin code.
-            web.get("/plugins/load_main/{name}", self.load_plugin_main_view),
-            web.get("/plugins/plugin_resource/{name}/{path:.+}", self.handle_sub_route),
-            web.get("/steam_resource/{path:.+}", self.get_steam_resource)
-        ])
+        server_instance.add_routes(
+            [
+                web.get("/frontend/{path:.*}", self.handle_frontend_assets),
+                web.get("/plugins", self.get_plugins),
+                web.get(
+                    "/plugins/{plugin_name}/frontend_bundle",
+                    self.handle_frontend_bundle,
+                ),
+                web.post(
+                    "/plugins/{plugin_name}/methods/{method_name}",
+                    self.handle_plugin_method_call,
+                ),
+                web.get(
+                    "/plugins/{plugin_name}/assets/{path:.*}",
+                    self.handle_plugin_frontend_assets,
+                ),
+                # The following is legacy plugin code.
+                web.get("/plugins/load_main/{name}", self.load_plugin_main_view),
+                web.get(
+                    "/plugins/plugin_resource/{name}/{path:.+}", self.handle_sub_route
+                ),
+                web.get("/steam_resource/{path:.+}", self.get_steam_resource),
+            ]
+        )
 
     async def enable_reload_wait(self):
         if self.live_reload:
@@ -107,36 +122,63 @@ class Loader:
 
     async def get_plugins(self, request):
         plugins = list(self.plugins.values())
-        return web.json_response([{"name": str(i) if not i.legacy else "$LEGACY_"+str(i), "version": i.version} for i in plugins])
+        return web.json_response(
+            [
+                {
+                    "name": str(i) if not i.legacy else "$LEGACY_" + str(i),
+                    "version": i.version,
+                }
+                for i in plugins
+            ]
+        )
 
     def handle_plugin_frontend_assets(self, request):
         plugin = self.plugins[request.match_info["plugin_name"]]
-        file = path.join(self.plugin_path, plugin.plugin_directory, "dist/assets", request.match_info["path"])
+        file = path.join(
+            self.plugin_path,
+            plugin.plugin_directory,
+            "dist/assets",
+            request.match_info["path"],
+        )
 
         return web.FileResponse(file, headers={"Cache-Control": "no-cache"})
 
     def handle_frontend_bundle(self, request):
         plugin = self.plugins[request.match_info["plugin_name"]]
 
-        with open(path.join(self.plugin_path, plugin.plugin_directory, "dist/index.js"), "r", encoding="utf-8") as bundle:
-            return web.Response(text=bundle.read(), content_type="application/javascript")
+        with open(
+            path.join(self.plugin_path, plugin.plugin_directory, "dist/index.js"),
+            "r",
+            encoding="utf-8",
+        ) as bundle:
+            return web.Response(
+                text=bundle.read(), content_type="application/javascript"
+            )
 
     def import_plugin(self, file, plugin_directory, refresh=False, batch=False):
         try:
             plugin = PluginWrapper(file, plugin_directory, self.plugin_path)
             if plugin.name in self.plugins:
-                    if not "debug" in plugin.flags and refresh:
-                        self.logger.info(f"Plugin {plugin.name} is already loaded and has requested to not be re-loaded")
-                        return
-                    else:
-                        self.plugins[plugin.name].stop()
-                        self.plugins.pop(plugin.name, None)
+                if "debug" not in plugin.flags and refresh:
+                    self.logger.info(
+                        f"Plugin {plugin.name} is already loaded and has requested to"
+                        " not be re-loaded"
+                    )
+                    return
+                else:
+                    self.plugins[plugin.name].stop()
+                    self.plugins.pop(plugin.name, None)
             if plugin.passive:
                 self.logger.info(f"Plugin {plugin.name} is passive")
             self.plugins[plugin.name] = plugin.start()
             self.logger.info(f"Loaded {plugin.name}")
             if not batch:
-                self.loop.create_task(self.dispatch_plugin(plugin.name if not plugin.legacy else "$LEGACY_" + plugin.name, plugin.version))
+                self.loop.create_task(
+                    self.dispatch_plugin(
+                        plugin.name if not plugin.legacy else "$LEGACY_" + plugin.name,
+                        plugin.version,
+                    )
+                )
         except Exception as e:
             self.logger.error(f"Could not load {file}. {e}")
             print_exc()
@@ -148,10 +190,20 @@ class Loader:
     def import_plugins(self):
         self.logger.info(f"import plugins from {self.plugin_path}")
 
-        directories = [i for i in listdir(self.plugin_path) if path.isdir(path.join(self.plugin_path, i)) and path.isfile(path.join(self.plugin_path, i, "plugin.json"))]
+        directories = [
+            i
+            for i in listdir(self.plugin_path)
+            if path.isdir(path.join(self.plugin_path, i))
+            and path.isfile(path.join(self.plugin_path, i, "plugin.json"))
+        ]
         for directory in directories:
             self.logger.info(f"found plugin: {directory}")
-            self.import_plugin(path.join(self.plugin_path, directory, "main.py"), directory, False, True)
+            self.import_plugin(
+                path.join(self.plugin_path, directory, "main.py"),
+                directory,
+                False,
+                True,
+            )
 
     async def handle_reloads(self):
         while True:
@@ -168,10 +220,10 @@ class Loader:
         except JSONDecodeError:
             args = {}
         try:
-          if method_name.startswith("_"):
-              raise RuntimeError("Tried to call private method")
-          res["result"] = await plugin.execute_method(method_name, args)
-          res["success"] = True
+            if method_name.startswith("_"):
+                raise RuntimeError("Tried to call private method")
+            res["result"] = await plugin.execute_method(method_name, args)
+            res["success"] = True
         except Exception as e:
             res["result"] = str(e)
             res["success"] = False
@@ -184,9 +236,14 @@ class Loader:
     can introduce it more smoothly and give people the chance to sample the new features even
     without plugin support. They will be removed once legacy plugins are no longer relevant.
     """
+
     async def load_plugin_main_view(self, request):
         plugin = self.plugins[request.match_info["name"]]
-        with open(path.join(self.plugin_path, plugin.plugin_directory, plugin.main_view_html), "r", encoding="utf-8") as template:
+        with open(
+            path.join(self.plugin_path, plugin.plugin_directory, plugin.main_view_html),
+            "r",
+            encoding="utf-8",
+        ) as template:
             template_data = template.read()
             ret = f"""
             <script src="/legacy/library.js"></script>
@@ -210,6 +267,11 @@ class Loader:
     async def get_steam_resource(self, request):
         tab = await get_tab("SP")
         try:
-            return web.Response(text=await tab.get_steam_resource(f"https://steamloopback.host/{request.match_info['path']}"), content_type="text/html")
+            return web.Response(
+                text=await tab.get_steam_resource(
+                    f"https://steamloopback.host/{request.match_info['path']}"
+                ),
+                content_type="text/html",
+            )
         except Exception as e:
             return web.Response(text=str(e), status=400)

backend/main.py

@@ -1,27 +1,35 @@
 # Change PyInstaller files permissions
 import sys
 from subprocess import call
-if hasattr(sys, '_MEIPASS'):
-    call(['chmod', '-R', '755', sys._MEIPASS])
+
+if hasattr(sys, "_MEIPASS"):
+    call(["chmod", "-R", "755", sys._MEIPASS])
 # Full imports
 from asyncio import new_event_loop, set_event_loop, sleep
-from json import dumps, loads
-from logging import DEBUG, INFO, basicConfig, getLogger
-from os import getenv, chmod, path
+from logging import basicConfig, getLogger
+from os import getenv, path
 from traceback import format_exc
 
 import aiohttp_cors
+
 # Partial imports
-from aiohttp import client_exceptions, WSMsgType
+from aiohttp import client_exceptions
 from aiohttp.web import Application, Response, get, run_app, static
 from aiohttp_jinja2 import setup as jinja_setup
 
 # local modules
 from browser import PluginBrowser
-from helpers import (REMOTE_DEBUGGER_UNIT, csrf_middleware, get_csrf_token,
-                     get_home_path, get_homebrew_path, get_user, get_user_group,
-                     stop_systemd_unit, start_systemd_unit)
-from injector import get_gamepadui_tab, Tab, get_tabs, close_old_tabs
+from helpers import (
+    REMOTE_DEBUGGER_UNIT,
+    csrf_middleware,
+    get_csrf_token,
+    get_homebrew_path,
+    get_user,
+    get_user_group,
+    stop_systemd_unit,
+    start_systemd_unit,
+)
+from injector import get_gamepadui_tab, Tab, close_old_tabs
 from loader import Loader
 from settings import SettingsManager
 from updater import Updater
@@ -42,35 +50,45 @@ CONFIG = {
 }
 
 basicConfig(
-    level=CONFIG["log_level"],
-    format="[%(module)s][%(levelname)s]: %(message)s"
+    level=CONFIG["log_level"], format="[%(module)s][%(levelname)s]: %(message)s"
 )
 
 logger = getLogger("Main")
 
+
 def chown_plugin_dir():
-    code_chown = call(["chown", "-R", USER+":"+GROUP, CONFIG["plugin_path"]])
+    code_chown = call(["chown", "-R", USER + ":" + GROUP, CONFIG["plugin_path"]])
     code_chmod = call(["chmod", "-R", "555", CONFIG["plugin_path"]])
     if code_chown != 0 or code_chmod != 0:
-        logger.error(f"chown/chmod exited with a non-zero exit code (chown: {code_chown}, chmod: {code_chmod})")
+        logger.error(
+            f"chown/chmod exited with a non-zero exit code (chown: {code_chown}, chmod:"
+            f" {code_chmod})"
+        )
 
-if CONFIG["chown_plugin_path"] == True:
+
+if CONFIG["chown_plugin_path"] is True:
     chown_plugin_dir()
 
+
 class PluginManager:
     def __init__(self, loop) -> None:
         self.loop = loop
         self.web_app = Application()
         self.web_app.middlewares.append(csrf_middleware)
-        self.cors = aiohttp_cors.setup(self.web_app, defaults={
-            "https://steamloopback.host": aiohttp_cors.ResourceOptions(
-                expose_headers="*",
-                allow_headers="*",
-                allow_credentials=True
-            )
-        })
-        self.plugin_loader = Loader(self.web_app, CONFIG["plugin_path"], self.loop, CONFIG["live_reload"])
-        self.plugin_browser = PluginBrowser(CONFIG["plugin_path"], self.plugin_loader.plugins, self.plugin_loader)
+        self.cors = aiohttp_cors.setup(
+            self.web_app,
+            defaults={
+                "https://steamloopback.host": aiohttp_cors.ResourceOptions(
+                    expose_headers="*", allow_headers="*", allow_credentials=True
+                )
+            },
+        )
+        self.plugin_loader = Loader(
+            self.web_app, CONFIG["plugin_path"], self.loop, CONFIG["live_reload"]
+        )
+        self.plugin_browser = PluginBrowser(
+            CONFIG["plugin_path"], self.plugin_loader.plugins, self.plugin_loader
+        )
         self.settings = SettingsManager("loader", path.join(HOMEBREW_PATH, "settings"))
         self.utilities = Utilities(self)
         self.updater = Updater(self)
@@ -92,8 +110,12 @@ class PluginManager:
 
         for route in list(self.web_app.router.routes()):
             self.cors.add(route)
-        self.web_app.add_routes([static("/static", path.join(path.dirname(__file__), 'static'))])
-        self.web_app.add_routes([static("/legacy", path.join(path.dirname(__file__), 'legacy'))])
+        self.web_app.add_routes(
+            [static("/static", path.join(path.dirname(__file__), "static"))]
+        )
+        self.web_app.add_routes(
+            [static("/legacy", path.join(path.dirname(__file__), "legacy"))]
+        )
 
     def exception_handler(self, loop, context):
         if context["message"] == "Unclosed connection":
@@ -117,7 +139,10 @@ class PluginManager:
             while not tab:
                 try:
                     tab = await get_gamepadui_tab()
-                except (client_exceptions.ClientConnectorError, client_exceptions.ServerDisconnectedError):
+                except (
+                    client_exceptions.ClientConnectorError,
+                    client_exceptions.ServerDisconnectedError,
+                ):
                     if not dc:
                         logger.debug("Couldn't connect to debugger, waiting...")
                         dc = True
@@ -148,7 +173,7 @@ class PluginManager:
                 # This is because of https://github.com/aio-libs/aiohttp/blob/3ee7091b40a1bc58a8d7846e7878a77640e96996/aiohttp/client_ws.py#L321
                 logger.info("CEF has disconnected...")
                 # At this point the loop starts again and we connect to the freshly started Steam client once it is ready.
-            except Exception as e:
+            except Exception:
                 logger.error("Exception while reading page events " + format_exc())
                 await tab.close_websocket()
                 pass
@@ -164,13 +189,29 @@ class PluginManager:
             if first:
                 if await tab.has_global_var("deckyHasLoaded", False):
                     await close_old_tabs()
-            await tab.evaluate_js("try{if (window.deckyHasLoaded){setTimeout(() => location.reload(), 100)}else{window.deckyHasLoaded = true;(async()=>{try{while(!window.SP_REACT){await new Promise(r => setTimeout(r, 10))};await import('http://localhost:1337/frontend/index.js')}catch(e){console.error(e)};})();}}catch(e){console.error(e)}", False, False, False)
-        except:
+            await tab.evaluate_js(
+                "try{if (window.deckyHasLoaded){setTimeout(() => location.reload(),"
+                " 100)}else{window.deckyHasLoaded ="
+                " true;(async()=>{try{while(!window.SP_REACT){await new Promise(r =>"
+                " setTimeout(r, 10))};await"
+                " import('http://localhost:1337/frontend/index.js')}catch(e){console.error(e)};})();}}catch(e){console.error(e)}",
+                False,
+                False,
+                False,
+            )
+        except Exception:
             logger.info("Failed to inject JavaScript into tab\n" + format_exc())
             pass
 
     def run(self):
-        return run_app(self.web_app, host=CONFIG["server_host"], port=CONFIG["server_port"], loop=self.loop, access_log=None)
+        return run_app(
+            self.web_app,
+            host=CONFIG["server_host"],
+            port=CONFIG["server_port"],
+            loop=self.loop,
+            access_log=None,
+        )
+
 
 if __name__ == "__main__":
     loop = new_event_loop()

backend/plugin.py

@@ -1,8 +1,15 @@
 import multiprocessing
-from asyncio import (Lock, get_event_loop, new_event_loop,
-                     open_unix_connection, set_event_loop, sleep,
-                     start_unix_server, IncompleteReadError, LimitOverrunError)
-from concurrent.futures import ProcessPoolExecutor
+from asyncio import (
+    Lock,
+    get_event_loop,
+    new_event_loop,
+    open_unix_connection,
+    set_event_loop,
+    sleep,
+    start_unix_server,
+    IncompleteReadError,
+    LimitOverrunError,
+)
 from importlib.util import module_from_spec, spec_from_file_location
 from json import dumps, load, loads
 from logging import getLogger
@@ -12,11 +19,11 @@ from signal import SIGINT, signal
 from sys import exit
 from time import time
 import helpers
-from updater import Updater
 
 multiprocessing.set_start_method("fork")
 
-BUFFER_LIMIT = 2 ** 20  # 1 MiB
+BUFFER_LIMIT = 2**20  # 1 MiB
+
 
 class PluginWrapper:
     def __init__(self, file, plugin_directory, plugin_path) -> None:
@@ -30,12 +37,23 @@ class PluginWrapper:
 
         self.version = None
 
-        json = load(open(path.join(plugin_path, plugin_directory, "plugin.json"), "r", encoding="utf-8"))
+        json = load(
+            open(
+                path.join(plugin_path, plugin_directory, "plugin.json"),
+                "r",
+                encoding="utf-8",
+            )
+        )
         if path.isfile(path.join(plugin_path, plugin_directory, "package.json")):
-            package_json = load(open(path.join(plugin_path, plugin_directory, "package.json"), "r", encoding="utf-8"))
+            package_json = load(
+                open(
+                    path.join(plugin_path, plugin_directory, "package.json"),
+                    "r",
+                    encoding="utf-8",
+                )
+            )
             self.version = package_json["version"]
 
-
         self.legacy = False
         self.main_view_html = json["main_view_html"] if "main_view_html" in json else ""
         self.tile_view_html = json["tile_view_html"] if "tile_view_html" in json else ""
@@ -62,18 +80,28 @@ class PluginWrapper:
             setgid(0 if "root" in self.flags else helpers.get_user_group_id())
             setuid(0 if "root" in self.flags else helpers.get_user_id())
             # export a bunch of environment variables to help plugin developers
-            environ["HOME"] = helpers.get_home_path("root" if "root" in self.flags else helpers.get_user())
+            environ["HOME"] = helpers.get_home_path(
+                "root" if "root" in self.flags else helpers.get_user()
+            )
             environ["USER"] = "root" if "root" in self.flags else helpers.get_user()
             environ["DECKY_VERSION"] = helpers.get_loader_version()
             environ["DECKY_USER"] = helpers.get_user()
             environ["DECKY_HOME"] = helpers.get_homebrew_path()
-            environ["DECKY_PLUGIN_SETTINGS_DIR"] = path.join(environ["DECKY_HOME"], "settings", self.plugin_directory)
+            environ["DECKY_PLUGIN_SETTINGS_DIR"] = path.join(
+                environ["DECKY_HOME"], "settings", self.plugin_directory
+            )
             helpers.mkdir_as_user(environ["DECKY_PLUGIN_SETTINGS_DIR"])
-            environ["DECKY_PLUGIN_RUNTIME_DIR"] = path.join(environ["DECKY_HOME"], "data", self.plugin_directory)
+            environ["DECKY_PLUGIN_RUNTIME_DIR"] = path.join(
+                environ["DECKY_HOME"], "data", self.plugin_directory
+            )
             helpers.mkdir_as_user(environ["DECKY_PLUGIN_RUNTIME_DIR"])
-            environ["DECKY_PLUGIN_LOG_DIR"] = path.join(environ["DECKY_HOME"], "logs", self.plugin_directory)
+            environ["DECKY_PLUGIN_LOG_DIR"] = path.join(
+                environ["DECKY_HOME"], "logs", self.plugin_directory
+            )
             helpers.mkdir_as_user(environ["DECKY_PLUGIN_LOG_DIR"])
-            environ["DECKY_PLUGIN_DIR"] = path.join(self.plugin_path, self.plugin_directory)
+            environ["DECKY_PLUGIN_DIR"] = path.join(
+                self.plugin_path, self.plugin_directory
+            )
             environ["DECKY_PLUGIN_NAME"] = self.name
             environ["DECKY_PLUGIN_VERSION"] = self.version
             environ["DECKY_PLUGIN_AUTHOR"] = self.author
@@ -86,24 +114,32 @@ class PluginWrapper:
                 get_event_loop().create_task(self.Plugin._main(self.Plugin))
             get_event_loop().create_task(self._setup_socket())
             get_event_loop().run_forever()
-        except:
+        except Exception:
             self.log.error("Failed to start " + self.name + "!\n" + format_exc())
             exit(0)
 
     async def _unload(self):
         try:
-            self.log.info("Attempting to unload with plugin " + self.name + "'s \"_unload\" function.\n")
+            self.log.info(
+                "Attempting to unload with plugin "
+                + self.name
+                + '\'s "_unload" function.\n'
+            )
             if hasattr(self.Plugin, "_unload"):
                 await self.Plugin._unload(self.Plugin)
                 self.log.info("Unloaded " + self.name + "\n")
             else:
-                self.log.info("Could not find \"_unload\" in " + self.name + "'s main.py" + "\n")
-        except:
+                self.log.info(
+                    'Could not find "_unload" in ' + self.name + "'s main.py" + "\n"
+                )
+        except Exception:
             self.log.error("Failed to unload " + self.name + "!\n" + format_exc())
             exit(0)
 
     async def _setup_socket(self):
-        self.socket = await start_unix_server(self._listen_for_method_call, path=self.socket_addr, limit=BUFFER_LIMIT)
+        self.socket = await start_unix_server(
+            self._listen_for_method_call, path=self.socket_addr, limit=BUFFER_LIMIT
+        )
 
     async def _listen_for_method_call(self, reader, writer):
         while True:
@@ -130,12 +166,14 @@ class PluginWrapper:
                 return
             d = {"res": None, "success": True}
             try:
-                d["res"] = await getattr(self.Plugin, data["method"])(self.Plugin, **data["args"])
+                d["res"] = await getattr(self.Plugin, data["method"])(
+                    self.Plugin, **data["args"]
+                )
             except Exception as e:
                 d["res"] = str(e)
                 d["success"] = False
             finally:
-                writer.write((dumps(d, ensure_ascii=False)+"\n").encode("utf-8"))
+                writer.write((dumps(d, ensure_ascii=False) + "\n").encode("utf-8"))
                 await writer.drain()
 
     async def _open_socket_if_not_exists(self):
@@ -143,9 +181,11 @@ class PluginWrapper:
             retries = 0
             while retries < 10:
                 try:
-                    self.reader, self.writer = await open_unix_connection(self.socket_addr, limit=BUFFER_LIMIT)
+                    self.reader, self.writer = await open_unix_connection(
+                        self.socket_addr, limit=BUFFER_LIMIT
+                    )
                     return True
-                except:
+                except Exception:
                     await sleep(2)
                     retries += 1
             return False
@@ -161,20 +201,32 @@ class PluginWrapper:
     def stop(self):
         if self.passive:
             return
+
         async def _(self):
             if await self._open_socket_if_not_exists():
-                self.writer.write((dumps({ "stop": True }, ensure_ascii=False)+"\n").encode("utf-8"))
+                self.writer.write(
+                    (dumps({"stop": True}, ensure_ascii=False) + "\n").encode("utf-8")
+                )
                 await self.writer.drain()
                 self.writer.close()
+
         get_event_loop().create_task(_(self))
 
     async def execute_method(self, method_name, kwargs):
         if self.passive:
-            raise RuntimeError("This plugin is passive (aka does not implement main.py)")
+            raise RuntimeError(
+                "This plugin is passive (aka does not implement main.py)"
+            )
         async with self.method_call_lock:
             if await self._open_socket_if_not_exists():
                 self.writer.write(
-                    (dumps({ "method": method_name, "args": kwargs }, ensure_ascii=False) + "\n").encode("utf-8"))
+                    (
+                        dumps(
+                            {"method": method_name, "args": kwargs}, ensure_ascii=False
+                        )
+                        + "\n"
+                    ).encode("utf-8")
+                )
                 await self.writer.drain()
                 line = bytearray()
                 while True:

backend/pyproject.toml

@@ -0,0 +1,15 @@
+[flake8]
+max-line-length = 88
+
+[tool.ruff]
+ignore = [
+    # Ignore line length check and let Black handle it
+    "E501",
+
+    # Ignore SyntaxError due to ruff not supporting pattern matching
+    # https://github.com/charliermarsh/ruff/issues/282
+    "E999",
+]
+
+# Assume Python 3.10.
+target-version = "py310"

backend/settings.py

@@ -2,33 +2,36 @@ from json import dump, load
 from os import mkdir, path, listdir, rename
 from shutil import chown
 
-from helpers import get_home_path, get_homebrew_path, get_user, get_user_group, get_user_owner
+from helpers import (
+    get_homebrew_path,
+    get_user,
+    get_user_group,
+    get_user_owner,
+)
 
 
 class SettingsManager:
-    def __init__(self, name, settings_directory = None) -> None:
+    def __init__(self, name, settings_directory=None) -> None:
         USER = get_user()
         GROUP = get_user_group()
         wrong_dir = get_homebrew_path()
-        if settings_directory == None:
+        if settings_directory is None:
             settings_directory = path.join(wrong_dir, "settings")
 
         self.path = path.join(settings_directory, name + ".json")
 
-        #Create the folder with the correct permission
+        # Create the folder with the correct permission
         if not path.exists(settings_directory):
             mkdir(settings_directory)
             chown(settings_directory, USER, GROUP)
 
-        #Copy all old settings file in the root directory to the correct folder
+        # Copy all old settings file in the root directory to the correct folder
         for file in listdir(wrong_dir):
             if file.endswith(".json"):
-                rename(path.join(wrong_dir,file),
-                       path.join(settings_directory, file)) 
+                rename(path.join(wrong_dir, file), path.join(settings_directory, file))
                 self.path = path.join(settings_directory, name + ".json")
 
-
-        #If the owner of the settings directory is not the user, then set it as the user:
+        # If the owner of the settings directory is not the user, then set it as the user:
         if get_user_owner(settings_directory) != USER:
             chown(settings_directory, USER, GROUP)
 
@@ -36,7 +39,7 @@ class SettingsManager:
 
         try:
             open(self.path, "x", encoding="utf-8")
-        except FileExistsError as e:
+        except FileExistsError:
             self.read()
             pass
 

backend/updater.py

@@ -16,6 +16,7 @@ from settings import SettingsManager
 
 logger = getLogger("Updater")
 
+
 class Updater:
     def __init__(self, context) -> None:
         self.context = context
@@ -26,7 +27,7 @@ class Updater:
             "get_version": self.get_version,
             "do_update": self.do_update,
             "do_restart": self.do_restart,
-            "check_for_updates": self.check_for_updates
+            "check_for_updates": self.check_for_updates,
         }
         self.remoteVer = None
         self.allRemoteVers = None
@@ -39,12 +40,14 @@ class Updater:
             self.currentBranch = self.get_branch(self.context.settings)
         except:
             self.currentBranch = 0
-            logger.error("Current branch could not be determined, defaulting to \"Stable\"")
+            logger.error(
+                'Current branch could not be determined, defaulting to "Stable"'
+            )
 
         if context:
-            context.web_app.add_routes([
-                web.post("/updater/{method_name}", self._handle_server_method_call)
-            ])
+            context.web_app.add_routes(
+                [web.post("/updater/{method_name}", self._handle_server_method_call)]
+            )
             context.loop.create_task(self.version_reloader())
 
     async def _handle_server_method_call(self, request):
@@ -89,7 +92,10 @@ class Updater:
             case 1 | 2:
                 url = "https://raw.githubusercontent.com/SteamDeckHomebrew/decky-loader/main/dist/plugin_loader-prerelease.service"
             case _:
-                logger.error("You have an invalid branch set... Defaulting to prerelease service, please send the logs to the devs!")
+                logger.error(
+                    "You have an invalid branch set... Defaulting to prerelease"
+                    " service, please send the logs to the devs!"
+                )
                 url = "https://raw.githubusercontent.com/SteamDeckHomebrew/decky-loader/main/dist/plugin_loader-prerelease.service"
         return str(url)
 
@@ -99,31 +105,58 @@ class Updater:
                 "current": self.localVer,
                 "remote": self.remoteVer,
                 "all": self.allRemoteVers,
-                "updatable": self.localVer != None
+                "updatable": self.localVer != None,
             }
         else:
-            return {"current": "unknown", "remote": self.remoteVer, "all": self.allRemoteVers, "updatable": False}
+            return {
+                "current": "unknown",
+                "remote": self.remoteVer,
+                "all": self.allRemoteVers,
+                "updatable": False,
+            }
 
     async def check_for_updates(self):
         logger.debug("checking for updates")
         selectedBranch = self.get_branch(self.context.settings)
         async with ClientSession() as web:
-            async with web.request("GET", "https://api.github.com/repos/SteamDeckHomebrew/decky-loader/releases", ssl=helpers.get_ssl_context()) as res:
+            async with web.request(
+                "GET",
+                "https://api.github.com/repos/SteamDeckHomebrew/decky-loader/releases",
+                ssl=helpers.get_ssl_context(),
+            ) as res:
                 remoteVersions = await res.json()
         self.allRemoteVers = remoteVersions
         logger.debug("determining release type to find, branch is %i" % selectedBranch)
         if selectedBranch == 0:
             logger.debug("release type: release")
-            self.remoteVer = next(filter(lambda ver: ver["tag_name"].startswith("v") and not ver["prerelease"] and ver["tag_name"], remoteVersions), None)
+            self.remoteVer = next(
+                filter(
+                    lambda ver: ver["tag_name"].startswith("v")
+                    and not ver["prerelease"]
+                    and ver["tag_name"],
+                    remoteVersions,
+                ),
+                None,
+            )
         elif selectedBranch == 1:
             logger.debug("release type: pre-release")
-            self.remoteVer = next(filter(lambda ver: ver["prerelease"] and ver["tag_name"].startswith("v") and ver["tag_name"].find("-pre"), remoteVersions), None)
+            self.remoteVer = next(
+                filter(
+                    lambda ver: ver["prerelease"]
+                    and ver["tag_name"].startswith("v")
+                    and ver["tag_name"].find("-pre"),
+                    remoteVersions,
+                ),
+                None,
+            )
         else:
             logger.error("release type: NOT FOUND")
             raise ValueError("no valid branch found")
         logger.info("Updated remote version information")
         tab = await get_gamepadui_tab()
-        await tab.evaluate_js(f"window.DeckyPluginLoader.notifyUpdates()", False, True, False)
+        await tab.evaluate_js(
+            f"window.DeckyPluginLoader.notifyUpdates()", False, True, False
+        )
         return await self.get_version()
 
     async def version_reloader(self):
@@ -133,7 +166,7 @@ class Updater:
                 await self.check_for_updates()
             except:
                 pass
-            await sleep(60 * 60 * 6) # 6 hours
+            await sleep(60 * 60 * 6)  # 6 hours
 
     async def do_update(self):
         logger.debug("Starting update.")
@@ -147,7 +180,9 @@ class Updater:
         async with ClientSession() as web:
             logger.debug("Downloading systemd service")
             # download the relevant systemd service depending upon branch
-            async with web.request("GET", service_url, ssl=helpers.get_ssl_context(), allow_redirects=True) as res:
+            async with web.request(
+                "GET", service_url, ssl=helpers.get_ssl_context(), allow_redirects=True
+            ) as res:
                 logger.debug("Downloading service file")
                 data = await res.content.read()
             logger.debug(str(data))
@@ -157,22 +192,33 @@ class Updater:
                     out.write(data)
             except Exception as e:
                 logger.error(f"Error at %s", exc_info=e)
-            with open(path.join(getcwd(), "plugin_loader.service"), "r", encoding="utf-8") as service_file:
+            with open(
+                path.join(getcwd(), "plugin_loader.service"), "r", encoding="utf-8"
+            ) as service_file:
                 service_data = service_file.read()
-            service_data = service_data.replace("${HOMEBREW_FOLDER}", helpers.get_homebrew_path())
-            with open(path.join(getcwd(), "plugin_loader.service"), "w", encoding="utf-8") as service_file:
-                    service_file.write(service_data)
-                    
+            service_data = service_data.replace(
+                "${HOMEBREW_FOLDER}", helpers.get_homebrew_path()
+            )
+            with open(
+                path.join(getcwd(), "plugin_loader.service"), "w", encoding="utf-8"
+            ) as service_file:
+                service_file.write(service_data)
+
             logger.debug("Saved service file")
             logger.debug("Copying service file over current file.")
             shutil.copy(service_file_path, "/etc/systemd/system/plugin_loader.service")
             if not os.path.exists(path.join(getcwd(), ".systemd")):
                 os.mkdir(path.join(getcwd(), ".systemd"))
-            shutil.move(service_file_path, path.join(getcwd(), ".systemd")+"/plugin_loader.service")
-            
+            shutil.move(
+                service_file_path,
+                path.join(getcwd(), ".systemd") + "/plugin_loader.service",
+            )
+
             logger.debug("Downloading binary")
-            async with web.request("GET", download_url, ssl=helpers.get_ssl_context(), allow_redirects=True) as res:
-                total = int(res.headers.get('content-length', 0))
+            async with web.request(
+                "GET", download_url, ssl=helpers.get_ssl_context(), allow_redirects=True
+            ) as res:
+                total = int(res.headers.get("content-length", 0))
                 # we need to not delete the binary until we have downloaded the new binary!
                 try:
                     remove(path.join(getcwd(), "PluginLoader"))
@@ -186,13 +232,22 @@ class Updater:
                         raw += len(c)
                         new_progress = round((raw / total) * 100)
                         if progress != new_progress:
-                            self.context.loop.create_task(tab.evaluate_js(f"window.DeckyUpdater.updateProgress({new_progress})", False, False, False))
+                            self.context.loop.create_task(
+                                tab.evaluate_js(
+                                    f"window.DeckyUpdater.updateProgress({new_progress})",
+                                    False,
+                                    False,
+                                    False,
+                                )
+                            )
                             progress = new_progress
 
-            with open(path.join(getcwd(), ".loader.version"), "w", encoding="utf-8") as out:
+            with open(
+                path.join(getcwd(), ".loader.version"), "w", encoding="utf-8"
+            ) as out:
                 out.write(version)
 
-            call(['chmod', '+x', path.join(getcwd(), "PluginLoader")])
+            call(["chmod", "+x", path.join(getcwd(), "PluginLoader")])
             logger.info("Updated loader installation.")
             await tab.evaluate_js("window.DeckyUpdater.finish()", False, False)
             await self.do_restart()

backend/utilities.py

@@ -3,13 +3,12 @@ import os
 from json.decoder import JSONDecodeError
 from traceback import format_exc
 
-from asyncio import sleep, start_server, gather, open_connection
+from asyncio import start_server, gather, open_connection
 from aiohttp import ClientSession, web
 
 from logging import getLogger
 from injector import inject_to_tab, get_gamepadui_tab, close_old_tabs
 import helpers
-import subprocess
 
 
 class Utilities:
@@ -31,7 +30,7 @@ class Utilities:
             "get_setting": self.get_setting,
             "filepicker_ls": self.filepicker_ls,
             "disable_rdt": self.disable_rdt,
-            "enable_rdt": self.enable_rdt
+            "enable_rdt": self.enable_rdt,
         }
 
         self.logger = getLogger("Utilities")
@@ -41,9 +40,9 @@ class Utilities:
         self.rdt_proxy_task = None
 
         if context:
-            context.web_app.add_routes([
-                web.post("/methods/{method_name}", self._handle_server_method_call)
-            ])
+            context.web_app.add_routes(
+                [web.post("/methods/{method_name}", self._handle_server_method_call)]
+            )
 
     async def _handle_server_method_call(self, request):
         method_name = request.match_info["method_name"]
@@ -61,12 +60,11 @@ class Utilities:
             res["success"] = False
         return web.json_response(res)
 
-    async def install_plugin(self, artifact="", name="No name", version="dev", hash=False):
+    async def install_plugin(
+        self, artifact="", name="No name", version="dev", hash=False
+    ):
         return await self.context.plugin_browser.request_plugin_install(
-            artifact=artifact,
-            name=name,
-            version=version,
-            hash=hash
+            artifact=artifact, name=name, version=version, hash=hash
         )
 
     async def confirm_plugin_install(self, request_id):
@@ -80,13 +78,11 @@ class Utilities:
 
     async def http_request(self, method="", url="", **kwargs):
         async with ClientSession() as web:
-            res = await web.request(method, url, ssl=helpers.get_ssl_context(), **kwargs)
+            res = await web.request(
+                method, url, ssl=helpers.get_ssl_context(), **kwargs
+            )
             text = await res.text()
-        return {
-            "status": res.status,
-            "headers": dict(res.headers),
-            "body": text
-        }
+        return {"status": res.status, "headers": dict(res.headers), "body": text}
 
     async def ping(self, **kwargs):
         return "pong"
@@ -95,26 +91,18 @@ class Utilities:
         try:
             result = await inject_to_tab(tab, code, run_async)
             if "exceptionDetails" in result["result"]:
-                return {
-                    "success": False,
-                    "result": result["result"]
-                }
+                return {"success": False, "result": result["result"]}
 
-            return {
-                "success": True,
-                "result": result["result"]["result"].get("value")
-            }
+            return {"success": True, "result": result["result"]["result"].get("value")}
         except Exception as e:
-            return {
-              "success": False,
-              "result": e
-            }
+            return {"success": False, "result": e}
 
     async def inject_css_into_tab(self, tab, style):
         try:
             css_id = str(uuid.uuid4())
 
-            result = await inject_to_tab(tab,
+            result = await inject_to_tab(
+                tab,
                 f"""
                 (function() {{
                     const style = document.createElement('style');
@@ -122,27 +110,21 @@ class Utilities:
                     document.head.append(style);
                     style.textContent = `{style}`;
                 }})()
-                """, False)
+                """,
+                False,
+            )
 
             if "exceptionDetails" in result["result"]:
-                return {
-                    "success": False,
-                    "result": result["result"]
-                }
+                return {"success": False, "result": result["result"]}
 
-            return {
-                "success": True,
-                "result": css_id
-            }
+            return {"success": True, "result": css_id}
         except Exception as e:
-            return {
-                "success": False,
-                "result": e
-            }
+            return {"success": False, "result": e}
 
     async def remove_css_from_tab(self, tab, css_id):
         try:
-            result = await inject_to_tab(tab,
+            result = await inject_to_tab(
+                tab,
                 f"""
                 (function() {{
                     let style = document.getElementById("{css_id}");
@@ -150,22 +132,16 @@ class Utilities:
                     if (style.nodeName.toLowerCase() == 'style')
                         style.parentNode.removeChild(style);
                 }})()
-                """, False)
+                """,
+                False,
+            )
 
             if "exceptionDetails" in result["result"]:
-                return {
-                    "success": False,
-                    "result": result
-                }
+                return {"success": False, "result": result}
 
-            return {
-                "success": True
-            }
+            return {"success": True}
         except Exception as e:
-            return {
-                "success": False,
-                "result": e
-            }
+            return {"success": False, "result": e}
 
     async def get_setting(self, key, default):
         return self.context.settings.getSetting(key, default)
@@ -187,7 +163,7 @@ class Utilities:
         #         return os.path.getmtime(os.path.join(path, file))
         #     return 0
         # file_names = sorted(os.listdir(path), key=sorter, reverse=True) # TODO provide more sort options
-        file_names = sorted(os.listdir(path)) # Alphabetical
+        file_names = sorted(os.listdir(path))  # Alphabetical
 
         files = []
 
@@ -196,16 +172,15 @@ class Utilities:
             is_dir = os.path.isdir(full_path)
 
             if is_dir or include_files:
-                files.append({
-                    "isdir": is_dir,
-                    "name": file,
-                    "realpath": os.path.realpath(full_path)
-                })
+                files.append(
+                    {
+                        "isdir": is_dir,
+                        "name": file,
+                        "realpath": os.path.realpath(full_path),
+                    }
+                )
 
-        return {
-            "realpath": os.path.realpath(path),
-            "files": files
-        }
+        return {"realpath": os.path.realpath(path), "files": files}
 
     # Based on https://stackoverflow.com/a/46422554/13174603
     def start_rdt_proxy(self, ip, port):
@@ -215,10 +190,10 @@ class Utilities:
                     writer.write(await reader.read(2048))
             finally:
                 writer.close()
+
         async def handle_client(local_reader, local_writer):
             try:
-                remote_reader, remote_writer = await open_connection(
-                    ip, port)
+                remote_reader, remote_writer = await open_connection(ip, port)
                 pipe1 = pipe(local_reader, remote_writer)
                 pipe2 = pipe(remote_reader, local_writer)
                 await gather(pipe1, pipe2)
@@ -239,11 +214,14 @@ class Utilities:
             self.stop_rdt_proxy()
             ip = self.context.settings.getSetting("developer.rdt.ip", None)
 
-            if ip != None:
+            if ip is not None:
                 self.logger.info("Connecting to React DevTools at " + ip)
                 async with ClientSession() as web:
-                    res = await web.request("GET", "http://" + ip + ":8097", ssl=helpers.get_ssl_context())
-                    script = """
+                    res = await web.request(
+                        "GET", "http://" + ip + ":8097", ssl=helpers.get_ssl_context()
+                    )
+                    script = (
+                        """
                     if (!window.deckyHasConnectedRDT) {
                         window.deckyHasConnectedRDT = true;
                         // This fixes the overlay when hovering over an element in RDT
@@ -254,7 +232,10 @@ class Utilities:
                                 return FocusNavController?.m_ActiveContext?.ActiveWindow || window;
                             }
                         });
-                    """ + await res.text() + "\n}"
+                    """
+                        + await res.text()
+                        + "\n}"
+                    )
                 if res.status != 200:
                     self.logger.error("Failed to connect to React DevTools at " + ip)
                     return False
@@ -265,7 +246,7 @@ class Utilities:
                 await close_old_tabs()
                 result = await tab.reload_and_evaluate(script)
                 self.logger.info(result)
-                        
+
         except Exception:
             self.logger.error("Failed to connect to React DevTools")
             self.logger.error(format_exc())