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

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

24class Device(aio.Resource):
25    """Device interface"""
26
27    @abc.abstractmethod
28    async def process_events(self,
29                             events: Collection[hat.event.common.Event]):
30        """Process received events
31
32        This method can be coroutine or regular function.
33
34        """

Device interface

@abc.abstractmethod

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

27    @abc.abstractmethod
28    async def process_events(self,
29                             events: Collection[hat.event.common.Event]):
30        """Process received events
31
32        This method can be coroutine or regular function.
33
34        """

Process received events

This method can be coroutine or regular function.

Inherited Members

hat.aio.group.Resource: async_group; is_open; is_closing; is_closed; wait_closing; wait_closed; close; async_close

Device configuration

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

Event type prefix

Create device callable

class DeviceInfo(typing.NamedTuple): View Source

53class DeviceInfo(typing.NamedTuple):
54    """Device info
55
56    Device is implemented as python module which is dynamically imported.
57    It is expected that this module contains `info` which is instance of
58    `DeviceInfo`.
59
60    If device defines JSON schema repository and JSON schema id, JSON schema
61    repository will be used for additional validation of device configuration
62    with JSON schema id.
63
64    """
65    type: str
66    create: CreateDevice
67    json_schema_id: str | None = None
68    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

Inherited Members

builtins.tuple: index; count

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

71def import_device_info(py_module_str: str) -> DeviceInfo:
72    """Import device info"""
73    py_module = importlib.import_module(py_module_str)
74    info = py_module.info
75
76    if not isinstance(info, DeviceInfo):
77        raise Exception('invalid device implementation')
78
79    return info

Import device info

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