fix linter warnings

Author: dmamelin

custom_components/pyscript/decorator.py

@@ -1,24 +1,26 @@
+"""Decorator registry and manager logic for pyscript decorators."""
+
 from __future__ import annotations
 
 import ast
 import asyncio
 import logging
 import os
+from typing import Any, ClassVar, Type, TypeVar
 import weakref
-from typing import Type, ClassVar, Any, TypeVar
 
 from homeassistant.config_entries import ConfigEntry
-from homeassistant.core import HomeAssistant, Context
+from homeassistant.core import Context, HomeAssistant
 
 from .decorator_abc import (
+    CallHandlerDecorator,
+    CallResultHandlerDecorator,
     Decorator,
     DecoratorManager,
-    DispatchData,
     DecoratorManagerStatus,
-    TriggerHandlerDecorator,
-    CallHandlerDecorator,
+    DispatchData,
     TriggerDecorator,
-    CallResultHandlerDecorator,
+    TriggerHandlerDecorator,
 )
 from .eval import AstEval, EvalFunc, EvalFuncVar
 from .function import Function
@@ -49,15 +51,16 @@ def init(cls, hass: HomeAssistant, config_entry: ConfigEntry = None) -> None:
             cls.prefix = ""
             space = "\n" + " " * 35
             border = space + "=" * 35
-            _LOGGER.warning(border + space + "DecoratorManager enabled by default" + border)
+            msg = border + space + "DecoratorManager enabled by default" + border
+            _LOGGER.warning(msg)
         else:
             cls.prefix = "e"
 
         DecoratorManager.hass = hass
 
         Function.register_ast({cls.prefix + "task.wait_until": DecoratorRegistry.wait_until_factory})
 
-        from .decorators import DECORATORS
+        from .decorators import DECORATORS  # noqa: PLC0415 pylint: disable=import-outside-toplevel
 
         for dec_type in DECORATORS:
             cls.register(dec_type)
@@ -171,6 +174,7 @@ class WaitUntilDecoratorManager(DecoratorManager):
     """Decorator manager for task.wait_until."""
 
     def __init__(self, ast_ctx: AstEval, **kwargs: dict[str, Any]) -> None:
+        """Initialize the task.wait_until decorator manager."""
         super().__init__(ast_ctx, ast_ctx.name)
         self.kwargs = kwargs
         self._future: asyncio.Future[DispatchData] = self.hass.loop.create_future()
@@ -212,6 +216,7 @@ class FunctionDecoratorManager(DecoratorManager):
     """Maintain and validate a set of decorators applied to a function."""
 
     def __init__(self, ast_ctx: AstEval, eval_func_var: EvalFuncVar) -> None:
+        """Initialize the function decorator manager."""
         super().__init__(ast_ctx, f"{ast_ctx.get_global_ctx_name()}.{eval_func_var.get_name()}")
         self.eval_func: EvalFunc = eval_func_var.func
 

custom_components/pyscript/decorator_abc.py

@@ -1,12 +1,15 @@
+"""Base abstractions for pyscript decorators and decorator managers."""
+
 from __future__ import annotations
 
-import logging
 from abc import ABC, abstractmethod
-from dataclasses import field, dataclass
+from dataclasses import dataclass, field
 from enum import StrEnum
-from typing import ClassVar, Any, TypeVar, Type, final
+import logging
+from typing import Any, ClassVar, final
 
 import voluptuous as vol
+
 from homeassistant.core import Context, HomeAssistant
 
 from . import trigger
@@ -109,6 +112,7 @@ async def stop(self):
         """Stop the decorator."""
 
     def __repr__(self):
+        """Represent the decorator as a string with the decorator name and arguments."""
         parts = []
         if self.raw_args is not None:
             parts.append(",".join(map(str, self.raw_args)))
@@ -117,15 +121,13 @@ def __repr__(self):
         return f"@{self.name}({', '.join(parts)})"
 
 
-DecoratorType = TypeVar("DecoratorType", bound=Decorator)
-
-
 class DecoratorManager(ABC):
-    """Maintain and validate a set of decorators"""
+    """Maintain and validate a set of decorators."""
 
     hass: ClassVar[HomeAssistant]
 
     def __init__(self, ast_ctx: AstEval, name: str) -> None:
+        """Initialize the manager."""
         self.ast_ctx = ast_ctx
         self.name = name
         self.func_name = name.split(".")[-1]
@@ -154,7 +156,7 @@ def add(self, decorator: Decorator) -> None:
         self._decorators.append(decorator)
         decorator.dm = self
 
-    def get_decorators(self, decorator_type: Type[DecoratorType] | None = None) -> list[DecoratorType]:
+    def get_decorators[DT](self, decorator_type: type[DT] | None = None) -> list[DT]:  # noqa: D102
         """Get decorators of a specific type."""
         if decorator_type is None:
             return self._decorators.copy()
@@ -221,16 +223,18 @@ async def stop(self):
 
     @abstractmethod
     async def dispatch(self, data: DispatchData) -> None:
