X-Git-Url: http://average.org/gitweb/?a=blobdiff_plain;f=gps303%2Fwsgateway.py;h=79937d8e5b1f5d1835e1dfab8a627f2844701f31;hb=311d3cc7b0692e66edb9b9bb9285b2bfc094d571;hp=7756fe9d52ae10d0bcdb8012869ac7e78f3393b9;hpb=96538346bd332d76d2cac5d6a0ef2b4e4a40de30;p=loctrkd.git

diff --git a/gps303/wsgateway.py b/gps303/wsgateway.py
index 7756fe9..79937d8 100644
--- a/gps303/wsgateway.py
+++ b/gps303/wsgateway.py
@@ -1,6 +1,7 @@
 """ Websocket Gateway """
 
-from json import loads
+from datetime import datetime, timezone
+from json import dumps, loads
 from logging import getLogger
 from socket import socket, AF_INET6, SOCK_STREAM, SOL_SOCKET, SO_REUSEADDR
 from time import time
@@ -17,13 +18,41 @@ from wsproto.utilities import RemoteProtocolError
 import zmq
 
 from . import common
-from .backlog import blinit, backlog
-from .zmsg import LocEvt
+from .evstore import initdb, fetch
+from .gps303proto import (
+    GPS_POSITIONING,
+    WIFI_POSITIONING,
+    parse_message,
+)
+from .zmsg import Bcast, topic
 
 log = getLogger("gps303/wsgateway")
 htmlfile = None
 
 
+def backlog(imei, numback):
+    result = []
+    for is_incoming, timestamp, packet in fetch(
+        imei,
+        ((True, GPS_POSITIONING.PROTO), (False, WIFI_POSITIONING.PROTO)),
+        numback,
+    ):
+        msg = parse_message(packet, is_incoming=is_incoming)
+        result.append(
+            {
+                "imei": imei,
+                "timestamp": str(
+                    datetime.fromtimestamp(timestamp).astimezone(
+                        tz=timezone.utc
+                    )
+                ),
+                "longitude": msg.longitude,
+                "latitude": msg.latitude,
+            }
+        )
+    return result
+
+
 def try_http(data, fd, e):
     global htmlfile
     try:
@@ -157,12 +186,17 @@ class Client:
         return msgs
 
     def wants(self, imei):
-        log.debug("wants %s? set is %s on fd %d", imei, self.imeis, self.sock.fileno())
+        log.debug(
+            "wants %s? set is %s on fd %d",
+            imei,
+            self.imeis,
+            self.sock.fileno(),
+        )
         return True  # TODO: check subscriptions
 
     def send(self, message):
-        if self.ready and message.imei in self.imeis:
-            self.ws_data += self.ws.send(Message(data=message.json))
+        if self.ready and message["imei"] in self.imeis:
+            self.ws_data += self.ws.send(Message(data=dumps(message)))
 
     def write(self):
         if self.ws_data:
@@ -202,7 +236,7 @@ class Clients:
     def send(self, msg):
         towrite = set()
         for fd, clnt in self.by_fd.items():
-            if clnt.wants(msg.imei):
+            if clnt.wants(msg["imei"]):
                 clnt.send(msg)
                 towrite.add(fd)
         return towrite
@@ -224,11 +258,11 @@ class Clients:
 def runserver(conf):
     global htmlfile
 
-    blinit(conf.get("storage", "dbfn"), conf.get("opencellid", "dbfn"))
+    initdb(conf.get("storage", "dbfn"))
     htmlfile = conf.get("wsgateway", "htmlfile")
     zctx = zmq.Context()
     zsub = zctx.socket(zmq.SUB)
-    zsub.connect(conf.get("lookaside", "publishurl"))
+    zsub.connect(conf.get("collector", "publishurl"))
     tcpl = socket(AF_INET6, SOCK_STREAM)
     tcpl.setblocking(False)
     tcpl.setsockopt(SOL_SOCKET, SO_REUSEADDR, 1)
@@ -245,9 +279,23 @@ def runserver(conf):
         while True:
             neededsubs = clients.subs()
             for imei in neededsubs - activesubs:
-                zsub.setsockopt(zmq.SUBSCRIBE, imei.encode())
+                zsub.setsockopt(
+                    zmq.SUBSCRIBE,
+                    topic(GPS_POSITIONING.PROTO, True, imei),
+                )
+                zsub.setsockopt(
+                    zmq.SUBSCRIBE,
+                    topic(WIFI_POSITIONING.PROTO, False, imei),
+                )
             for imei in activesubs - neededsubs:
-                zsub.setsockopt(zmq.UNSUBSCRIBE, imei.encode())
+                zsub.setsockopt(
+                    zmq.UNSUBSCRIBE,
+                    topic(GPS_POSITIONING.PROTO, True, imei),
+                )
+                zsub.setsockopt(
+                    zmq.UNSUBSCRIBE,
+                    topic(WIFI_POSITIONING.PROTO, False, imei),
+                )
             activesubs = neededsubs
             log.debug("Subscribed to: %s", activesubs)
             tosend = []
@@ -259,8 +307,21 @@ def runserver(conf):
                 if sk is zsub:
                     while True:
                         try:
-                            zmsg = LocEvt(zsub.recv(zmq.NOBLOCK))
-                            tosend.append(zmsg)
+                            zmsg = Bcast(zsub.recv(zmq.NOBLOCK))
+                            msg = parse_message(zmsg.packet, zmsg.is_incoming)
+                            log.debug("Got %s with %s", zmsg, msg)
+                            tosend.append(
+                                {
+                                    "imei": zmsg.imei,
+                                    "timestamp": str(
+                                        datetime.fromtimestamp(
+                                            zmsg.when
+                                        ).astimezone(tz=timezone.utc)
+                                    ),
+                                    "longitude": msg.longitude,
+                                    "latitude": msg.latitude,
+                                }
+                            )
                         except zmq.Again:
                             break
                 elif sk == tcpfd:
@@ -276,9 +337,10 @@ def runserver(conf):
                         for msg in received:
                             log.debug("Received from %d: %s", sk, msg)
                             if msg.get("type", None) == "subscribe":
-                                imei = msg.get("imei")
-                                if imei:
-                                    tosend.extend(backlog(imei[0], 5))
+                                imeis = msg.get("imei")
+                                numback = msg.get("backlog", 5)
+                                for imei in imeis:
+                                    tosend.extend(backlog(imei, numback))
                         towrite.add(sk)
                 elif fl & zmq.POLLOUT:
                     log.debug("Write now open for fd %d", sk)