hat.event.manager.main

  1from collections.abc import Collection
  2from pathlib import Path
  3import argparse
  4import asyncio
  5import contextlib
  6import sys
  7
  8from hat import aio
  9from hat import json
 10from hat.drivers import net
 11
 12from hat.event import eventer
 13from hat.event.manager import common
 14from hat.event.manager import rest
 15from hat.event.manager import server
 16
 17
 18def create_argument_parser() -> argparse.ArgumentParser:
 19    parser = argparse.ArgumentParser()
 20    parser.add_argument(
 21        '--host', type=str, default='127.0.0.1',
 22        help="server host name (default '127.0.0.1')")
 23    parser.add_argument(
 24        '--port', type=int, default='23012',
 25        help="server TCP port (default 23012)")
 26    parser.add_argument(
 27        '--client-name', metavar='NAME', type=str, default='manager',
 28        help="client name (default 'manager')")
 29    parser.add_argument(
 30        '--client-token', metavar='TOKEN', type=str, default=None,
 31        help="client token")
 32    subparsers = parser.add_subparsers(
 33        title='actions', dest='action', required=True)
 34
 35    register_parser = subparsers.add_parser(
 36        'register', description="register new event")
 37    register_parser.add_argument(
 38        '--source-timestamp', metavar='TIMESTAMP', type=_parse_timestamp,
 39        default=None,
 40        help="source timestamp")
 41    register_parser.add_argument(
 42        '--payload-type', metavar='TYPE',
 43        choices=['json', 'yaml', 'toml', 'binary', 'none'],
 44        default=None,
 45        help="payload type")
 46    register_parser.add_argument(
 47        '--binary-type', metavar='TYPE', type=str, default='',
 48        help="binary payload type (default '')")
 49    register_parser.add_argument(
 50        '--payload-path', metavar='PATH', type=Path, default=Path('-'),
 51        help="payload file path or '-' for stdin (default '-')")
 52    register_parser.add_argument(
 53        'event_type', metavar='EVENT_TYPE', type=_parse_event_type,
 54        help="event type where segments are delimited by '/'")
 55
 56    query_parser = subparsers.add_parser(
 57        'query', description="query events")
 58    query_subparsers = query_parser.add_subparsers(
 59        title='query types', dest='query_type', required=True)
 60
 61    query_latest_parser = query_subparsers.add_parser(
 62        'latest', description="query latest events")
 63    query_latest_parser.add_argument(
 64        '--event-types', metavar='EVENT_TYPE', type=_parse_event_type,
 65        default=None, nargs='*',
 66        help='query event types')
 67
 68    query_timeseries_parser = query_subparsers.add_parser(
 69        'timeseries', description="query timeseries events")
 70    query_timeseries_parser.add_argument(
 71        '--event-types', metavar='EVENT_TYPE', type=_parse_event_type,
 72        default=None, nargs='*',
 73        help='query event types')
 74    query_timeseries_parser.add_argument(
 75        '--t-from', metavar='TIMESTAMP', type=_parse_timestamp, default=None,
 76        help="from timestamp")
 77    query_timeseries_parser.add_argument(
 78        '--t-to', metavar='TIMESTAMP', type=_parse_timestamp, default=None,
 79        help="to timestamp")
 80    query_timeseries_parser.add_argument(
 81        '--source-t-from', metavar='TIMESTAMP', type=_parse_timestamp,
 82        default=None,
 83        help="from source timestamp")
 84    query_timeseries_parser.add_argument(
 85        '--source-t-to', metavar='TIMESTAMP', type=_parse_timestamp,
 86        default=None,
 87        help="to source timestamp")
 88    query_timeseries_parser.add_argument(
 89        '--order', type=_parse_order, choices=[i.name for i in common.Order],
 90        default=common.Order.DESCENDING,
 91        help="order (default 'DESCENDING')")
 92    query_timeseries_parser.add_argument(
 93        '--order-by', type=_parse_order_by,
 94        choices=[i.name for i in common.OrderBy],
 95        default=common.OrderBy.TIMESTAMP,
 96        help="order (default 'TIMESTAMP')")
 97    query_timeseries_parser.add_argument(
 98        '--max-results', metavar='N', type=int, default=None,
 99        help="maximum number of results")
