hat.gateway.common

Common gateway interfaces

View Source

 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.SchemaRepository(
19        json.json_schema_repo,
20        json.SchemaRepository.from_json(_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

class Device(hat.aio.group.Resource): View Source

30class Device(aio.Resource):
31    """Device interface"""
32
33    @abc.abstractmethod
34    async def process_events(self,
35                             events: Collection[hat.event.common.Event]):
36        """Process received events
37
38        This method can be coroutine or regular function.
39
40        """

Device interface

@abc.abstractmethod

async def process_events(self, events: Collection[hat.event.common.common.Event]): View Source

33    @abc.abstractmethod
34    async def process_events(self,
35                             events: Collection[hat.event.common.Event]):
36        """Process received events
37
38        This method can be coroutine or regular function.
39
40        """

Process received events

This method can be coroutine or regular function.

Device configuration

EventTypePrefix: TypeAlias = tuple[str, str, str]

Event type prefix (gateway/<device_type>/<device_name>)

Create device callable

class DeviceInfo(typing.NamedTuple): View Source

58class DeviceInfo(typing.NamedTuple):
59    """Device info
60
61    Device is implemented as python module which is dynamically imported.
62    It is expected that this module contains `info` which is instance of
63    `DeviceInfo`.
64
65    If device defines JSON schema repository and JSON schema id, JSON schema
66    repository will be used for additional validation of device configuration
67    with JSON schema id.
68
69    """
70    type: str
71    create: CreateDevice
72    json_schema_id: str | None = None
73    json_schema_repo: json.SchemaRepository | None = None

Device info

Device is implemented as python module which is dynamically imported. It is expected that this module contains info which is instance of DeviceInfo.

If device defines JSON schema repository and JSON schema id, JSON schema repository will be used for additional validation of device configuration with JSON schema id.

Create new instance of DeviceInfo(type, create, json_schema_id, json_schema_repo)

type: str

Alias for field number 0

Alias for field number 1

json_schema_id: str | None

Alias for field number 2

json_schema_repo: hat.json.repository.SchemaRepository | None

Alias for field number 3

def import_device_info(py_module_str: str) -> DeviceInfo: View Source

76def import_device_info(py_module_str: str) -> DeviceInfo:
77    """Import device info"""
78    py_module = importlib.import_module(py_module_str)
79    info = py_module.info
80
81    if not isinstance(info, DeviceInfo):
82        raise Exception('invalid device implementation')
83
84    return info

Import device info

json_schema_repo: hat.json.repository.SchemaRepository = <hat.json.repository.SchemaRepository object>

sbs_repo: hat.sbs.repository.Repository = <hat.sbs.repository.Repository object>