Technical: typing and cleanup

custom_components/pyscript/__init__.py

@@ -1,14 +1,15 @@
 """Component to allow running Python scripts."""
 
 import asyncio
+from collections.abc import Awaitable, Callable
 import glob
 import json
 import logging
 import os
 import shutil
 import time
 import traceback
-from typing import Any, Callable, Dict, List, Set, Union
+from typing import Any
 
 import voluptuous as vol
 from watchdog.events import DirModifiedEvent, FileSystemEvent, FileSystemEventHandler
@@ -22,7 +23,7 @@
     EVENT_STATE_CHANGED,
     SERVICE_RELOAD,
 )
-from homeassistant.core import Event as HAEvent, HomeAssistant, ServiceCall
+from homeassistant.core import Event as HAEvent, HomeAssistant, ServiceCall, SupportsResponse
 from homeassistant.exceptions import HomeAssistantError
 import homeassistant.helpers.config_validation as cv
 from homeassistant.helpers.restore_state import DATA_RESTORE_STATE
@@ -40,7 +41,6 @@
     REQUIREMENTS_FILE,
     SERVICE_GENERATE_STUBS,
     SERVICE_JUPYTER_KERNEL_START,
-    SERVICE_RESPONSE_ONLY,
     UNSUB_LISTENERS,
     WATCHDOG_TASK,
 )
@@ -98,7 +98,7 @@ async def update_yaml_config(hass: HomeAssistant, config_entry: ConfigEntry) ->
         conf = await async_hass_config_yaml(hass)
     except HomeAssistantError as err:
         _LOGGER.error(err)
-        return
+        return False
 
     config = PYSCRIPT_SCHEMA(conf.get(DOMAIN, {}))
 
@@ -128,7 +128,7 @@ async def update_yaml_config(hass: HomeAssistant, config_entry: ConfigEntry) ->
     return False
 
 
-def start_global_contexts(global_ctx_only: str = None) -> None:
+def start_global_contexts(global_ctx_only: str | None = None) -> None:
     """Start all the file and apps global contexts."""
     start_list = []
     for global_ctx_name, global_ctx in GlobalContextMgr.items():
@@ -145,7 +145,9 @@ def start_global_contexts(global_ctx_only: str = None) -> None:
 
 
 async def watchdog_start(
-    hass: HomeAssistant, pyscript_folder: str, reload_scripts_handler: Callable[[None], None]
+    hass: HomeAssistant,
+    pyscript_folder: str,
+    reload_scripts_handler: Callable[[ServiceCall], Awaitable[None]],
 ) -> None:
     """Start watchdog thread to look for changed files in pyscript_folder."""
     if WATCHDOG_TASK in hass.data[DOMAIN]:
@@ -201,7 +203,7 @@ def on_deleted(self, event: FileSystemEvent) -> None:
             self.process(event)
 
     async def task_watchdog(watchdog_q: asyncio.Queue) -> None:
-        def check_event(event, do_reload: bool) -> bool:
+        def check_event(event: FileSystemEvent, do_reload: bool) -> bool:
             """Check if event should trigger a reload."""
             if event.is_directory:
                 # don't reload if it's just a directory modified
@@ -230,7 +232,7 @@ def check_event(event, do_reload: bool) -> bool:
                         do_reload = check_event(
                             await asyncio.wait_for(watchdog_q.get(), timeout=0.05), do_reload
                         )
-                    except asyncio.TimeoutError:
+                    except TimeoutError:
                         break
                 if do_reload:
                     await reload_scripts_handler(None)
@@ -304,14 +306,14 @@ async def reload_scripts_handler(call: ServiceCall) -> None:
 
     hass.services.async_register(DOMAIN, SERVICE_RELOAD, reload_scripts_handler)
 
-    async def generate_stubs_service(call: ServiceCall) -> Dict[str, Any]:
+    async def generate_stubs_service(call: ServiceCall) -> dict[str, Any]:
         """Generate pyscript IDE stub files."""
 
         generator = StubsGenerator(hass)
         generated_body = await generator.build()
         stubs_path = os.path.join(hass.config.path(FOLDER), "modules", "stubs")
 
-        def write_stubs(path) -> dict[str, Any]:
+        def write_stubs(path: str) -> dict[str, Any]:
             res: dict[str, Any] = {}
             try:
                 os.makedirs(path, exist_ok=True)