-        pass
+        """Dispatch a trigger call."""
 
     def __repr__(self):
+        """Return a string representation of the manager with status and decorators."""
         return f"{self.__class__.__name__}({self.status}) {self._decorators} for {self.name}()>"
 
 
 class TriggerDecorator(Decorator, ABC):
     """Base class for trigger-based decorators."""
 
     def __init_subclass__(cls, **kwargs):
+        """Initialize the decorator class."""
         super().__init_subclass__(**kwargs)
         # kwargs for all triggers
         if "kwargs" not in cls.kwargs_schema.schema.keys():
@@ -280,7 +284,6 @@ class CallHandlerDecorator(Decorator, ABC):
     @abstractmethod
     async def handle_call(self, data: DispatchData) -> bool | None:
         """Handle an action call. Return False for stop calling."""
-        pass
 
 
 class CallResultHandlerDecorator(Decorator, ABC):
@@ -289,4 +292,3 @@ class CallResultHandlerDecorator(Decorator, ABC):
     @abstractmethod
     async def handle_call_result(self, data: DispatchData, result: Any) -> None:
         """Handle an action call result."""
-        pass

custom_components/pyscript/decorators/__init__.py

@@ -1,9 +1,11 @@
+"""Pyscript decorators."""
+
 from .event import EventTriggerDecorator
 from .mqtt import MQTTTriggerDecorator
 from .service import ServiceDecorator
-from .state import StateTriggerDecorator, StateActiveDecorator
+from .state import StateActiveDecorator, StateTriggerDecorator
 from .task import TaskUniqueDecorator
