hat.monitor.common

Common functionality shared between clients and monitor server

  1"""Common functionality shared between clients and monitor server"""
  2
  3import importlib.resources
  4import typing
  5
  6from hat import chatter
  7from hat import json
  8from hat import sbs
  9
 10
 11with importlib.resources.path(__package__, 'json_schema_repo.json') as _path:
 12    json_schema_repo: json.SchemaRepository = json.SchemaRepository(
 13        json.json_schema_repo,
 14        json.SchemaRepository.from_json(_path))
 15
 16with importlib.resources.path(__package__, 'sbs_repo.json') as _path:
 17    sbs_repo: sbs.Repository = sbs.Repository(
 18        chatter.sbs_repo,
 19        sbs.Repository.from_json(_path))
 20
 21
 22class BlessingReq(typing.NamedTuple):
 23    token: typing.Optional[int]
 24    timestamp: typing.Optional[float]
 25
 26
 27class BlessingRes(typing.NamedTuple):
 28    token: typing.Optional[int]
 29    ready: bool
 30
 31
 32class ComponentInfo(typing.NamedTuple):
 33    cid: int
 34    mid: int
 35    name: typing.Optional[str]
 36    group: typing.Optional[str]
 37    data: json.Data
 38    rank: int
 39    blessing_req: BlessingReq
 40    blessing_res: BlessingRes
 41
 42
 43class MsgClient(typing.NamedTuple):
 44    name: str
 45    group: str
 46    data: json.Data
 47    blessing_res: BlessingRes
 48
 49
 50class MsgServer(typing.NamedTuple):
 51    cid: int
 52    mid: int
 53    components: typing.List[ComponentInfo]
 54
 55
 56def blessing_req_to_sbs(blessing: BlessingReq) -> sbs.Data:
 57    """Convert blessing request to SBS data"""
 58    return {'token': _value_to_sbs_maybe(blessing.token),
 59            'timestamp': _value_to_sbs_maybe(blessing.timestamp)}
 60
 61
 62def blessing_req_from_sbs(data: sbs.Data) -> BlessingReq:
 63    """Convert SBS data to blessing request"""
 64    return BlessingReq(token=_value_from_sbs_maybe(data['token']),
 65                       timestamp=_value_from_sbs_maybe(data['timestamp']))
 66
 67
 68def blessing_res_to_sbs(res: BlessingRes) -> sbs.Data:
 69    """Convert blessing response to SBS data"""
 70    return {'token': _value_to_sbs_maybe(res.token),
 71            'ready': res.ready}
 72
 73
 74def blessing_res_from_sbs(data: sbs.Data) -> BlessingRes:
 75    """Convert SBS data to blessing response"""
 76    return BlessingRes(token=_value_from_sbs_maybe(data['token']),
 77                       ready=data['ready'])
 78
 79
 80def component_info_to_sbs(info: ComponentInfo) -> sbs.Data:
 81    """Convert component info to SBS data"""
 82    return {'cid': info.cid,
 83            'mid': info.mid,
 84            'name': _value_to_sbs_maybe(info.name),
 85            'group': _value_to_sbs_maybe(info.group),
 86            'data': json.encode(info.data),
 87            'rank': info.rank,
 88            'blessingReq': blessing_req_to_sbs(info.blessing_req),
 89            'blessingRes': blessing_res_to_sbs(info.blessing_res)}
 90
 91
 92def component_info_from_sbs(data: sbs.Data) -> ComponentInfo:
 93    """Convert SBS data to component info"""
 94    return ComponentInfo(
 95        cid=data['cid'],
 96        mid=data['mid'],
 97        name=_value_from_sbs_maybe(data['name']),
 98        group=_value_from_sbs_maybe(data['group']),
 99        data=json.decode(data['data']),
100        rank=data['rank'],
101        blessing_req=blessing_req_from_sbs(data['blessingReq']),
102        blessing_res=blessing_res_from_sbs(data['blessingRes']))
103
104
105def msg_client_to_sbs(msg: MsgClient) -> sbs.Data:
106    """Convert MsgClient to SBS data"""
107    return {'name': msg.name,
108            'group': msg.group,
109            'data': json.encode(msg.data),
110            'blessingRes': blessing_res_to_sbs(msg.blessing_res)}
111
112
113def msg_client_from_sbs(data: sbs.Data) -> MsgClient:
114    """Convert SBS data to MsgClient"""
115    return MsgClient(name=data['name'],
116                     group=data['group'],
117                     data=json.decode(data['data']),
118                     blessing_res=blessing_res_from_sbs(data['blessingRes']))
119
120
121def msg_server_to_sbs(msg: MsgServer) -> sbs.Data:
122    """Convert MsgServer to SBS data"""
123    return {'cid': msg.cid,
124            'mid': msg.mid,
125            'components': [component_info_to_sbs(info)
126                           for info in msg.components]}
127
128
129def msg_server_from_sbs(data: sbs.Data) -> MsgServer:
130    """Convert SBS data to MsgServer"""
131    return MsgServer(cid=data['cid'],
132                     mid=data['mid'],
133                     components=[component_info_from_sbs(info)
134                                 for info in data['components']])
135
136
137def _value_to_sbs_maybe(value):
138    return ('Just', value) if value is not None else ('Nothing', None)
139
140
141def _value_from_sbs_maybe(maybe):
142    return maybe[1]
class BlessingReq(typing.NamedTuple):
23class BlessingReq(typing.NamedTuple):
24    token: typing.Optional[int]
25    timestamp: typing.Optional[float]

