hat.monitor.common

  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]