implementation of positioner

src/main/java/telraam/App.java

@@ -20,6 +20,8 @@
 import telraam.logic.lapper.Lapper;
 import telraam.logic.lapper.external.ExternalLapper;
 import telraam.logic.lapper.robust.RobustLapper;
+import telraam.logic.positioner.Positioner;
+import telraam.logic.positioner.simple.SimplePositioner;
 import telraam.station.Fetcher;
 import telraam.util.AcceptedLapsUtil;
 import telraam.websocket.WebSocketConnection;
@@ -131,10 +133,15 @@ public void run(AppConfiguration configuration, Environment environment) throws
                 lapper.registerAPI(jersey);
             }
 
+            // Set up positioners
+            Set<Positioner> positioners = new HashSet<>();
+
+            positioners.add(new SimplePositioner(this.database));
+
             // Start fetch thread for each station
             StationDAO stationDAO = this.database.onDemand(StationDAO.class);
             for (Station station : stationDAO.getAll()) {
-                new Thread(() -> new Fetcher(this.database, station, lappers).fetch()).start();
+                new Thread(() -> new Fetcher(this.database, station, lappers, positioners).fetch()).start();
             }
         }
 

src/main/java/telraam/logic/positioner/Position.java

@@ -1,4 +1,4 @@
-package telraam.logic.positioners;
+package telraam.logic.positioner;
 
 import lombok.Getter;
 import lombok.Setter;