@@ -342,7 +344,7 @@ def write_stubs(path) -> dict[str, Any]:
         return result
 
     hass.services.async_register(
-        DOMAIN, SERVICE_GENERATE_STUBS, generate_stubs_service, supports_response=SERVICE_RESPONSE_ONLY
+        DOMAIN, SERVICE_GENERATE_STUBS, generate_stubs_service, supports_response=SupportsResponse.ONLY
     )
 
     async def jupyter_kernel_start(call: ServiceCall) -> None:
@@ -443,7 +445,7 @@ async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) ->
     return True
 
 
-async def unload_scripts(global_ctx_only: str = None, unload_all: bool = False) -> None:
+async def unload_scripts(global_ctx_only: str | None = None, unload_all: bool = False) -> None:
     """Unload all scripts from GlobalContextMgr with given name prefixes."""
     ctx_delete = {}
     for global_ctx_name, global_ctx in GlobalContextMgr.items():
@@ -462,7 +464,9 @@ async def unload_scripts(global_ctx_only: str = None, unload_all: bool = False)
 
 
 @bind_hass
-async def load_scripts(hass: HomeAssistant, config_data: Dict[str, Any], global_ctx_only: str = None):
+async def load_scripts(
+    hass: HomeAssistant, config_data: dict[str, Any], global_ctx_only: str | None = None
+) -> None:
     """Load all python scripts in FOLDER."""
 
     class SourceFile:
@@ -496,8 +500,8 @@ def __init__(
     pyscript_dir = hass.config.path(FOLDER)
 
     def glob_read_files(
-        load_paths: List[Set[Union[str, bool]]], apps_config: Dict[str, Any]
-    ) -> Dict[str, SourceFile]:
+        load_paths: list[tuple[str, str, bool, bool]], apps_config: dict[str, Any]
+    ) -> dict[str, SourceFile]:
         """Expand globs and read all the source files."""
         ctx2source = {}
         for path, match, check_config, autoload in load_paths:

custom_components/pyscript/const.py

@@ -1,20 +1,5 @@
 """Define pyscript-wide constants."""
 
-#
-# 2023.7 supports service response; handle older versions by defaulting enum
-# Should eventually deprecate this and just use SupportsResponse import
-#
-try:
-    from homeassistant.core import SupportsResponse
-
-    SERVICE_RESPONSE_NONE = SupportsResponse.NONE
-    SERVICE_RESPONSE_OPTIONAL = SupportsResponse.OPTIONAL
-    SERVICE_RESPONSE_ONLY = SupportsResponse.ONLY
-except ImportError:
-    SERVICE_RESPONSE_NONE = None
-    SERVICE_RESPONSE_OPTIONAL = None
-    SERVICE_RESPONSE_ONLY = None
-
 DOMAIN = "pyscript"
 
 CONFIG_ENTRY = "config_entry"

custom_components/pyscript/eval.py

@@ -13,11 +13,13 @@
 import sys
 import time
 import traceback
+from typing import TYPE_CHECKING, Any
 import weakref
 
 import yaml
 
 from homeassistant.const import SERVICE_RELOAD
+from homeassistant.core import SupportsResponse
 from homeassistant.helpers.service import async_set_service_schema
 
 from .const import (
@@ -27,11 +29,15 @@
     DOMAIN,
     LOGGER_PATH,
     SERVICE_JUPYTER_KERNEL_START,
-    SERVICE_RESPONSE_NONE,
 )
 from .function import Function
 from .state import State
 
+if TYPE_CHECKING:
+    from .global_ctx import GlobalContext
+
+type SymTable = dict[str, Any]
+
 _LOGGER = logging.getLogger(LOGGER_PATH + ".eval")
 
 #
@@ -315,7 +321,14 @@ def getattr(self):
 class EvalFunc:
     """Class for a callable pyscript function."""
 
-    def __init__(self, func_def, code_list, code_str, global_ctx, async_func=False):
+    def __init__(
+        self,
+        func_def: ast.FunctionDef,
+        code_list: list[str],
+        code_str: str,
+        global_ctx: "GlobalContext",
+        async_func: bool = False,
+    ) -> None:
         """Initialize a function calling context."""
         self.func_def = func_def
         self.name = func_def.name
