hat.gateway.common

 1"""Common gateway interfaces"""
 2
 3from collections.abc import Collection
 4import abc
 5import importlib.resources
 6import typing
 7
 8from hat import aio
 9from hat import json
10from hat import sbs
11import hat.event.common
12import hat.event.eventer
13import hat.monitor.common
14
15
16with importlib.resources.as_file(importlib.resources.files(__package__) /
17                                 'json_schema_repo.json') as _path:
18    json_schema_repo: json.SchemaRepository = json.merge_schema_repositories(
19        json.json_schema_repo,
20        json.decode_file(_path))
21    """JSON schema repository"""
22
23with importlib.resources.as_file(importlib.resources.files(__package__) /
24                                 'sbs_repo.json') as _path:
25    sbs_repo: sbs.Repository = sbs.Repository.from_json(_path)
26    """SBS schema repository"""
27
28
29class Device(aio.Resource):
30    """Device interface"""
31
32    @abc.abstractmethod
33    async def process_events(self,
34                             events: Collection[hat.event.common.Event]):
35        """Process received events
36
37        This method can be coroutine or regular function.
38
39        """
40
41
42DeviceConf: typing.TypeAlias = json.Data
43"""Device configuration"""
44
45EventTypePrefix: typing.TypeAlias = tuple[hat.event.common.EventTypeSegment,
46                                          hat.event.common.EventTypeSegment,
47                                          hat.event.common.EventTypeSegment]
48"""Event type prefix (``gateway/<device_type>/<device_name>``)"""
49
50CreateDevice: typing.TypeAlias = aio.AsyncCallable[[DeviceConf,
51                                                    hat.event.eventer.Client,
52                                                    EventTypePrefix],
53                                                   Device]
54"""Create device callable"""
55
56
57class DeviceInfo(typing.NamedTuple):
58    """Device info
59
60    Device is implemented as python module which is dynamically imported.
61    It is expected that this module contains `info` which is instance of
62    `DeviceInfo`.
63
64    If device defines JSON schema repository and JSON schema id, JSON schema
65    repository will be used for additional validation of device configuration
66    with JSON schema id.
67
68    """
69    type: str
70    create: CreateDevice
71    json_schema_id: str | None = None
72    json_schema_repo: json.SchemaRepository | None = None
73
74
75def import_device_info(py_module_str: str) -> DeviceInfo:
76    """Import device info"""
77    py_module = importlib.import_module(py_module_str)
78    info = py_module.info
79
80    if not isinstance(info, DeviceInfo):
81        raise Exception('invalid device implementation')
82
83    return info