BlessingReq(token, timestamp)

BlessingReq(token: Optional[int], timestamp: Optional[float])

Create new instance of BlessingReq(token, timestamp)

token: Optional[int]

Alias for field number 0

timestamp: Optional[float]

Alias for field number 1

Inherited Members
builtins.tuple
index
count
class BlessingRes(typing.NamedTuple):
28class BlessingRes(typing.NamedTuple):
29    token: typing.Optional[int]
30    ready: bool

BlessingRes(token, ready)

BlessingRes(token: Optional[int], ready: bool)

Create new instance of BlessingRes(token, ready)

token: Optional[int]

Alias for field number 0

ready: bool

Alias for field number 1

Inherited Members
builtins.tuple
index
count
class ComponentInfo(typing.NamedTuple):
33class ComponentInfo(typing.NamedTuple):
34    cid: int
35    mid: int
36    name: typing.Optional[str]
37    group: typing.Optional[str]
38    data: json.Data
39    rank: int
40    blessing_req: BlessingReq
41    blessing_res: BlessingRes

ComponentInfo(cid, mid, name, group, data, rank, blessing_req, blessing_res)

ComponentInfo( cid: int, mid: int, name: Optional[str], group: Optional[str], data: ~Data, rank: int, blessing_req: hat.monitor.common.BlessingReq, blessing_res: hat.monitor.common.BlessingRes)

Create new instance of ComponentInfo(cid, mid, name, group, data, rank, blessing_req, blessing_res)

cid: int

Alias for field number 0

mid: int

Alias for field number 1

name: Optional[str]

Alias for field number 2

group: Optional[str]

Alias for field number 3

data: ~Data

Alias for field number 4

rank: int

Alias for field number 5

Alias for field number 6

Alias for field number 7

Inherited Members
builtins.tuple
index
count
class MsgClient(typing.NamedTuple):
44class MsgClient(typing.NamedTuple):
45    name: str
46    group: str
47    data: json.Data
48    blessing_res: BlessingRes

MsgClient(name, group, data, blessing_res)

MsgClient( name: str, group: str, data: ~Data, blessing_res: hat.monitor.common.BlessingRes)

Create new instance of MsgClient(name, group, data, blessing_res)

name: str

Alias for field number 0

group: str

Alias for field number 1

data: ~Data

Alias for field number 2

Alias for field number 3

Inherited Members
builtins.tuple
index
count
class MsgServer(typing.NamedTuple):
51class MsgServer(typing.NamedTuple):
52    cid: int
53    mid: int
54    components: typing.List[ComponentInfo]

MsgServer(cid, mid, components)

MsgServer( cid: int, mid: int, components: List[hat.monitor.common.ComponentInfo])

Create new instance of MsgServer(cid, mid, components)

cid: int

Alias for field number 0

mid: int

Alias for field number 1

Alias for field number 2

Inherited Members
builtins.tuple
index
count
def blessing_req_to_sbs(blessing: hat.monitor.common.BlessingReq) -> ~Data:
57def blessing_req_to_sbs(blessing: BlessingReq) -> sbs.Data:
58    """Convert blessing request to SBS data"""
59    return {'token': _value_to_sbs_maybe(blessing.token),
60            'timestamp': _value_to_sbs_maybe(blessing.timestamp)}