@@ -533,7 +546,7 @@ async def do_service_call(func, ast_ctx, data):
                         domain,
                         name,
                         pyscript_service_factory(func_name, self),
-                        dec_kwargs.get("supports_response", SERVICE_RESPONSE_NONE),
+                        dec_kwargs.get("supports_response", SupportsResponse.NONE),
                     )
                     async_set_service_schema(Function.hass, domain, name, service_desc)
                     self.trigger_service.add(srv_name)
@@ -838,12 +851,12 @@ async def check_for_closure(self, arg):
 class EvalFuncVar:
     """Class for a callable pyscript function."""
 
-    def __init__(self, func):
+    def __init__(self, func: EvalFunc) -> None:
         """Initialize instance with given EvalFunc function."""
         self.func = func
-        self.ast_ctx = None
+        self.ast_ctx: AstEval | None = None
 
-    def get_func(self):
+    def get_func(self) -> EvalFunc:
         """Return the EvalFunc function."""
         return self.func
 
@@ -895,7 +908,7 @@ async def __call__(self, *args, **kwargs):
 class EvalFuncVarClassInst(EvalFuncVar):
     """Class for a callable pyscript class instance function."""
 
-    def __init__(self, func, ast_ctx, class_inst_weak):
+    def __init__(self, func: EvalFunc, ast_ctx: "AstEval", class_inst_weak: weakref.ReferenceType) -> None:
         """Initialize instance with given EvalFunc function."""
         super().__init__(func)
         self.ast_ctx = ast_ctx
@@ -913,25 +926,25 @@ async def __call__(self, *args, **kwargs):
 class AstEval:
     """Python interpreter AST object evaluator."""
 
-    def __init__(self, name, global_ctx, logger_name=None):
+    def __init__(self, name: str, global_ctx: "GlobalContext", logger_name: str | None = None) -> None:
         """Initialize an interpreter execution context."""
         self.name = name
         self.str = None
         self.ast = None
         self.global_ctx = global_ctx
-        self.global_sym_table = global_ctx.get_global_sym_table() if global_ctx else {}
-        self.sym_table_stack = []
+        self.global_sym_table: SymTable = global_ctx.get_global_sym_table() if global_ctx else {}
+        self.sym_table_stack: list[SymTable] = []
         self.sym_table = self.global_sym_table
-        self.local_sym_table = {}
-        self.user_locals = {}
-        self.curr_func = None
+        self.local_sym_table: SymTable = {}
+        self.user_locals: SymTable = {}
+        self.curr_func: EvalFunc | None = None
         self.filename = name
-        self.code_str = None
-        self.code_list = None
-        self.exception = None
-        self.exception_obj = None
-        self.exception_long = None
-        self.exception_curr = None
+        self.code_str: str | None = None
+        self.code_list: list[str] | None = None
+        self.exception: str | None = None
+        self.exception_obj: Exception | None = None
+        self.exception_long: str | None = None
+        self.exception_curr: Exception | None = None
         self.lineno = 1
         self.col_offset = 0
         self.logger_handlers = set()
@@ -2159,7 +2172,7 @@ async def get_names(self, this_ast=None, nonlocal_names=None, global_names=None,
             await self.get_names_set(this_ast, names, nonlocal_names, global_names, local_names)
         return names
 
-    def parse(self, code_str, filename=None, mode="exec"):
+    def parse(self, code_str: str, filename: str | None = None, mode: str = "exec") -> bool:
         """Parse the code_str source code into an AST tree."""
         self.exception = None
         self.exception_obj = None
@@ -2298,7 +2311,7 @@ def completions(self, root):
                     for attr in var.__dict__:
                         if attr.lower().startswith(attr_root) and (attr_root != "" or attr[0:1] != "_"):
                             words.add(f"{name}.{attr}")
-                except Exception:
+                except Exception:  # noqa: S110
                     pass
         for keyw in set(keyword.kwlist) - {"yield"}:
             if keyw.lower().startswith(root):
@@ -2313,7 +2326,7 @@ def completions(self, root):
                 words.add(name)
         return words
 
-    async def eval(self, new_state_vars=None, merge_local=False):
+    async def eval(self, new_state_vars: dict[str, Any] | None = None, merge_local: bool = False) -> None:
         """Execute parsed code, with the optional state variables added to the scope."""
         self.exception = None
         self.exception_obj = None