hat.gateway.common
Common gateway interfaces
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"""Event type prefix (``gateway/<device_type>/<device_name>``)""" 43 44CreateDevice: typing.TypeAlias = aio.AsyncCallable[[DeviceConf, 45 hat.event.eventer.Client, 46 EventTypePrefix], 47 Device] 48"""Create device callable""" 49 50 51class DeviceInfo(typing.NamedTuple): 52 """Device info 53 54 Device is implemented as python module which is dynamically imported. 55 It is expected that this module contains `info` which is instance of 56 `DeviceInfo`. 57 58 If device defines JSON schema repository and JSON schema id, JSON schema 59 repository will be used for additional validation of device configuration 60 with JSON schema id. 61 62 """ 63 type: str 64 create: CreateDevice 65 json_schema_id: str | None = None 66 json_schema_repo: json.SchemaRepository | None = None 67 68 69def import_device_info(py_module_str: str) -> DeviceInfo: 70 """Import device info""" 71 py_module = importlib.import_module(py_module_str) 72 info = py_module.info 73 74 if not isinstance(info, DeviceInfo): 75 raise Exception('invalid device implementation') 76 77 return info
class
Device(hat.aio.group.Resource):
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: Collection[hat.event.common.common.Event]):
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
DeviceConf: TypeAlias =
None | bool | int | float | str | list[ForwardRef('Data')] | dict[str, ForwardRef('Data')]
Device configuration
EventTypePrefix: TypeAlias =
tuple[str, str, str]
Event type prefix (gateway/<device_type>/<device_name>
)
CreateDevice: TypeAlias =
Callable[[None | bool | int | float | str | list[ForwardRef('Data')] | dict[str, ForwardRef('Data')], hat.event.eventer.client.Client, tuple[str, str, str]], Union[Device, Awaitable[Device]]]
Create device callable
class
DeviceInfo(typing.NamedTuple):
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
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.
DeviceInfo( type: str, create: Callable[[None | bool | int | float | str | list[ForwardRef('Data')] | dict[str, ForwardRef('Data')], hat.event.eventer.client.Client, tuple[str, str, str]], Union[Device, Awaitable[Device]]], json_schema_id: str | None = None, json_schema_repo: hat.json.repository.SchemaRepository | None = None)
Create new instance of DeviceInfo(type, create, json_schema_id, json_schema_repo)
create: Callable[[None | bool | int | float | str | list[ForwardRef('Data')] | dict[str, ForwardRef('Data')], hat.event.eventer.client.Client, tuple[str, str, str]], Union[Device, Awaitable[Device]]]
Alias for field number 1
Inherited Members
- builtins.tuple
- index
- count
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
Import device info
json_schema_repo: hat.json.repository.SchemaRepository =
<hat.json.repository.SchemaRepository object>