fix ruff warnings

Author: dmamelin

custom_components/pyscript/decorator.py

@@ -6,7 +6,7 @@
 import asyncio
 import logging
 import os
-from typing import Any, ClassVar, Type, TypeVar
+from typing import Any, ClassVar, TypeVar
 import weakref
 
 from homeassistant.config_entries import ConfigEntry
@@ -32,7 +32,7 @@
 class DecoratorRegistry:
     """Decorator registry."""
 
-    _decorators: dict[str, Type[Decorator]]  # decorator name to class
+    _decorators: dict[str, type[Decorator]]  # decorator name to class
     hass: ClassVar[HomeAssistant]
     prefix: ClassVar[str] = "e"
 
@@ -60,13 +60,13 @@ def init(cls, hass: HomeAssistant, config_entry: ConfigEntry = None) -> None:
 
         Function.register_ast({cls.prefix + "task.wait_until": DecoratorRegistry.wait_until_factory})
 
-        from .decorators import DECORATORS  # noqa: PLC0415 pylint: disable=import-outside-toplevel
+        from .decorators import DECORATORS  # pylint: disable=import-outside-toplevel
 
         for dec_type in DECORATORS:
             cls.register(dec_type)
 
     @classmethod
-    def register(cls, dec_type: Type[Decorator]):
+    def register(cls, dec_type: type[Decorator]) -> None:
         """Register a decorator."""
         if not dec_type.name:
             raise TypeError(f"Decorator name is required {dec_type}")
@@ -103,7 +103,7 @@ async def get_decorator_by_expr(cls, ast_ctx: AstEval, dec_expr: ast.expr) -> De
         return None
 
     @classmethod
-    async def wait_until(cls, ast_ctx: AstEval, *arg, **kwargs):
+    async def wait_until(cls, ast_ctx: AstEval, *_arg: Any, **kwargs: Any) -> Any:
         """Build a temporary decorator manager that waits until one of trigger decorators fires."""
         func_args = set(kwargs.keys())
         if len(func_args) == 0:

custom_components/pyscript/decorator_abc.py

@@ -46,7 +46,7 @@ class DispatchData:
     call_ast_ctx: AstEval | None = field(default=None, kw_only=True)
     hass_context: Context | None = field(default=None, kw_only=True)
 
-    # Normally shouldn’t be used.
+    # Normally shouldn't be used.
     exception: Exception | None = field(default=None, kw_only=True)
     exception_text: str | None = field(default=None, kw_only=True)
 
@@ -90,7 +90,8 @@ async def validate(self) -> None:
             self.kwargs = self.kwargs_schema(self.raw_kwargs)
 
         except vol.Invalid as err:
-            # FIXME For test compatibility. Update the message in the future.
+            # Keep this wording for transition compatibility. Once the legacy
+            # subsystem is removed, update the message and related tests.
             if len(err.path) == 1:
                 if "extra keys not allowed" in err.msg:
                     message = f"invalid keyword argument '{err.path[0]}'"
@@ -105,9 +106,11 @@ async def validate(self) -> None:
             )
             raise type_error from err
 
+    @abstractmethod
     async def start(self):
         """Start the decorator."""
 
+    @abstractmethod
     async def stop(self):
         """Stop the decorator."""
 
@@ -156,7 +159,7 @@ def add(self, decorator: Decorator) -> None:
         self._decorators.append(decorator)
         decorator.dm = self
 
-    def get_decorators[DT](self, decorator_type: type[DT] | None = None) -> list[DT]:  # noqa: D102
+    def get_decorators[DT](self, decorator_type: type[DT] | None = None) -> list[DT]:
         """Get decorators of a specific type."""
         if decorator_type is None:
             return self._decorators.copy()
@@ -242,7 +245,7 @@ def __init_subclass__(cls, **kwargs):
                 {vol.Optional("kwargs"): vol.Coerce(dict[str, Any], msg="should be type dict")}
             )
 
-    async def dispatch(self, data: DispatchData):
+    async def dispatch(self, data: DispatchData) -> None:
         """Dispatch a trigger call to the function."""
         if not data.trigger:
             data.trigger = self
@@ -260,7 +263,8 @@ async def validate(self) -> None:
         await super().validate()
         decorators = self.dm.get_decorators(TriggerDecorator)
         if len(decorators) == 0:
