fixed some wrong type hints

run/conf_testing/lib/HABAppTests/event_waiter.py

@@ -5,7 +5,6 @@
 
 from HABApp.core.events.filter import EventFilter
 from HABApp.core.internals import (
- HINT_EVENT_FILTER_OBJ,
  EventBusListener,
  EventFilterBase,
  get_current_context,
@@ -24,7 +23,7 @@
 
 class EventWaiter:
  def __init__(self, name: BaseValueItem | str,
- event_filter: HINT_EVENT_FILTER_OBJ, timeout=1) -> None:
+ event_filter: EventFilterBase, timeout: float = 1) -> None:
  if isinstance(name, BaseValueItem):
  name = name.name
  assert isinstance(name, str)

run/conf_testing/rules/openhab/test_item_funcs.py

@@ -133,10 +133,12 @@ def test_func(self, item_type, func_name, test_vals) -> None:
  getattr(tmpitem, func_name)(val)
  waiter.wait_for_state(val)
 
- for val in test_vals:
- tmpitem.set_value(val)
- getattr(tmpitem, func_name)()
- waiter.wait_for_state(val)
+ # only the openhab functions can be called without a value
+ if func_name.startswith('oh_'):
+ for val in test_vals:
+ tmpitem.set_value(val)
+ getattr(tmpitem, func_name)()
+ waiter.wait_for_state(val)
 
  @OpenhabTmpItem.create('Number', arg_name='oh_item')
  def test_post_update_if(self, oh_item: OpenhabTmpItem) -> None:

src/HABApp/core/events/filter/groups.py

@@ -1,11 +1,11 @@
 from typing import Any
 
-from HABApp.core.internals import HINT_EVENT_FILTER_OBJ, EventFilterBase
+from HABApp.core.internals import EventFilterBase
 
 
 class EventFilterBaseGroup(EventFilterBase):
- def __init__(self, *args: HINT_EVENT_FILTER_OBJ) -> None:
- self.filters: tuple[HINT_EVENT_FILTER_OBJ, ...] = args
+ def __init__(self, *args: EventFilterBase) -> None:
+ self.filters: tuple[EventFilterBase, ...] = args
 
  def trigger(self, event) -> bool:
  raise NotImplementedError()

src/HABApp/core/internals/__init__.py

@@ -1,14 +1,15 @@
-from .proxy import uses_get_item, uses_item_registry, uses_post_event, uses_event_bus, setup_internals
-from .context import ContextProvidingObj, Context, ContextBoundObj, get_current_context, Context, AutoContextBoundObj
+from .context import AutoContextBoundObj, Context, ContextBoundObj, ContextProvidingObj, get_current_context
+from .proxy import setup_internals, uses_event_bus, uses_get_item, uses_item_registry, uses_post_event
+
 
 # isort: split
 
-from .event_filter import EventFilterBase, HINT_EVENT_FILTER_OBJ
 from .event_bus import EventBus
+from .event_filter import EventFilterBase
 from .item_registry import ItemRegistry, ItemRegistryItem
 
+
 # isort: split
 
-from .event_bus_listener import HINT_EVENT_BUS_LISTENER, EventBusListener, ContextBoundEventBusListener
-from .event_filter import EventFilterBase, HINT_EVENT_FILTER_OBJ
+from .event_bus_listener import ContextBoundEventBusListener, EventBusListener
 from .wrapped_function import WrappedFunctionBase, wrap_func

src/HABApp/core/internals/context/get_context.py

@@ -1,13 +1,14 @@
 # noinspection PyProtectedMember
 from sys import _getframe as sys_get_frame
-from types import FrameType
-from typing import TYPE_CHECKING, Optional
+from typing import TYPE_CHECKING, Any, Optional
 
 from HABApp.core.errors import ContextNotFoundError, ContextNotSetError
 from HABApp.core.internals.context import Context, ContextBoundObj, ContextProvidingObj
 
 
 if TYPE_CHECKING:
+ from types import FrameType
+
  import HABApp
 
 
