this repo has no description
Add plc-activity.py
+46
+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(60)