@@ -7,23 +7,13 @@
 
 @Getter @Setter
 public class Position {
-    private Team team;
+    private int teamId;
     private float progress; // Progress of the lap. Between 0-1
     private float speed; // Current speed. Progress / second
 
-    public Position(Team team) {
-        this.team = team;
+    public Position(int teamId) {
+        this.teamId = teamId;
         this.progress = 0;
         this.speed = 0;
     }
-
-    private String toWebsocketMessage() {
-        return String.format("{\"topic\": \"position\", \"data\": {\"team\": %d, \"progress\": %.4f, \"speed\": %.5f}}", team.getId(), this.progress, this.speed);
-    }
-
-    public void send() {
-        WebSocketMessageSingleton.getInstance().sendToAll(
-                this.toWebsocketMessage()
-        );
-    }
 }

src/main/java/telraam/logic/positioner/PositionSender.java

@@ -0,0 +1,31 @@
+package telraam.logic.positioner;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.NoArgsConstructor;
+import telraam.websocket.WebSocketMessage;
+import telraam.websocket.WebSocketMessageSingleton;
+
+import java.util.List;
+import java.util.logging.Logger;
+
+public class PositionSender {
+    private static final Logger logger = Logger.getLogger(PositionSender.class.getName());
+    private final ObjectMapper mapper = new ObjectMapper();
+    private final WebSocketMessage message = new WebSocketMessage();
+
+    public PositionSender() {
+        this.message.setTopic("position");
+    }
+
+    public void send(List<Position> position) {
+        try {
+            String json = mapper.writeValueAsString(position);
+            this.message.setData(json);
+            WebSocketMessageSingleton.getInstance().sendToAll(this.message);
+        } catch (JsonProcessingException e) {
+            logger.severe("Json conversion error for \"%s\"".formatted(position.toString()));
+        }
+    }
+
+}

src/main/java/telraam/logic/positioner/Positioner.java

@@ -0,0 +1,9 @@
+package telraam.logic.positioner;
+
+import telraam.database.models.Detection;
+
+public interface Positioner {
+    void handle(Detection detection);
+
+    void calculatePositions();
+}

src/main/java/telraam/logic/positioner/simple/SimplePositioner.java

@@ -0,0 +1,104 @@
+package telraam.logic.positioner.simple;
+
+import org.jdbi.v3.core.Jdbi;
+import telraam.database.daos.BatonSwitchoverDAO;
+import telraam.database.daos.StationDAO;
+import telraam.database.daos.TeamDAO;
+import telraam.database.models.BatonSwitchover;
+import telraam.database.models.Detection;
+import telraam.database.models.Station;
+import telraam.database.models.Team;
+import telraam.logic.positioner.CircularQueue;
+import telraam.logic.positioner.Position;
+import telraam.logic.positioner.PositionSender;
+import telraam.logic.positioner.Positioner;
+
+import java.util.Comparator;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.logging.Logger;
+
+public class SimplePositioner implements Positioner {
+    private final int QUEUE_SIZE = 50;
+    private final int MIN_RSSI = -85;
+    private final int DEBOUNCE_TIMEOUT = 1;
+    private boolean debounceScheduled;
+    private final ScheduledExecutorService scheduler;
+    private static final Logger logger = Logger.getLogger(SimplePositioner.class.getName());
+    private final PositionSender positionSender;
+    private final Map<Integer, Team> batonIdToTeam;
+    private final Map<Team, CircularQueue<Detection>> teamDetections;
+    private final List<Integer> stations;
+    private final Map<Team, Position> teamPositions;
+
+    public SimplePositioner(Jdbi jdbi) {
+        this.debounceScheduled = false;
+        this.scheduler = Executors.newScheduledThreadPool(1);
+        this.positionSender = new PositionSender();
+        this.batonIdToTeam = new HashMap<>();
+        this.teamDetections = new HashMap<>();
+        this.teamPositions = new HashMap<>();
+
+        TeamDAO teamDAO = jdbi.onDemand(TeamDAO.class);
+        List<Team> teams = teamDAO.getAll();
+        for (Team team: teams) {
+            teamDetections.put(team, new CircularQueue<>(QUEUE_SIZE));
+            teamPositions.put(team, new Position(team.getId()));
+        }
+        List<BatonSwitchover> switchovers = jdbi.onDemand(BatonSwitchoverDAO.class).getAll();
+        switchovers.sort(Comparator.comparing(BatonSwitchover::getTimestamp));
+
+        for (BatonSwitchover switchover: switchovers) {
+            batonIdToTeam.put(switchover.getNewBatonId(), teamDAO.getById(switchover.getTeamId()).get());
+        }
+
+        List<Station> stationList = jdbi.onDemand(StationDAO.class).getAll();
+        stationList.sort(Comparator.comparing(Station::getDistanceFromStart));
+        stations = stationList.stream().map(Station::getId).toList();
+    }
+
+    public void calculatePositions() {
+        logger.info("SimplePositioner: Calculating positions...");
+        for (Map.Entry<Team, CircularQueue<Detection>> entry: teamDetections.entrySet()) {
+            List<Detection> detections = teamDetections.get(entry.getKey());
+            detections.sort(Comparator.comparing(Detection::getTimestamp));
+
+            int currentStationRssi = MIN_RSSI;
+            int currentStationPosition = 0;
+            for (Detection detection: detections) {
+                if (detection.getRssi() > currentStationRssi) {
+                    currentStationRssi = detection.getRssi();
+                    currentStationPosition = detection.getStationId();
+                }
+            }
+
+            float progress = ((float) 100 / stations.size()) * currentStationPosition;
+            teamPositions.get(entry.getKey()).setProgress(progress);
+        }
+
+        positionSender.send(teamPositions.values().stream().toList());
+        logger.info("SimplePositioner: Done calculating positions");
+    }
+
+    public void handle(Detection detection) {
+        Team team = batonIdToTeam.get(detection.getBatonId());
+        teamDetections.get(team).add(detection);
+
+        if (! debounceScheduled) {
+            debounceScheduled = true;
+            scheduler.schedule(() -> {
+                try {
+                    calculatePositions();
+                } catch (Exception e) {
+                    logger.severe(e.getMessage());
+                }
+                debounceScheduled = false;
+            }, DEBOUNCE_TIMEOUT, TimeUnit.SECONDS);
+        }
+    }
+
+}

src/main/java/telraam/station/Fetcher.java

@@ -8,6 +8,7 @@
 import telraam.database.models.Detection;
 import telraam.database.models.Station;
 import telraam.logic.lapper.Lapper;
+import telraam.logic.positioner.Positioner;
 import telraam.station.models.RonnyDetection;
 import telraam.station.models.RonnyResponse;
 
@@ -29,6 +30,7 @@
 
 public class Fetcher {
     private final Set<Lapper> lappers;
+    private final Set<Positioner> positioners;
     private Station station;
 
     private final BatonDAO batonDAO;
@@ -48,11 +50,12 @@ public class Fetcher {
     private final static int IDLE_TIMEOUT_MS = 4000; // Wait 4 seconds
 
 
-    public Fetcher(Jdbi database, Station station, Set<Lapper> lappers) {
+    public Fetcher(Jdbi database, Station station, Set<Lapper> lappers, Set<Positioner> positioners) {
         this.batonDAO = database.onDemand(BatonDAO.class);
         this.detectionDAO = database.onDemand(DetectionDAO.class);
         this.stationDAO = database.onDemand(StationDAO.class);
         this.lappers = lappers;
+        this.positioners = positioners;
 
         this.station = station;
     }
@@ -158,7 +161,10 @@ public void fetch() {
             }
             if (!new_detections.isEmpty()) {
                 detectionDAO.insertAll(new_detections);
-                new_detections.forEach((detection) -> lappers.forEach((lapper) -> lapper.handle(detection)));
+                new_detections.forEach((detection) -> {
+                    lappers.forEach((lapper) -> lapper.handle(detection));
+                    positioners.forEach((positioner) -> positioner.handle(detection));
+                });
             }
 
             this.logger.finer("Fetched " + detections.size() + " detections from " + station.getName() + ", Saved " + new_detections.size());

src/main/java/telraam/websocket/WebSocketMessage.java

@@ -0,0 +1,10 @@
+package telraam.websocket;
+
+import lombok.Getter;
+import lombok.Setter;
+
+@Getter @Setter
+public class WebSocketMessage {
+    private String topic;
+    private String data;
+}

src/main/java/telraam/websocket/WebSocketMessageSingleton.java

@@ -1,20 +1,24 @@
 package telraam.websocket;
 
+import com.fasterxml.jackson.core.JsonGenerationException;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonMappingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import lombok.Getter;
+import lombok.NoArgsConstructor;
 
 import java.util.HashSet;
 import java.util.Set;
 import java.util.logging.Logger;
 
+@NoArgsConstructor
 public class WebSocketMessageSingleton {
     private static final Logger logger = Logger.getLogger(WebSocketMessageSingleton.class.getName());
 
     @Getter
     private static final WebSocketMessageSingleton instance = new WebSocketMessageSingleton();
     private static final Set<WebSocketConnection> registeredConnections = new HashSet<>();
-
-    private WebSocketMessageSingleton() {
-    }
+    private ObjectMapper mapper = new ObjectMapper();
 
     public void registerConnection(WebSocketConnection conn) {
         boolean modified = registeredConnections.add(conn);
@@ -34,4 +38,13 @@ public void sendToAll(String s) {
         logger.finest("Sending \"%s\" to all registered WebSocketConnection instances".formatted(s));
         registeredConnections.forEach(conn -> conn.send(s));
     }
+
+    public void sendToAll(WebSocketMessage message) {
+        try {
+            String json = mapper.writeValueAsString(message);
+            this.sendToAll(json);
+        } catch (JsonProcessingException e) {
+            logger.severe("Json conversion error for \"%s\"".formatted(message.toString()));
+        }
+    }
 }