@@ -33,7 +34,7 @@ def get_current_context(obj: ContextProvidingObj | None = None) -> 'HABApp.rule_
 
 
 class AutoContextBoundObj(ContextBoundObj):
- def __init__(self, parent_ctx: Optional['Context'] = None, **kwargs) -> None:
+ def __init__(self, parent_ctx: Optional['Context'] = None, **kwargs: Any) -> None:
  if parent_ctx is None:
  parent_ctx = get_current_context()
  super().__init__(parent_ctx=parent_ctx, **kwargs)

src/HABApp/core/internals/event_bus_listener.py

@@ -1,6 +1,8 @@
-from typing import TypeVar
+from typing import Any, TypeVar
 
-from HABApp.core.internals import HINT_EVENT_FILTER_OBJ, AutoContextBoundObj, Context, uses_event_bus
+from typing_extensions import override
+
+from HABApp.core.internals import AutoContextBoundObj, EventFilterBase, uses_event_bus
 from HABApp.core.internals.event_bus import EventBusBaseListener
 from HABApp.core.internals.wrapped_function import WrappedFunctionBase
 
@@ -9,14 +11,14 @@
 
 
 class EventBusListener(EventBusBaseListener):
- def __init__(self, topic: str, callback: WrappedFunctionBase, event_filter: HINT_EVENT_FILTER_OBJ, **kwargs) -> None:
+ def __init__(self, topic: str, callback: WrappedFunctionBase, event_filter: EventFilterBase, **kwargs: Any) -> None:
  super().__init__(topic, **kwargs)
 
  assert isinstance(callback, WrappedFunctionBase)
  self.func: WrappedFunctionBase = callback
- self.filter: HINT_EVENT_FILTER_OBJ = event_filter
+ self.filter: EventFilterBase = event_filter
 
- def notify_listeners(self, event) -> None:
+ def notify_listeners(self, event: Any) -> None:
  if self.filter.trigger(event):
  self.func.run(event)
 
@@ -28,29 +30,14 @@ def cancel(self) -> None:
  event_bus.remove_listener(self)
 
 
-HINT_EVENT_BUS_LISTENER = TypeVar('HINT_EVENT_BUS_LISTENER', bound=EventBusListener)
-
-
 class ContextBoundEventBusListener(EventBusListener, AutoContextBoundObj):
- def __init__(self, topic: str, callback: WrappedFunctionBase, event_filter: HINT_EVENT_FILTER_OBJ,
- parent_ctx: Context | None = None) -> None:
- super().__init__(topic=topic, callback=callback, event_filter=event_filter, parent_ctx=parent_ctx)
-
- assert isinstance(callback, WrappedFunctionBase)
- self.func: WrappedFunctionBase = callback
- self.filter: HINT_EVENT_FILTER_OBJ = event_filter
-
- def notify_listeners(self, event) -> None:
- if self.filter.trigger(event):
- self.func.run(event)
-
- def describe(self) -> str:
- return f'"{self.topic}" (filter={self.filter.describe()})'
 
+ @override
  def _ctx_unlink(self):
  event_bus.remove_listener(self)
  return super()._ctx_unlink()
 
+ @override
  def cancel(self) -> None:
  """Stop listening on the event bus"""
  self._ctx_unlink()

src/HABApp/core/internals/event_filter.py

@@ -1,4 +1,4 @@
-from typing import Any, TypeVar
+from typing import Any
 
 
 class EventFilterBase:
@@ -10,7 +10,3 @@ def describe(self) -> str:
 
  def __repr__(self) -> str:
  return f'<{self.describe()} at 0x{id(self):X}>'
-
-
-# Hints for functions that use an item class as an input parameter
-HINT_EVENT_FILTER_OBJ = TypeVar('HINT_EVENT_FILTER_OBJ', bound=EventFilterBase)

src/HABApp/core/items/base_item.py