-            # FIXME For test compatibility. Update the message in the future.
+            # Keep this wording for transition compatibility. Once the legacy
+            # subsystem is removed, update the message and related tests.
             trig_decorators_reqd = {
                 "event_trigger",
                 "mqtt_trigger",

custom_components/pyscript/decorators/service.py

@@ -54,10 +54,12 @@ async def validate(self) -> None:
 
         if len(self.args) != 2:
             self.args = [DOMAIN, self.dm.func_name]
-        # FIXME This condition doesn't verify the domain - it may not be Pyscript.
-        #       The error is kept for backward compatibility.
+        # This condition still does not verify the domain. Keep the behavior
+        # for transition compatibility and revisit it after the legacy
+        # subsystem is removed.
         if self.args[1] in (SERVICE_RELOAD, SERVICE_JUPYTER_KERNEL_START):
-            # FIXME For test compatibility. Update the message in the future.
+            # Keep this wording for transition compatibility. Once the legacy
+            # subsystem is removed, update the message and related tests.
             raise SyntaxError(
                 f"function '{self.dm.func_name}' defined in {self.dm.ast_ctx.get_global_ctx_name()}: "
                 f"@service conflicts with builtin service"
@@ -72,7 +74,7 @@ async def validate(self) -> None:
             try:
                 desc = desc[4:].lstrip(" \n\r")
                 file_desc = io.StringIO(desc)
-                self.description = yaml.load(file_desc, Loader=yaml.BaseLoader) or OrderedDict()
+                self.description = yaml.safe_load(file_desc) or OrderedDict()
                 file_desc.close()
             except Exception as exc:
                 self.dm.logger.error(

custom_components/pyscript/decorators/state.py

@@ -28,7 +28,7 @@ class StateActiveDecorator(TriggerHandlerDecorator, ExpressionDecorator):
         vol.All(
             vol.Length(
                 min=1, max=1, msg="got 2 arguments, expected 1"
-            ),  # FIXME For test compatibility. Update the message in the future.
+            ),  # Keep this wording for transition compatibility.
             vol.All([str]),
         )
     )
@@ -149,7 +149,7 @@ def _diff(self, dt: float, now: float) -> str:
             return "None"
         return f"{(now - dt):g} ago"
 
-    async def _check_new_state(self, trig_ok: bool):
+    async def _check_new_state(self, trig_ok: bool) -> None:
         now = asyncio.get_running_loop().time()
         if _LOGGER.isEnabledFor(logging.DEBUG):
             msg = f"check_new_state: {self}"
@@ -206,7 +206,7 @@ async def _check_new_state(self, trig_ok: bool):
                     _LOGGER.debug("state_hold_false started, %s", self)
                     self.false_entered_at = now
 
-    async def _check_state_hold(self):
+    async def _check_state_hold(self) -> None:
         if self.true_entered_at is None:
             raise RuntimeError(f"state_hold not started for {self}")
 
@@ -218,7 +218,7 @@ async def _check_state_hold(self):
                 DispatchData(self.last_func_args, trigger_context={"new_vars": self.last_new_vars})
             )
 
-    async def _cycle(self):
+    async def _cycle(self) -> None:
         """Run the trigger cycle with state_hold and state_hold_false logic."""
         loop = asyncio.get_running_loop()
 
@@ -283,7 +283,7 @@ async def _cycle(self):
                 else:
                     trig_ok = False
                 await self._check_new_state(trig_ok)
-            except asyncio.TimeoutError:
+            except TimeoutError:
                 await self._check_state_hold()
 
     async def _is_trig_ok(self) -> bool:
@@ -296,7 +296,7 @@ def _on_task_done(self, task: asyncio.Task) -> None:
             return
         exc = task.exception()
         if exc is not None:
-            self.dm.logger.exception(f"{self} failed", exc_info=exc)
+            self.dm.logger.error("%s failed", self, exc_info=exc)
 
     async def start(self) -> None:
         """Start the trigger."""

custom_components/pyscript/decorators/timing.py

@@ -21,7 +21,7 @@
 
 def dt_now():
     """Return current time."""
-    # FIXME For test compatibility. The tests patch this function
+    # Keep this wrapper so tests can patch it during the transition.
     return trigger.dt_now()
 
 
@@ -70,7 +70,7 @@ class TimeTriggerDecorator(TriggerDecorator):
             vol.Length(min=0),
             vol.All(
                 [str], msg="argument 2 should be a string"
-            ),  # FIXME For test compatibility. Update the message in the future.
+            ),  # Keep this wording for transition compatibility.
         )
     )
 
@@ -154,7 +154,7 @@ def _on_task_done(self, task: asyncio.Task) -> None:
             return
         exc = task.exception()
         if exc is not None:
-            self.dm.logger.exception(f"{self} failed", exc_info=exc)
+            self.dm.logger.error("%s failed", self, exc_info=exc)
 
     async def start(self) -> None:
         """Start the decorator."""

tests/test_init.py

@@ -15,7 +15,7 @@
 from custom_components.pyscript.global_ctx import GlobalContextMgr
 from custom_components.pyscript.state import State
 from homeassistant import loader
-from homeassistant.const import EVENT_HOMEASSISTANT_STARTED, EVENT_STATE_CHANGED
+from homeassistant.const import EVENT_STATE_CHANGED
 from homeassistant.core import Context
 from homeassistant.helpers.service import async_get_all_descriptions
 from homeassistant.setup import async_setup_component