hat.monitor.common

 1"""Common functionality shared between clients and monitor server"""
 2
 3import importlib.resources
 4import typing
 5
 6from hat import json
 7from hat import sbs
 8
 9
10with importlib.resources.as_file(importlib.resources.files(__package__) /
11                                 'json_schema_repo.json') as _path:
12    json_schema_repo: json.SchemaRepository = json.merge_schema_repositories(
13        json.json_schema_repo,
14        json.decode_file(_path))
15
16with importlib.resources.as_file(importlib.resources.files(__package__) /
17                                 'sbs_repo.json') as _path:
18    sbs_repo: sbs.Repository = sbs.Repository.from_json(_path)
19
20
21Cid: typing.TypeAlias = int
22
23Mid: typing.TypeAlias = int
24
25
26class BlessingReq(typing.NamedTuple):
27    token: int | None
28    timestamp: float | None
29
30
31class BlessingRes(typing.NamedTuple):
32    token: int | None
33    ready: bool
34
35
36class ComponentInfo(typing.NamedTuple):
37    cid: Cid
38    mid: Mid
39    name: str | None
40    group: str | None
41    data: json.Data
42    rank: int
43    blessing_req: BlessingReq
44    blessing_res: BlessingRes