100    query_timeseries_parser.add_argument(
101        '--last-event-id', metavar='SERVER_ID,SESSION_ID,INSTANCE_ID',
102        type=_parse_event_id, default=None,
103        help="last event id")
104
105    query_server_parser = query_subparsers.add_parser(
106        'server', description="query server events")
107    query_server_parser.add_argument(
108        '--persisted', action='store_true',
109        help="persisted events")
110    query_server_parser.add_argument(
111        '--max-results', metavar='N', type=int, default=None,
112        help="maximum number of results")
113    query_server_parser.add_argument(
114        '--last-event-id', metavar='SERVER_ID,SESSION_ID,INSTANCE_ID',
115        type=_parse_event_id, default=None,
116        help="last event id")
117    query_server_parser.add_argument(
118        'server_id', metavar='SERVER_ID', type=int,
119        help="server id")
120
121    subscribe_parser = subparsers.add_parser(
122        'subscribe', description="watch newly registered events")
123    subscribe_parser.add_argument(
124        '--server-id', type=int, default=None,
125        help="server id")
126    subscribe_parser.add_argument(
127        '--persisted', action='store_true',
128        help="persisted events")
129    subscribe_parser.add_argument(
130        'event_types', metavar='EVENT_TYPE', type=_parse_event_type, nargs='*',
131        help='query event type')
132
133    server_parser = subparsers.add_parser(
134        'server', description="run manager server with web ui")
135    server_parser.add_argument(
136        '--server-id', type=int, default=None,
137        help="server id")
138    server_parser.add_argument(
139        '--persisted', action='store_true',
140        help="persisted events")
141    server_parser.add_argument(
142        'event_types', metavar='EVENT_TYPE', type=_parse_event_type, nargs='*',
143        help='query event type')
144    server_parser.add_argument(
145        '--http-host', type=str, default='127.0.0.1',
146        help="listening host name (default '127.0.0.1')")
147    server_parser.add_argument(
148        '--http-port', type=int, default=23025,
149        help="listening TCP port (default 23025)")
150
151    rest_parser = subparsers.add_parser(
152        'rest', description="run manager rest api server")
153    rest_parser.add_argument(
154        '--http-host', type=str, default='127.0.0.1',
155        help="listening host name (default '127.0.0.1')")
156    rest_parser.add_argument(
157        '--http-port', type=int, default=23017,
158        help="listening TCP port (default 23017)")
159
160    return parser
161
162
163def main():
164    parser = create_argument_parser()
165    args = parser.parse_args()
166
167    addr = net.TcpAddress(args.host, args.port)
168
169    if args.action == 'register':
170        register_event = common.RegisterEvent(
171            type=args.event_type,
172            source_timestamp=args.source_timestamp,
173            payload=_read_payload(payload_type=args.payload_type,
174                                  binary_type=args.binary_type,
175                                  path=args.payload_path))
176
177        co = register(addr=addr,
178                      client_name=args.client_name,
179                      client_token=args.client_token,
180                      register_event=register_event)
181
182    elif args.action == 'query':
183        if args.query_type == 'latest':
184            params = common.QueryLatestParams(
185                event_types=args.event_types)
186
187        elif args.query_type == 'timeseries':
188            params = common.QueryTimeseriesParams(
189                event_types=args.event_types,
190                t_from=args.t_from,
191                t_to=args.t_to,
192                source_t_from=args.source_t_from,
193                source_t_to=args.source_t_to,
194                order=args.order,
195                order_by=args.order_by,
196                max_results=args.max_results,
197                last_event_id=args.last_event_id)
198
199        elif args.query_type == 'server':
200            params = common.QueryTimeseriesParams(
201                server_id=args.server_id,
202                persisted=args.persisted,
203                max_results=args.max_results,
204                last_event_id=args.last_event_id)
205
206        else:
207            raise ValueError('unsupported query type')
208
209        co = query(addr=addr,
210                   client_name=args.client_name,
211                   client_token=args.client_token,
212                   params=params)
213
214    elif args.action == 'subscribe':
215        subscriptions = args.event_types or [('*', )]
216
217        co = subscribe(addr=addr,
218                       client_name=args.client_name,
219                       client_token=args.client_token,
220                       subscriptions=subscriptions,
221                       server_id=args.server_id,
222                       persisted=args.persisted)
223
224    elif args.action == 'server':
225        subscriptions = args.event_types or [('*', )]
226        http_addr = net.TcpAddress(args.http_host, args.http_port)
227
228        co = server.run(addr=addr,
229                        client_name=args.client_name,
230                        client_token=args.client_token,
231                        subscriptions=subscriptions,
232                        server_id=args.server_id,
233                        persisted=args.persisted,
234                        http_addr=http_addr)
235
236    elif args.action == 'rest':
237        http_addr = net.TcpAddress(args.http_host, args.http_port)
238
239        co = rest.run(addr=addr,
240                      client_name=args.client_name,
241                      client_token=args.client_token,
242                      http_addr=http_addr)
243
244    else:
245        raise ValueError('unsupported action')
246
247    aio.init_asyncio()
248    with contextlib.suppress(asyncio.CancelledError):
249        return aio.run_asyncio(co)
250
251
252async def register(addr: net.StreamAddress,
253                   client_name: str,
254                   client_token: str | None,
255                   register_event: common.RegisterEvent):
256    client = await eventer.connect(addr=addr,
257                                   client_name=client_name,
258                                   client_token=client_token)
259
260    try:
261        result = await client.register([register_event])
262
263        if result is None:
264            return 1
265
266    finally:
267        await aio.uncancellable(client.async_close())
268
269
270async def query(addr: net.StreamAddress,
271                client_name: str,
272                client_token: str | None,
273                params: common.QueryParams):
274    client = await eventer.connect(addr=addr,
275                                   client_name=client_name,
276                                   client_token=client_token)
277
278    try:
279        result = await client.query(params)
280
281        result_json = common.query_result_to_json(result)
282        print(json.encode(result_json))
283
284    finally:
285        await aio.uncancellable(client.async_close())
286
287
288async def subscribe(addr: net.StreamAddress,
289                    client_name: str,
290                    client_token: str | None,
291                    subscriptions: Collection[common.EventType],
292                    server_id: int | None,
293                    persisted: bool):
294
295    def on_events(client, events):
296        events_json = [common.event_to_json(event) for event in events]
297        print(json.encode(events_json))
298
299    client = await eventer.connect(addr=addr,
300                                   client_name=client_name,
301                                   client_token=client_token,
302                                   subscriptions=subscriptions,
303                                   server_id=server_id,
304                                   persisted=persisted,
305                                   events_cb=on_events)
306
307    try:
308        await client.wait_closing()
309
310    finally:
311        await aio.uncancellable(client.async_close())
312
313
314def _parse_timestamp(t):
315    if t == 'now':
316        return common.now()
317
318    return common.timestamp_from_float(float(t))
319
320
321def _parse_event_type(event_type):
322    return tuple(event_type.split('/'))
323
324
325def _parse_event_id(event_id):
326    return common.EventId(event_id.split(','))
327
328
329def _parse_order(order):
330    return common.Order[order]
331
332
333def _parse_order_by(order_by):
334    return common.OrderBy[order_by]
335
336
337def _read_payload(payload_type, binary_type, path):
338    if payload_type == 'none':
339        return
340
341    if path == Path('-'):
342        if payload_type == 'json' or payload_type is None:
343            json_format = json.Format.JSON
344
345        elif payload_type == 'yaml':
346            json_format = json.Format.YAML
347
348        elif payload_type == 'toml':
349            json_format = json.Format.TOML
350
351        elif payload_type == 'binary':
352            json_format = None
353
354        else:
355            raise ValueError('unsupported payload type')
356
357        if json_format is None or json_format == json.Format.TOML:
358            stdin, sys.stdin = sys.stdin.detach(), None
359
360        else:
361            stdin = sys.stdin
362
363        if json_format:
364            data = json.decode_stream(stdin, json_format)
365            return common.EventPayloadJson(data)
366
367        else:
368            data = stdin.read()
369            return common.EventPayloadBinary(type=binary_type,
370                                             data=data)
371
372    if payload_type is None:
373        try:
374            json_format = json.get_file_format(path)
375
376        except ValueError:
377            json_format = None
378
379    elif payload_type == 'json':
380        json_format = json.Format.JSON
381
382    elif payload_type == 'yaml':
383        json_format = json.Format.YAML
384
385    elif payload_type == 'toml':
386        json_format = json.Format.TOML
387
388    elif payload_type == 'binary':
389        json_format = None
390
391    else:
392        raise ValueError('unsupported payload type')
393
394    if json_format:
395        data = json.decode_file(path, json_format)
396        return common.EventPayloadJson(data)
397
398    data = path.read_bytes()
399    return common.EventPayloadBinary(type=binary_type,
400                                     data=data)
401
402
403if __name__ == '__main__':
404    sys.argv[0] = 'hat-event-manager'
405    sys.exit(main())