edavis.dev
this repo has no description
feat: drop plc-activity.py
-46
plc-activity.py
-46
plc-activity.py
···
1
-
#!/usr/bin/env python3
2
-
3
-
from datetime import datetime, timezone
4
-
import sys
5
-
import json
6
-
import redis
7
-
import requests
8
-
import time
9
-
10
-
PLC_EXPORT_URL = 'https://plc.directory/export'
11
-
12
-
cids = []
13
-
redis_conn = redis.Redis()
14
-
redis_pipe = redis_conn.pipeline()
15
-
16
-
while True:
17
-
plc_latest = redis_conn.get('dev.edavis.muninsky.plc_latest')
18
-
assert plc_latest is not None, 'manually set the `plc_latest` redis key first'
19
-
ts = datetime.fromisoformat(plc_latest.decode())
20
-
ts = ts.isoformat('T', 'milliseconds').replace('+00:00', 'Z')
21
-
22
-
qs = '?after={ts}'.format(ts=ts)
23
-
24
-
print(f'Requesting {PLC_EXPORT_URL}{qs}', end=' ')
25
-
resp = requests.get(PLC_EXPORT_URL + qs)
26
-
resp.raise_for_status()
27
-
28
-
ops = 0
29
-
after = datetime.now(timezone.utc)
30
-
for line in resp.iter_lines():
31
-
doc = json.loads(line)
32
-
after = datetime.strptime(doc['createdAt'], '%Y-%m-%dT%H:%M:%S.%fZ').replace(tzinfo=timezone.utc)
33
-
if doc['cid'] in cids:
34
-
continue
35
-
cids.insert(0, doc['cid'])
36
-
redis_pipe.incr('dev.edavis.muninsky.plc_ops')
37
-
ops += 1
38
-
39
-
print(f'Fetched {ops} operations')
40
-
sys.stdout.flush()
41
-
cids = cids[:25]
42
-
43
-
redis_pipe.set('dev.edavis.muninsky.plc_latest', after.isoformat())
44
-
redis_pipe.execute()
45
-
46
-
time.sleep(5)