@@ -4,8 +4,8 @@
 
 from HABApp.core.const.hints import TYPE_EVENT_CALLBACK
 from HABApp.core.internals import (
- HINT_EVENT_BUS_LISTENER,
- HINT_EVENT_FILTER_OBJ,
+ EventBusListener,
+ EventFilterBase,
  get_current_context,
  uses_get_item,
  uses_item_registry,
@@ -86,7 +86,7 @@ def watch_update(self, secs: TH_POSITIVE_TIME_DIFF) -> ItemNoUpdateWatch:
  return self._last_update.add_watch(secs)
 
  def listen_event(self, callback: TYPE_EVENT_CALLBACK,
- event_filter: HINT_EVENT_FILTER_OBJ | None = None) -> HINT_EVENT_BUS_LISTENER:
+ event_filter: EventFilterBase | None = None) -> EventBusListener:
  """
  Register an event listener which listens to all event that the item receives
 

src/HABApp/core/items/base_item_watch.py

@@ -1,9 +1,9 @@
+from __future__ import annotations
+
 import logging
 import typing
 
-import HABApp
 from HABApp.core.asyncio import run_func_from_async
-from HABApp.core.const.hints import TYPE_EVENT_CALLBACK
 from HABApp.core.events import EventFilter, ItemNoChangeEvent, ItemNoUpdateEvent
 from HABApp.core.internals import (
  AutoContextBoundObj,
@@ -15,13 +15,18 @@
 from HABApp.core.lib import PendingFuture
 
 
+if typing.TYPE_CHECKING:
+ import HABApp
+ from HABApp.core.const.hints import TYPE_EVENT_CALLBACK
+
+
 log = logging.getLogger('HABApp')
 
 post_event = uses_post_event()
 
 
 class BaseWatch(AutoContextBoundObj):
- EVENT: typing.Type[ItemNoUpdateEvent] | typing.Type[ItemNoChangeEvent]
+ EVENT: type[ItemNoUpdateEvent | ItemNoChangeEvent]
 
  def __init__(self, name: str, secs: int | float) -> None:
  super().__init__()
@@ -40,7 +45,7 @@ def cancel(self) -> None:
  self._ctx_unlink()
  run_func_from_async(self.__cancel_watch)
 
- def listen_event(self, callback: TYPE_EVENT_CALLBACK) -> 'HABApp.core.base.HINT_EVENT_BUS_LISTENER':
+ def listen_event(self, callback: TYPE_EVENT_CALLBACK) -> HABApp.core.base.EventBusListener:
  """Listen to (only) the event that is emitted by this watcher"""
  context = get_current_context()
  return context.add_event_listener(

src/HABApp/core/items/item_aggregation.py

@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import asyncio
 import collections
 import time
@@ -8,7 +10,6 @@
 from HABApp.core.errors import ItemNotFoundException
 from HABApp.core.events import EventFilter, ValueChangeEvent, ValueUpdateEvent
 from HABApp.core.internals import (
- HINT_EVENT_BUS_LISTENER,
  EventBusListener,
  uses_event_bus,
  uses_get_item,
@@ -27,7 +28,7 @@
 class AggregationItem(BaseValueItem):
 
  @classmethod
- def get_create_item(cls, name: str):
+ def get_create_item(cls, name: str) -> AggregationItem:
  """Creates a new AggregationItem in HABApp and returns it or returns the
  already existing item with the given name
 
@@ -50,14 +51,14 @@ def __init__(self, name: str) -> None:
  self.__period: float = 0
  self.__aggregation_func: typing.Callable[[typing.Iterable], typing.Any] = lambda x: x
 
- self._ts: typing.Deque[float] = collections.deque()
- self._vals: typing.Deque[typing.Any] = collections.deque()
+ self._ts: collections.deque[float] = collections.deque()
+ self._vals: collections.deque[typing.Any] = collections.deque()
 
- self.__listener: HINT_EVENT_BUS_LISTENER | None = None
+ self.__listener: EventBusListener | None = None
 
  self.__task: asyncio.Future | None = None
 
