data_mover_rpi7: upgrade to new data format

prog/data_mover_rpi7.py

@@ -5,26 +5,20 @@ import asyncio
 import datetime
 import sys, os
 import pathlib
+import data_utils
 
-def only_printeble_acii(s):
-    return all(ord(x) >= ord(' ') for x in s) and s.isascii()
 async def main():
     #s = await SSHRunSocket().connect('ssh localhost "/dev/stderr"')
     #s = await SSHRunSocket().connect('ssh localhost "tee /dev/stderr"')
-    s = await communication.SSHRunSocket().connect('ssh rpi0.jk.dyn.blatto.eu /aux/jiri/jr/prog/run_py server.py')
-    c = communication.DownloadServer(s)
+    s = await communication.SSHRunSocket().connect('ssh jr@externalArch.jk.dyn.blatto.eu /mnt/jr/jr/prog/run_py admin_server.py')
+    c = communication.AdminServer(s)
     #await s.write(b"Ahoj!")
     for dt in await c.list_realtime_data(datetime.datetime(2023, 1, 1, 0, 0), datetime.datetime(2100, 1, 1, 0, 0)):
         path = "data/realtime/"+dt_to_path(dt)
-        if not os.path.isdir(path):
+        if not (os.path.isdir(path) or os.path.isfile(path+".json.zst") or os.path.isfile(path+".json.gzip")):
             print("WILL DOWNLOAD", path)
-            pathlib.Path(path).mkdir(parents=True)
-            for fname, data in (await c.get_data(dt)).items():
-                assert '/' not in fname and only_printeble_acii(fname)
-                with open(path+'/'+fname, "wb") as f:
-                    f.write(data)
-                    # Todo tmpdir
-        if (datetime.datetime.now(communication.local_timezone)-dt).total_seconds() > 60*60*24 * 300:
+            data_utils.write_data(dt, await c.get_data(dt))
+        if (datetime.datetime.now(communication.local_timezone)-dt).total_seconds() > 60*60*24 * 250:
             print("REMOVE", path)
             await c.remove_data(dt)
 

prog/data_utils.py

@@ -224,3 +224,33 @@ class TripPoint:
         self.openapi_last_stop = json['properties']['last_position']['last_stop']
         self.openapi_next_stop = json['properties']['last_position']['next_stop']
         self.openapi_delay = json['properties']['last_position']['delay']
+
+def write_data(path, dt):
+    import random
+
+    def only_printable_acii(s):
+        return all(ord(x) >= ord(' ') for x in s) and s.isascii()
+
+    tmp_file = dt.strftime(f"tmp/realtime-%Y-%m-%d--%H-%M-%S-{randint(random.randint(0,100000))}")
+
+    if  type(data) == dict:
+        os.mkdir(tmp_file)
+        pathlib.Path(path)
+        for fname, d in data:
+            assert '/' not in fname and only_printable_acii(fname)
+            with open(tmp_file+'/'+fname, "wb") as f:
+                f.write(d)
+    else:
+        with open(tmp_file, "xb") as f:
+            assert type(data) == bytes and data.startswith(b'(\xb5/\xfd')
+            f.write(data)
+
+    file_path = [ "data/realtime", dt.strftime('%Y-%m-%d'), dt.strftime('%H'), dt.strftime("%M-%S") if type(data) == dict else dt.strftime("%M-%S.json.zst")]
+
+    for i in range(2,4):
+        try:
+            os.mkdir('/'.join(file_path[:i]))
+        except FileExistsError:
+            pass
+
+    os.rename(tmp_file, "/".join(file_path))