-from .timing import TimeTriggerDecorator, TimeActiveDecorator
+from .timing import TimeActiveDecorator, TimeTriggerDecorator
 from .webhook import WebhookTriggerDecorator
 
 DECORATORS = [

custom_components/pyscript/decorators/base.py

@@ -1,7 +1,7 @@
 """Base mixins for pyscript decorators."""
 
-import logging
 from abc import ABC
+import logging
 from typing import Any
 
 import voluptuous as vol

custom_components/pyscript/decorators/event.py

@@ -1,10 +1,13 @@
+"""Event decorator."""
+
 import logging
 
 import voluptuous as vol
-from homeassistant.core import Event, CALLBACK_TYPE
 
-from .base import ExpressionDecorator
+from homeassistant.core import CALLBACK_TYPE, Event
+
 from ..decorator_abc import DispatchData, TriggerDecorator
+from .base import ExpressionDecorator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -29,7 +32,6 @@ async def validate(self) -> None:
             self.create_expression(self.args[1])
 
     async def _event_callback(self, event: Event) -> None:
-        """Callback for the event trigger."""
         _LOGGER.debug("Event trigger received: %s %s", type(event), event)
         func_args = {
             "trigger_type": "event",

custom_components/pyscript/decorators/mqtt.py

@@ -1,16 +1,17 @@
-"""Trigger decorator implementations."""
+"""Mqtt decorator."""
 
 from __future__ import annotations
 
 import json
 import logging
 
 import voluptuous as vol
+
 from homeassistant.components import mqtt
 from homeassistant.core import CALLBACK_TYPE
 
-from .base import ExpressionDecorator, AutoKwargsDecorator
 from ..decorator_abc import DispatchData, TriggerDecorator
+from .base import AutoKwargsDecorator, ExpressionDecorator
 
 _LOGGER = logging.getLogger(__name__)
 

custom_components/pyscript/decorators/service.py

@@ -3,25 +3,27 @@
 from __future__ import annotations
 
 import ast
+from collections import OrderedDict
 import io
 import logging
 import typing
-from collections import OrderedDict
 
 import voluptuous as vol
 import yaml
+
 from homeassistant.const import SERVICE_RELOAD
 from homeassistant.core import ServiceCall, SupportsResponse
 from homeassistant.helpers.service import async_set_service_schema
-from ..decorator import FunctionDecoratorManager
 
-from ..decorator_abc import Decorator
 from .. import DOMAIN, SERVICE_JUPYTER_KERNEL_START, AstEval, Function, State
+from ..decorator import FunctionDecoratorManager
+from ..decorator_abc import Decorator
 
 _LOGGER = logging.getLogger(__name__)
 
 
 def service_validator(args: list[str]) -> list[str]:
+    """Validate and normalize service name."""
     if len(args) == 0:
         return []
     s = str(args[0]).strip()
@@ -47,11 +49,12 @@ class ServiceDecorator(Decorator):
     description: dict
 
     async def validate(self) -> None:
+        """Validate the arguments."""
         await super().validate()
 
         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.
+        # FIXME This condition doesn't verify the domain - it may not be Pyscript.
         #       The error is kept for backward compatibility.
         if self.args[1] in (SERVICE_RELOAD, SERVICE_JUPYTER_KERNEL_START):
             # FIXME For test compatibility. Update the message in the future.
@@ -81,7 +84,6 @@ async def validate(self) -> None:
         else:
             fields = OrderedDict()
             for arg in ast_funcdef.args.posonlyargs + ast_funcdef.args.args:
-                # _LOGGER.warning(f"------ {type(arg.arg)} {arg.arg}")
                 fields[arg.arg] = OrderedDict(description=f"argument {arg.arg}")
             self.description = {"description": desc, "fields": fields}
 
@@ -99,7 +101,6 @@ async def _service_callback(self, call: ServiceCall) -> None:
         }
         func_args.update(call.data)
 
-        #
         async def do_service_call(func, ast_ctx, data):
             try:
                 _LOGGER.debug("Service call start: %s", func.name)
@@ -112,7 +113,6 @@ async def do_service_call(func, ast_ctx, data):
                 _LOGGER.exception(exc)
                 return None
 
-        #
         task = Function.create_task(do_service_call(self.dm.eval_func, ast_ctx, func_args))
         await task
         return task.result()

custom_components/pyscript/decorators/state.py

@@ -1,13 +1,19 @@
+"""State decorators."""
+
 import asyncio
+import logging
 import re
+from typing import Any
+
+import voluptuous as vol
 
 from homeassistant.helpers import config_validation as cv
 
-from .base import ExpressionDecorator, AutoKwargsDecorator
 from ..decorator import WaitUntilDecoratorManager
-from ..decorator_abc import *
+from ..decorator_abc import DecoratorManagerStatus, DispatchData, TriggerDecorator, TriggerHandlerDecorator
 from ..state import State
 from ..trigger import ident_any_values_changed, ident_values_changed
+from .base import AutoKwargsDecorator, ExpressionDecorator
 
 STATE_RE = re.compile(r"\w+\.\w+(\.((\w+)|\*))?$")
 
@@ -36,6 +42,7 @@ async def validate(self) -> None:
         self.var_names = await self._ast_expression.get_names()
 
     async def handle_dispatch(self, data: DispatchData) -> bool:
+        """Handle dispatch events."""
         new_vars = data.trigger_context.get("new_vars", {})
         active_vars = State.notify_var_get(self.var_names, new_vars)
         return await self.check_expression_vars(active_vars)
@@ -91,7 +98,11 @@ class StateTriggerDecorator(TriggerDecorator, ExpressionDecorator, AutoKwargsDec
     true_entered_at: float | None
     false_entered_at: float | None
 
+    last_func_args: dict[str, Any]
+    last_new_vars: dict[str, Any]
+
     async def validate(self) -> None:
+        """Validate and normalize arguments."""
         await super().validate()
         self.state_trig_ident = set()
         self.state_trig_ident_any = set()
@@ -136,18 +147,19 @@ async def validate(self) -> None:
     def _diff(self, dt: float, now: float) -> str:
         if dt is None:
             return "None"
-        else:
-            return f"{(now-dt):g} ago"
+        return f"{(now - dt):g} ago"
 
     async def _check_new_state(self, trig_ok: bool):
         now = asyncio.get_running_loop().time()
         if _LOGGER.isEnabledFor(logging.DEBUG):
             msg = f"check_new_state: {self}"
             msg += f"\ntrig_ok: {trig_ok} now {now} func_args: {self.last_func_args} new_vars: {self.last_new_vars}"
             if self.true_entered_at:
-                msg += f"\ntrue_entered_at: {self.true_entered_at}({(now-self.true_entered_at):g} ago)\n"
+                msg += f"\ntrue_entered_at: {self.true_entered_at}({(now - self.true_entered_at):g} ago)\n"
             if self.false_entered_at:
-                msg += f"\nfalse_entered_at: {self.false_entered_at}({(now-self.false_entered_at):g} ago)\n"
+                msg += (
+                    f"\nfalse_entered_at: {self.false_entered_at}({(now - self.false_entered_at):g} ago)\n"
+                )
             _LOGGER.debug(msg)
 
         state_hold_false_passed = False
@@ -277,8 +289,7 @@ async def _cycle(self):
     async def _is_trig_ok(self) -> bool:
         if self.has_expression():
             return await self.check_expression_vars(self.last_new_vars)
-        else:
-            return True
+        return True
 
     def _on_task_done(self, task: asyncio.Task) -> None:
         if task.cancelled():

custom_components/pyscript/decorators/task.py

@@ -5,11 +5,12 @@
 import logging
 
 import voluptuous as vol
+
 from homeassistant.helpers import config_validation as cv
 
-from .base import AutoKwargsDecorator
-from ..decorator_abc import DispatchData, CallHandlerDecorator
+from ..decorator_abc import CallHandlerDecorator, DispatchData
 from ..function import Function
+from .base import AutoKwargsDecorator
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -24,6 +25,7 @@ class TaskUniqueDecorator(CallHandlerDecorator, AutoKwargsDecorator):
     kill_me: bool
 
     async def handle_call(self, data: DispatchData) -> bool:
+        """Handle call."""
         if self.kill_me:
             if Function.unique_name_used(data.call_ast_ctx, self.args[0]):
                 _LOGGER.debug(