Convert blessing request to SBS data

def blessing_req_from_sbs(data: ~Data) -> hat.monitor.common.BlessingReq:
63def blessing_req_from_sbs(data: sbs.Data) -> BlessingReq:
64    """Convert SBS data to blessing request"""
65    return BlessingReq(token=_value_from_sbs_maybe(data['token']),
66                       timestamp=_value_from_sbs_maybe(data['timestamp']))

Convert SBS data to blessing request

def blessing_res_to_sbs(res: hat.monitor.common.BlessingRes) -> ~Data:
69def blessing_res_to_sbs(res: BlessingRes) -> sbs.Data:
70    """Convert blessing response to SBS data"""
71    return {'token': _value_to_sbs_maybe(res.token),
72            'ready': res.ready}

Convert blessing response to SBS data

def blessing_res_from_sbs(data: ~Data) -> hat.monitor.common.BlessingRes:
75def blessing_res_from_sbs(data: sbs.Data) -> BlessingRes:
76    """Convert SBS data to blessing response"""
77    return BlessingRes(token=_value_from_sbs_maybe(data['token']),
78                       ready=data['ready'])

Convert SBS data to blessing response

def component_info_to_sbs(info: hat.monitor.common.ComponentInfo) -> ~Data:
81def component_info_to_sbs(info: ComponentInfo) -> sbs.Data:
82    """Convert component info to SBS data"""
83    return {'cid': info.cid,
84            'mid': info.mid,
85            'name': _value_to_sbs_maybe(info.name),
86            'group': _value_to_sbs_maybe(info.group),
87            'data': json.encode(info.data),
88            'rank': info.rank,
89            'blessingReq': blessing_req_to_sbs(info.blessing_req),
90            'blessingRes': blessing_res_to_sbs(info.blessing_res)}

Convert component info to SBS data

def component_info_from_sbs(data: ~Data) -> hat.monitor.common.ComponentInfo:
 93def component_info_from_sbs(data: sbs.Data) -> ComponentInfo:
 94    """Convert SBS data to component info"""
 95    return ComponentInfo(
 96        cid=data['cid'],
 97        mid=data['mid'],
 98        name=_value_from_sbs_maybe(data['name']),
 99        group=_value_from_sbs_maybe(data['group']),
100        data=json.decode(data['data']),
101        rank=data['rank'],
102        blessing_req=blessing_req_from_sbs(data['blessingReq']),
103        blessing_res=blessing_res_from_sbs(data['blessingRes']))

Convert SBS data to component info

def msg_client_to_sbs(msg: hat.monitor.common.MsgClient) -> ~Data:
106def msg_client_to_sbs(msg: MsgClient) -> sbs.Data:
107    """Convert MsgClient to SBS data"""
108    return {'name': msg.name,
109            'group': msg.group,
110            'data': json.encode(msg.data),
111            'blessingRes': blessing_res_to_sbs(msg.blessing_res)}

Convert MsgClient to SBS data

def msg_client_from_sbs(data: ~Data) -> hat.monitor.common.MsgClient:
114def msg_client_from_sbs(data: sbs.Data) -> MsgClient:
115    """Convert SBS data to MsgClient"""
116    return MsgClient(name=data['name'],
117                     group=data['group'],
118                     data=json.decode(data['data']),
119                     blessing_res=blessing_res_from_sbs(data['blessingRes']))

Convert SBS data to MsgClient

def msg_server_to_sbs(msg: hat.monitor.common.MsgServer) -> ~Data:
122def msg_server_to_sbs(msg: MsgServer) -> sbs.Data:
123    """Convert MsgServer to SBS data"""
124    return {'cid': msg.cid,
125            'mid': msg.mid,
126            'components': [component_info_to_sbs(info)
127                           for info in msg.components]}

Convert MsgServer to SBS data

def msg_server_from_sbs(data: ~Data) -> hat.monitor.common.MsgServer:
130def msg_server_from_sbs(data: sbs.Data) -> MsgServer:
131    """Convert SBS data to MsgServer"""
132    return MsgServer(cid=data['cid'],
133                     mid=data['mid'],
134                     components=[component_info_from_sbs(info)
135                                 for info in data['components']])

Convert SBS data to MsgServer

json_schema_repo: hat.json.repository.SchemaRepository = <hat.json.repository.SchemaRepository object>
sbs_repo: hat.sbs.repository.Repository = <hat.sbs.repository.Repository object>