feat: add parser library

 - clean up unused main
 - drop support for python 3.9 since it is end of life
 - implement a basic parser registry

Author: jterrace

.github/workflows/ci.yml

@@ -36,7 +36,6 @@ jobs:
       fail-fast: false
       matrix:
         python-version:
-          - "3.9"
           - "3.10"
           - "3.11"
           - "3.12"

pyproject.toml

@@ -11,17 +11,17 @@ license = { text = "Apache Software License 2.0" }
 authors = [
   { name = "Open Video Libs", email = "openvideolibs@koston.org" },
 ]
-requires-python = ">=3.9"
+requires-python = ">=3.10"
 classifiers = [
   "Development Status :: 2 - Pre-Alpha",
   "Intended Audience :: Developers",
   "Natural Language :: English",
   "Operating System :: OS Independent",
-  "Programming Language :: Python :: 3.9",
   "Programming Language :: Python :: 3.10",
   "Programming Language :: Python :: 3.11",
   "Programming Language :: Python :: 3.12",
   "Programming Language :: Python :: 3.13",
+  "Programming Language :: Python :: 3.14",
   "Topic :: Software Development :: Libraries",
 ]
 
@@ -35,6 +35,7 @@ urls.repository = "https://github.com/openvideolibs/onvif-parsers"
 [dependency-groups]
 dev = [
   "pytest>=8,<9",
+  "pytest-asyncio>=1.2.0",
   "pytest-cov>=6,<8",
 ]
 docs = [
@@ -88,6 +89,7 @@ addopts = """\
     --cov-report=xml
     """
 pythonpath = [ "src" ]
+asyncio_default_fixture_loop_scope = "session"
 
 [tool.coverage.run]
 branch = true

src/onvif_parsers/__init__.py

@@ -1 +1,7 @@
+from . import registry
+
 __version__ = "0.0.1"
+
+__all__ = ["get"]
+
+get = registry.get_parser

src/onvif_parsers/main.py

@@ -0,0 +1,28 @@
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass
+class EventEntity:
+    """Represents a ONVIF event entity."""
+
+    # Unique identifier for the entity
+    uid: str
+    # Human-readable name for the entity
+    name: str
+    # Type of platform (e.g., sensor, binary_sensor)
+    platform: str
+    # Optional device class (e.g., motion, alarm, safety). The options vary based on the
+    # platform.
+    # See https://www.home-assistant.io/integrations/homeassistant/#device-class
+    device_class: str | None = None
+    # Optional unit of measurement (e.g., percent)
+    unit_of_measurement: str | None = None
+    # Current value of the entity. Most onvif events are boolean (true/false), but this
+    # could be an integer or timestamp or other data types supported as well.
+    value: Any = None
+    # Optional entity category (e.g., diagnostic, configuration). The default (sensor)
+    # does not need to be specified.
+    entity_category: str | None = None
+    # Indicates whether the entity is enabled by default. Defaults to True.
+    entity_enabled: bool = True

src/onvif_parsers/model.py

@@ -0,0 +1,51 @@
+import typing
+from collections.abc import Callable
+
+from . import model
+
+# Type alias for a parser callable. It should be an async function.
+# Args:
+#  str: The uid of the entity.
+#  Any: The raw event data. zeep.xsd.ComplexType or zeep.xsd.AnySimpleType.
+#       TODO: could we make this zeep.Type or zeep.AnyType?
+# Returns:
+#  Awaitable[model.EventEntity]: The parsed EventEntity.
+ParserCallable: typing.TypeAlias = Callable[
+    [str, typing.Any], typing.Awaitable[model.EventEntity | None]
+]
+
+
+class Registry:
+    """A registry of parsers."""
+
+    def __init__(self) -> None:
+        self.registry: dict[str, ParserCallable] = {}
+
+    def register(self, key: str, f: ParserCallable) -> None:
+        """Register a parser function under a given key."""
+        if key in self.registry:
+            raise ValueError(f"Key {key} already registered")
+
+        self.registry[key] = f
+
+    def get(self, key: str) -> ParserCallable | None:
+        """Get a parser function by key."""
+        return self.registry.get(key)
+
+
+_REGISTRY = Registry()
+
+
+def register(topic: str) -> Callable[[ParserCallable], ParserCallable]:
+    """Register an onvif parser callable with the given topic."""
+
+    def wrapper(func: ParserCallable) -> ParserCallable:
+        _REGISTRY.register(topic, func)
+        return func
+
+    return wrapper
+
+
+def get_parser(topic: str) -> ParserCallable | None:
+    """Get a parser callable for the given topic."""
+    return _REGISTRY.get(topic)

src/onvif_parsers/registry.py

@@ -0,0 +1,51 @@
+import typing
+
+import pytest
+
+from onvif_parsers import model, registry
+
+
+@registry.register("decorator_test_topic")
+async def parser1(uid: str, _: typing.Any) -> model.EventEntity | None:
+    return model.EventEntity(
+        uid=uid,
+        name="Test",
+        platform="sensor",
+    )
+
+
+def test_get_parser_none():
+    """Getting a non-registered parser returns None."""
+    parser = registry.get_parser("non_existent_topic")
+    assert parser is None
+
+
+@pytest.mark.asyncio
+async def test_registration_works():
+    """Registering a parser works as expected."""
+    registry.register("test_topic1")(parser1)
+    parser = registry.get_parser("test_topic1")
+    assert parser is not None
+    assert callable(parser)
+    assert await parser("entity_1", None) == model.EventEntity(
+        uid="entity_1", name="Test", platform="sensor"
+    )
+
+
+@pytest.mark.asyncio
+async def test_decorator_registration():
+    """Registering a parser via decorator works as expected."""
+    parser = registry.get_parser("decorator_test_topic")
+    assert parser is not None
+    assert callable(parser)
+    assert await parser("entity_1", None) == model.EventEntity(
+        uid="entity_1", name="Test", platform="sensor"
+    )
+
+
+def test_double_registration():
+    """Registering the same topic twice raises ValueError."""
+    registry.register("test_topic2")(parser1)
+
+    with pytest.raises(ValueError):
+        registry.register("test_topic2")(parser1)