mirror of
https://github.com/SteamDeckHomebrew/decky-loader.git
synced 2026-07-11 18:01:54 +00:00
remove some type: ignore and make some specific
This commit is contained in:
@@ -1,4 +1,3 @@
|
|||||||
from platform import version
|
|
||||||
import re
|
import re
|
||||||
import ssl
|
import ssl
|
||||||
import uuid
|
import uuid
|
||||||
@@ -70,10 +69,10 @@ def get_loader_version() -> str:
|
|||||||
|
|
||||||
version_str = f'v{v.major}.{v.minor}.{v.micro}'
|
version_str = f'v{v.major}.{v.minor}.{v.micro}'
|
||||||
|
|
||||||
if v.pre: # type: ignore
|
if v.pre:
|
||||||
version_str += f'-pre{v.pre[1]}'
|
version_str += f'-pre{v.pre[1]}'
|
||||||
|
|
||||||
if v.post: # type: ignore
|
if v.post:
|
||||||
version_str += f'-dev{v.post}'
|
version_str += f'-dev{v.post}'
|
||||||
|
|
||||||
return version_str
|
return version_str
|
||||||
@@ -89,7 +88,7 @@ def get_system_pythonpaths() -> list[str]:
|
|||||||
# run as normal normal user if on linux to also include user python paths
|
# run as normal normal user if on linux to also include user python paths
|
||||||
proc = subprocess.run(["python3" if localplatform.ON_LINUX else "python", "-c", "import sys; print('\\n'.join(x for x in sys.path if x))"],
|
proc = subprocess.run(["python3" if localplatform.ON_LINUX else "python", "-c", "import sys; print('\\n'.join(x for x in sys.path if x))"],
|
||||||
# TODO make this less insane
|
# TODO make this less insane
|
||||||
capture_output=True, user=localplatform.localplatform._get_user_id() if localplatform.ON_LINUX else None, env={} if localplatform.ON_LINUX else None) # type: ignore
|
capture_output=True, user=localplatform.localplatform._get_user_id() if localplatform.ON_LINUX else None, env={} if localplatform.ON_LINUX else None) # pyright: ignore [reportPrivateUsage]
|
||||||
return [x.strip() for x in proc.stdout.decode().strip().split("\n")]
|
return [x.strip() for x in proc.stdout.decode().strip().split("\n")]
|
||||||
except Exception as e:
|
except Exception as e:
|
||||||
logger.warn(f"Failed to execute get_system_pythonpaths(): {str(e)}")
|
logger.warn(f"Failed to execute get_system_pythonpaths(): {str(e)}")
|
||||||
|
|||||||
@@ -33,7 +33,7 @@ class Tab:
|
|||||||
|
|
||||||
async def open_websocket(self):
|
async def open_websocket(self):
|
||||||
self.client = ClientSession()
|
self.client = ClientSession()
|
||||||
self.websocket = await self.client.ws_connect(self.ws_url) # type: ignore
|
self.websocket = await self.client.ws_connect(self.ws_url)
|
||||||
|
|
||||||
async def close_websocket(self):
|
async def close_websocket(self):
|
||||||
if self.websocket:
|
if self.websocket:
|
||||||
|
|||||||
@@ -4,12 +4,12 @@ from logging import getLogger
|
|||||||
from os import listdir, path
|
from os import listdir, path
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from traceback import format_exc, print_exc
|
from traceback import format_exc, print_exc
|
||||||
from typing import Any, Tuple, Dict
|
from typing import Any, Tuple, Dict, cast
|
||||||
|
|
||||||
from aiohttp import web
|
from aiohttp import web
|
||||||
from os.path import exists
|
from os.path import exists
|
||||||
from watchdog.events import RegexMatchingEventHandler, DirCreatedEvent, DirModifiedEvent, FileCreatedEvent, FileModifiedEvent # type: ignore
|
from watchdog.events import RegexMatchingEventHandler, DirCreatedEvent, DirModifiedEvent, FileCreatedEvent, FileModifiedEvent
|
||||||
from watchdog.observers import Observer # type: ignore
|
from watchdog.observers import Observer
|
||||||
|
|
||||||
from typing import TYPE_CHECKING, List
|
from typing import TYPE_CHECKING, List
|
||||||
if TYPE_CHECKING:
|
if TYPE_CHECKING:
|
||||||
@@ -24,7 +24,7 @@ ReloadQueue = Queue[Tuple[str, str, bool | None] | Tuple[str, str]]
|
|||||||
|
|
||||||
class FileChangeHandler(RegexMatchingEventHandler):
|
class FileChangeHandler(RegexMatchingEventHandler):
|
||||||
def __init__(self, queue: ReloadQueue, plugin_path: str) -> None:
|
def __init__(self, queue: ReloadQueue, plugin_path: str) -> None:
|
||||||
super().__init__(regexes=[r'^.*?dist\/index\.js$', r'^.*?main\.py$']) # type: ignore
|
super().__init__(regexes=[r'^.*?dist\/index\.js$', r'^.*?main\.py$']) # pyright: ignore [reportUnknownMemberType]
|
||||||
self.logger = getLogger("file-watcher")
|
self.logger = getLogger("file-watcher")
|
||||||
self.plugin_path = plugin_path
|
self.plugin_path = plugin_path
|
||||||
self.queue = queue
|
self.queue = queue
|
||||||
@@ -37,8 +37,8 @@ class FileChangeHandler(RegexMatchingEventHandler):
|
|||||||
if exists(path.join(self.plugin_path, plugin_dir, "plugin.json")):
|
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: DirCreatedEvent | FileCreatedEvent):
|
def on_created(self, event: DirCreatedEvent | FileCreatedEvent): # pyright: ignore [reportIncompatibleMethodOverride]
|
||||||
src_path = event.src_path
|
src_path = cast(str, event.src_path) # type: ignore the correct type for this is in later versions of watchdog
|
||||||
if "__pycache__" in src_path:
|
if "__pycache__" in src_path:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -51,8 +51,8 @@ class FileChangeHandler(RegexMatchingEventHandler):
|
|||||||
self.logger.debug(f"file created: {src_path}")
|
self.logger.debug(f"file created: {src_path}")
|
||||||
self.maybe_reload(src_path)
|
self.maybe_reload(src_path)
|
||||||
|
|
||||||
def on_modified(self, event: DirModifiedEvent | FileModifiedEvent):
|
def on_modified(self, event: DirModifiedEvent | FileModifiedEvent): # pyright: ignore [reportIncompatibleMethodOverride]
|
||||||
src_path = event.src_path
|
src_path = cast(str, event.src_path) # type: ignore
|
||||||
if "__pycache__" in src_path:
|
if "__pycache__" in src_path:
|
||||||
return
|
return
|
||||||
|
|
||||||
@@ -81,7 +81,7 @@ class Loader:
|
|||||||
if live_reload:
|
if live_reload:
|
||||||
self.observer = Observer()
|
self.observer = Observer()
|
||||||
self.watcher = FileChangeHandler(self.reload_queue, plugin_path)
|
self.watcher = FileChangeHandler(self.reload_queue, plugin_path)
|
||||||
self.observer.schedule(self.watcher, self.plugin_path, recursive=True) # type: ignore
|
self.observer.schedule(self.watcher, self.plugin_path, recursive=True) # pyright: ignore [reportUnknownMemberType]
|
||||||
self.observer.start()
|
self.observer.start()
|
||||||
self.loop.create_task(self.enable_reload_wait())
|
self.loop.create_task(self.enable_reload_wait())
|
||||||
|
|
||||||
@@ -179,7 +179,7 @@ class Loader:
|
|||||||
async def handle_reloads(self):
|
async def handle_reloads(self):
|
||||||
while True:
|
while True:
|
||||||
args = await self.reload_queue.get()
|
args = await self.reload_queue.get()
|
||||||
self.import_plugin(*args) # type: ignore
|
self.import_plugin(*args) # pyright: ignore [reportArgumentType]
|
||||||
|
|
||||||
async def handle_plugin_method_call_legacy(self, plugin_name: str, method_name: str, kwargs: Dict[Any, Any]):
|
async def handle_plugin_method_call_legacy(self, plugin_name: str, method_name: str, kwargs: Dict[Any, Any]):
|
||||||
res: Dict[Any, Any] = {}
|
res: Dict[Any, Any] = {}
|
||||||
|
|||||||
@@ -75,7 +75,7 @@ class UnixSocket:
|
|||||||
try:
|
try:
|
||||||
line.extend(await reader.readuntil())
|
line.extend(await reader.readuntil())
|
||||||
except asyncio.LimitOverrunError:
|
except asyncio.LimitOverrunError:
|
||||||
line.extend(await reader.read(reader._limit)) # type: ignore
|
line.extend(await reader.read(reader._limit)) # pyright: ignore [reportUnknownMemberType, reportUnknownArgumentType, reportAttributeAccessIssue]
|
||||||
continue
|
continue
|
||||||
except asyncio.IncompleteReadError as err:
|
except asyncio.IncompleteReadError as err:
|
||||||
line.extend(err.partial)
|
line.extend(err.partial)
|
||||||
|
|||||||
@@ -14,10 +14,10 @@ from os import path
|
|||||||
from traceback import format_exc
|
from traceback import format_exc
|
||||||
import multiprocessing
|
import multiprocessing
|
||||||
|
|
||||||
import aiohttp_cors # type: ignore
|
import aiohttp_cors # pyright: ignore [reportMissingTypeStubs]
|
||||||
# Partial imports
|
# Partial imports
|
||||||
from aiohttp import client_exceptions
|
from aiohttp import client_exceptions
|
||||||
from aiohttp.web import Application, Response, Request, get, run_app, static # type: ignore
|
from aiohttp.web import Application, Response, Request, get, run_app, static # pyright: ignore [reportUnknownVariableType]
|
||||||
from aiohttp_jinja2 import setup as jinja_setup
|
from aiohttp_jinja2 import setup as jinja_setup
|
||||||
|
|
||||||
# local modules
|
# local modules
|
||||||
@@ -87,7 +87,7 @@ class PluginManager:
|
|||||||
self.web_app.add_routes([get("/auth/token", self.get_auth_token)])
|
self.web_app.add_routes([get("/auth/token", self.get_auth_token)])
|
||||||
|
|
||||||
for route in list(self.web_app.router.routes()):
|
for route in list(self.web_app.router.routes()):
|
||||||
self.cors.add(route) # type: ignore
|
self.cors.add(route) # pyright: ignore [reportUnknownMemberType]
|
||||||
self.web_app.add_routes([static("/static", path.join(path.dirname(__file__), '..', 'static'))])
|
self.web_app.add_routes([static("/static", path.join(path.dirname(__file__), '..', 'static'))])
|
||||||
|
|
||||||
def exception_handler(self, loop: AbstractEventLoop, context: Dict[str, str]):
|
def exception_handler(self, loop: AbstractEventLoop, context: Dict[str, str]):
|
||||||
|
|||||||
@@ -6,7 +6,7 @@ from json.decoder import JSONDecodeError
|
|||||||
from os.path import splitext
|
from os.path import splitext
|
||||||
import re
|
import re
|
||||||
from traceback import format_exc
|
from traceback import format_exc
|
||||||
from stat import FILE_ATTRIBUTE_HIDDEN # type: ignore
|
from stat import FILE_ATTRIBUTE_HIDDEN # pyright: ignore [reportAttributeAccessIssue, reportUnknownVariableType]
|
||||||
|
|
||||||
from asyncio import StreamReader, StreamWriter, start_server, gather, open_connection
|
from asyncio import StreamReader, StreamWriter, start_server, gather, open_connection
|
||||||
from aiohttp import ClientSession
|
from aiohttp import ClientSession
|
||||||
|
|||||||
@@ -24,8 +24,8 @@ class MessageType(IntEnum):
|
|||||||
# WSMessage with slightly better typings
|
# WSMessage with slightly better typings
|
||||||
class WSMessageExtra(WSMessage):
|
class WSMessageExtra(WSMessage):
|
||||||
# TODO message typings here too
|
# TODO message typings here too
|
||||||
data: Any # type: ignore yes you can extend it
|
data: Any # pyright: ignore [reportIncompatibleVariableOverride]
|
||||||
type: WSMsgType # type: ignore
|
type: WSMsgType # pyright: ignore [reportIncompatibleVariableOverride]
|
||||||
|
|
||||||
# see wsrouter.ts for typings
|
# see wsrouter.ts for typings
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user