- def aggregation_func(self, func: typing.Callable[[typing.Iterable], typing.Any]) -> 'AggregationItem':
+ def aggregation_func(self, func: typing.Callable[[typing.Iterable], typing.Any]) -> AggregationItem:
  """Set the function which will be used to aggregate all values. E.g. ``min`` or ``max``
 
  :param func: The function which takes an iterator an returns an aggregated value.
@@ -66,7 +67,7 @@ def aggregation_func(self, func: typing.Callable[[typing.Iterable], typing.Any])
  self.__aggregation_func = func
  return self
 
- def aggregation_period(self, period: float | int | timedelta) -> 'AggregationItem':
+ def aggregation_period(self, period: float | int | timedelta) -> AggregationItem:
  """Set the period in which the items will be aggregated
 
  :param period: period in seconds
@@ -85,7 +86,7 @@ def aggregation_period(self, period: float | int | timedelta) -> 'AggregationIte
  return self
 
  def aggregation_source(self, source: BaseValueItem | str,
- only_changes: bool = False) -> 'AggregationItem':
+ only_changes: bool = False) -> AggregationItem:
  """Set the source item which changes will be aggregated
 
  :param source: name or Item obj

src/HABApp/rule/rule.py

@@ -15,10 +15,9 @@
 from HABApp.core.asyncio import create_task
 from HABApp.core.const.hints import TYPE_EVENT_CALLBACK
 from HABApp.core.internals import (
- HINT_EVENT_BUS_LISTENER,
- HINT_EVENT_FILTER_OBJ,
  ContextBoundEventBusListener,
  ContextProvidingObj,
+ EventBusListener,
  EventFilterBase,
  uses_item_registry,
  uses_post_event,
@@ -120,8 +119,8 @@ def post_event(self, name: HINT_ITEM_OBJ | str, event: Any):
 
  def listen_event(self, name: HINT_ITEM_OBJ | str,
  callback: TYPE_EVENT_CALLBACK,
- event_filter: HINT_EVENT_FILTER_OBJ | None = None
- ) -> HINT_EVENT_BUS_LISTENER:
+ event_filter: EventFilterBase | None = None
+ ) -> EventBusListener:
  """
  Register an event listener
 

src/HABApp/rule_ctx/rule_ctx.py

@@ -1,11 +1,11 @@
 from __future__ import annotations
 
 import logging
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, TypeVar
 
 import HABApp
 from HABApp.core.const.topics import ALL_TOPICS
-from HABApp.core.internals import HINT_EVENT_BUS_LISTENER, Context, uses_event_bus, uses_item_registry
+from HABApp.core.internals import Context, EventBusListener, uses_event_bus, uses_item_registry
 from HABApp.core.internals.event_bus import EventBusBaseListener
 
 
@@ -21,6 +21,9 @@
 log = logging.getLogger('HABApp.Rule')
 
 
+TB = TypeVar('TB', bound=EventBusListener)
+
+
 class HABAppRuleContext(Context):
  def __init__(self, rule: Rule) -> None:
  super().__init__()
@@ -29,11 +32,11 @@ def __init__(self, rule: Rule) -> None:
  def get_callback_name(self, callback: Callable) -> str | None:
  return f'{self.rule.rule_name}.{callback.__name__}' if self.rule.rule_name else None
 
- def add_event_listener(self, listener: HINT_EVENT_BUS_LISTENER) -> HINT_EVENT_BUS_LISTENER:
+ def add_event_listener(self, listener: TB) -> TB:
  event_bus.add_listener(listener)
  return listener
 
- def remove_event_listener(self, listener: HINT_EVENT_BUS_LISTENER) -> HINT_EVENT_BUS_LISTENER:
+ def remove_event_listener(self, listener: TB) -> TB:
  event_bus.remove_listener(listener)
  return listener
 
@@ -60,7 +63,6 @@ def unload_rule(self) -> None:
  # user implementation
  rule.on_rule_removed()
 
-
  def check_rule(self) -> None:
  with HABApp.core.wrapper.ExceptionToHABApp(log):
  # We need items if we want to run the test