utils.match_monitor

22class MatchMonitorConnection:
23    """Listener + uploader pair, wired together and started."""
24
25    def __init__(self) -> None:
26        self.listener: Optional[ControlListener] = None
27        self.uploader: Optional[LogUploader] = None
28
29        try:
30            self.listener = ControlListener()
31            self.listener.start()
32        except Exception:
33            wpilib.reportError("Unable to create ControlListener", printTrace=True)
34            return
35
36        try:
37            self.uploader = LogUploader(self.listener)
38        except Exception:
39            wpilib.reportError("Unable to create LogUploader", printTrace=True)
40            return
41
42        self.listener.on_force_upload = self.uploader.start_upload
43        self.listener.on_stop_upload = self.uploader.stop_upload
44        self.listener.on_clear_manifest_done = self._on_clear_manifest
45
46    def start_upload(self) -> None:
47        if self.uploader is not None:
48            self.uploader.start_upload()
49
50    def stop_upload(self) -> None:
51        if self.uploader is not None:
52            self.uploader.stop_upload()
53
54    def _on_clear_manifest(self) -> None:
55        # Let current file finish uploading, then stop, clear, restart.
56        self.uploader.stop_and_wait()
57        count = self.listener._clear_manifests()
58        self.listener.send_message({'type': 'MANIFEST_CLEARED', 'count': count})
59        self.uploader.start_upload()

 21class ControlListener:
 22    """Listens for a host TCP connection and maintains a keepalive channel.
 23
 24    Protocol (newline-delimited JSON over TCP):
 25        Host → Robot: HELLO {http_port}
 26        Robot → Host: HELLO_ACK
 27        Host → Robot: PING
 28        Robot → Host: PONG
 29        Robot → Host: UPLOAD_STARTING / UPLOAD_COMPLETE
 30        Host → Robot: CLEAR_MANIFEST / FORCE_UPLOAD / STOP_UPLOAD / LIST_LOGS
 31    """
 32
 33    status = ntproperty('/ControlListener/status', 'idle',
 34                         writeDefault=True, persistent=False)
 35
 36    def __init__(self) -> None:
 37        self._host_ip: Optional[str] = None
 38        self._http_port: Optional[int] = None
 39        self._conn: Optional[socket.socket] = None
 40        self._conn_lock = threading.Lock()
 41        self._thread: Optional[threading.Thread] = None
 42        self.on_force_upload: Optional[Callable[[], None]] = None
 43        self.on_stop_upload: Optional[Callable[[], None]] = None
 44        self.on_clear_manifest_done: Optional[Callable[[], None]] = None
 45
 46    @property
 47    def host_ip(self) -> Optional[str]:
 48        return self._host_ip
 49
 50    @property
 51    def http_port(self) -> Optional[int]:
 52        return self._http_port
 53
 54    @property
 55    def is_connected(self) -> bool:
 56        return self._host_ip is not None and self._conn is not None
 57
 58    def start(self) -> None:
 59        """Start the listener thread. Safe to call multiple times."""
 60        if self._thread is not None and self._thread.is_alive():
 61            return
 62        self._thread = threading.Thread(target=self._listen_loop, daemon=True)
 63        self._thread.start()
 64
 65    def send_message(self, msg: dict) -> bool:
 66        """Send a JSON message to the connected host. Returns False if not connected."""
 67        with self._conn_lock:
 68            if self._conn is None:
 69                return False
 70            try:
 71                data = json.dumps(msg) + '\n'
 72                self._conn.sendall(data.encode('utf-8'))
 73                return True
 74            except Exception:
 75                logger.exception("Failed to send message to host")
 76                return False
 77
 78    def send_upload_starting(self, event_name: str, match_type: str,
 79                             match_number: str) -> bool:
 80        return self.send_message({
 81            'type': 'UPLOAD_STARTING',
 82            'event_name': event_name,
 83            'match_type': match_type,
 84            'match_number': match_number,
 85        })
 86
 87    def send_upload_complete(self, event_name: str, match_type: str,
 88                             match_number: str) -> bool:
 89        return self.send_message({
 90            'type': 'UPLOAD_COMPLETE',
 91            'event_name': event_name,
 92            'match_type': match_type,
 93            'match_number': match_number,
 94        })
 95
 96    def _listen_loop(self) -> None:
 97        """Main listener loop — binds, accepts, handles one connection at a time."""
 98        server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
 99        server_sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
100        server_sock.bind(('0.0.0.0', CONTROL_PORT))
101        server_sock.listen(1)
102        logger.info(f"ControlListener listening on port {CONTROL_PORT}")
103        self.status = 'listening'
104        print(f"[ControlListener] Listening on port {CONTROL_PORT}")
105
106        while True:
107            try:
108                conn, addr = server_sock.accept()
109                print(f"[ControlListener] Connection from {addr[0]}:{addr[1]}")
110                self._handle_connection(conn, addr)
111            except Exception:
112                logger.exception("Error accepting connection")
113
114    def _handle_connection(self, conn: socket.socket, addr: tuple) -> None:
115        """Handle a single host connection: handshake, then keepalive loop."""
116        # Close any existing connection
117        with self._conn_lock:
118            if self._conn is not None:
119                try:
120                    self._conn.close()
121                except Exception:
122                    logger.debug("Error closing previous connection", exc_info=True)
123            self._conn = conn
124
125        conn.settimeout(35)  # slightly more than keepalive interval + timeout
126
127        try:
128            buf = ''
129            # Read HELLO message
130            buf = self._read_message(conn, buf)
131            if buf is None:
132                return
133
134            msg, buf = self._parse_next_message(buf)
135            if msg is None or msg.get('type') != 'HELLO':
136                logger.warning(f"Expected HELLO, got: {msg}")
137                return
138
139            host_ip = addr[0]
140            http_port = msg.get('http_port', 5800)
141            self._host_ip = host_ip
142            self._http_port = int(http_port)
143
144            # Send HELLO_ACK
145            ack = json.dumps({'type': 'HELLO_ACK'}) + '\n'
146            conn.sendall(ack.encode('utf-8'))
147
148            self.status = f'connected ({host_ip})'
149            logger.info(f"Host connected: {host_ip}, HTTP port {http_port}")
150            print(f"[ControlListener] Host connected: {host_ip}:{http_port}")
151
152            # Keepalive loop — respond to PINGs and host commands
153            while True:
154                buf = self._read_message(conn, buf)
155                if buf is None:
156                    break
157                msg, buf = self._parse_next_message(buf)
158                if msg is None:
159                    continue
160                self._handle_host_message(msg, conn)
161
162        except socket.timeout:
163            logger.warning("Host connection timed out (no keepalive)")
164            print("[ControlListener] Host connection timed out")
165        except (ConnectionResetError, BrokenPipeError, OSError):
166            logger.info("Host disconnected")
167            print("[ControlListener] Host disconnected")
168        except Exception:
169            logger.exception("Error in host connection handler")
170        finally:
171            self._host_ip = None
172            self._http_port = None
173            with self._conn_lock:
174                self._conn = None
175            try:
176                conn.close()
177            except Exception:
178                logger.debug("Error closing connection socket", exc_info=True)
179            self.status = 'listening'
180            print("[ControlListener] Back to listening")
181
182    def _handle_host_message(self, msg: dict, conn: socket.socket) -> None:
183        """Process a message from the host."""
184        msg_type = msg.get('type', '')
185
186        if msg_type == 'PING':
187            pong = json.dumps({'type': 'PONG'}) + '\n'
188            conn.sendall(pong.encode('utf-8'))
189
190        elif msg_type == 'CLEAR_MANIFEST':
191            if self.on_clear_manifest_done:
192                try:
193                    # Callback handles: stop upload, wait, clear, respond, restart
194                    self.on_clear_manifest_done()
195                except Exception:
196                    logger.exception("on_clear_manifest_done callback failed")
197            else:
198                # No callback wired — just clear and respond
199                count = self._clear_manifests()
200                self.send_message({'type': 'MANIFEST_CLEARED', 'count': count})
201
202        elif msg_type == 'FORCE_UPLOAD':
203            if self.on_force_upload:
204                try:
205                    self.on_force_upload()
206                except Exception:
207                    logger.exception("on_force_upload callback failed")
208            self.send_message({'type': 'FORCE_UPLOAD_ACK'})
209
210        elif msg_type == 'STOP_UPLOAD':
211            if self.on_stop_upload:
212                try:
213                    self.on_stop_upload()
214                except Exception:
215                    logger.exception("on_stop_upload callback failed")
216            self.send_message({'type': 'STOP_UPLOAD_ACK'})
217
218        elif msg_type == 'LIST_LOGS':
219            files = self._list_log_files()
220            self.send_message({'type': 'LIST_LOGS_RESPONSE', 'files': files})
221
222        else:
223            logger.debug(f"Unexpected message from host: {msg}")
224
225    def _clear_manifests(self) -> int:
226        """Delete all .uploaded_manifest files from log directories."""
227        from utils.match_monitor.log_uploader import LOG_DIRS, MANIFEST_FILENAME
228        count = 0
229        for log_dir in LOG_DIRS:
230            manifest = log_dir / MANIFEST_FILENAME
231            if manifest.exists():
232                try:
233                    manifest.unlink()
234                    count += 1
235                    logger.info(f"Deleted manifest: {manifest}")
236                except Exception:
237                    logger.exception(f"Failed to delete {manifest}")
238        return count
239
240    def _list_log_files(self) -> list:
241        """Scan log directories and return file info with upload status."""
242        from utils.match_monitor.log_uploader import LOG_DIRS, LOG_EXTENSIONS, MANIFEST_FILENAME
243        files = []
244        for log_dir in LOG_DIRS:
245            if not log_dir.is_dir():
246                continue
247            # Load manifest for this directory
248            manifest_path = log_dir / MANIFEST_FILENAME
249            manifest = set()
250            if manifest_path.exists():
251                try:
252                    manifest = set(manifest_path.read_text().strip().splitlines())
253                except Exception:
254                    logger.debug("Could not read manifest: %s", manifest_path, exc_info=True)
255            for f in sorted(log_dir.iterdir()):
256                if f.is_file() and f.suffix.lower() in LOG_EXTENSIONS:
257                    files.append({
258                        'name': f.name,
259                        'size': f.stat().st_size,
260                        'uploaded': f.name in manifest,
261                        'dir': str(log_dir),
262                    })
263        return files
264
265    @staticmethod
266    def _read_message(conn: socket.socket, buf: str) -> Optional[str]:
267        """Read data from socket until we have at least one newline. Returns updated buf or None on disconnect."""
268        while '\n' not in buf:
269            data = conn.recv(4096)
270            if not data:
271                return None
272            buf += data.decode('utf-8')
273        return buf
274
275    @staticmethod
276    def _parse_next_message(buf: str) -> tuple:
277        """Extract the next JSON message from the buffer. Returns (msg_dict, remaining_buf)."""
278        idx = buf.index('\n')
279        line = buf[:idx].strip()
280        remaining = buf[idx + 1:]
281        if not line:
282            return None, remaining
283        try:
284            return json.loads(line), remaining
285        except json.JSONDecodeError:
286            logger.warning(f"Invalid JSON from host: {line!r}")
287            return None, remaining

 34class LogUploader:
 35    """Uploads robot log files to a match_monitor receiver via HTTP.
 36
 37    The host IP and port are discovered via the ControlListener TCP channel.
 38    Uploads run in a background daemon thread triggered from disabledInit().
 39    Calling start_upload() when an upload is already running is a safe no-op.
 40    Call stop_upload() when leaving disabled mode to free bandwidth.
 41    """
 42
 43    upload_enabled = ntproperty('/LogUploader/enabled', True,
 44                                writeDefault=True, persistent=True)
 45    status = ntproperty('/LogUploader/status', 'idle',
 46                        writeDefault=True, persistent=False)
 47
 48    def __init__(self, control_listener: ControlListener) -> None:
 49        self._control = control_listener
 50        self._thread: Optional[threading.Thread] = None
 51        self._lock = threading.Lock()
 52        self._stop_event = threading.Event()
 53
 54    def start_upload(self) -> None:
 55        """Begin a non-blocking upload cycle. No-op if already uploading."""
 56        if not self.upload_enabled:
 57            self.status = 'disabled'
 58            logger.info("Upload disabled via NT, skipping")
 59            return
 60        if not self._control.is_connected:
 61            self.status = 'no host connected'
 62            logger.info("No host connected, skipping upload")
 63            return
 64
 65        host_ip = self._control.host_ip
 66        http_port = self._control.http_port
 67        logger.info("start_upload: host=%s:%s", host_ip, http_port)
 68
 69        with self._lock:
 70            if self._thread is not None and self._thread.is_alive():
 71                logger.info("Upload already in progress, skipping start_upload")
 72                return
 73            logger.info("Starting upload worker thread")
 74            self._stop_event.clear()
 75            self._thread = threading.Thread(
 76                target=self._upload_worker, daemon=True
 77            )
 78            self._thread.start()
 79
 80    def stop_upload(self) -> None:
 81        """Signal the upload thread to stop after the current file."""
 82        self._stop_event.set()
 83
 84    def stop_and_wait(self, timeout: float = 60) -> None:
 85        """Stop the upload thread and wait for it to finish."""
 86        self._stop_event.set()
 87        if self._thread is not None and self._thread.is_alive():
 88            self._thread.join(timeout=timeout)
 89
 90    def is_uploading(self) -> bool:
 91        return self._thread is not None and self._thread.is_alive()
 92
 93    def _upload_worker(self) -> None:
 94        try:
 95            host_ip = self._control.host_ip
 96            http_port = self._control.http_port
 97            if not host_ip or not http_port:
 98                self.status = 'no host connected'
 99                return
100
101            self.status = 'scanning'
102            log_dirs = self._find_log_dirs()
103            logger.info("Found log dirs: %s", log_dirs)
104            if not log_dirs:
105                self.status = 'no log directory found'
106                return
107
108            # Gather new files (not in manifest)
109            new_files: List[Path] = []
110            for log_dir in log_dirs:
111                new_files.extend(self._find_new_files(log_dir))
112
113            # Gather previously uploaded files that may have changed
114            manifest_files: List[Path] = []
115            for log_dir in log_dirs:
116                manifest_files.extend(self._find_manifest_files(log_dir))
117
118            logger.info("new=%d, manifest=%d", len(new_files), len(manifest_files))
119            if not new_files and not manifest_files:
120                self.status = 'idle (no files)'
121                return
122
123            self.status = 'connecting'
124            if not self._check_receiver(host_ip, http_port):
125                self.status = 'receiver unreachable'
126                logger.warning("Receiver unreachable at %s:%s", host_ip, http_port)
127                return
128
129            event_name, match_type, match_number = self._get_match_metadata()
130
131            # Notify host that uploads are starting
132            self._control.send_upload_starting(event_name, match_type, match_number)
133
134            uploaded_count = 0
135            total = len(new_files) + len(manifest_files)
136
137            # Upload new files first
138            for filepath in new_files:
139                if self._stop_event.is_set():
140                    self.status = f'stopped ({uploaded_count}/{total} uploaded)'
141                    return
142                # Re-check connection in case host disappeared
143                if not self._control.is_connected:
144                    self.status = 'host disconnected during upload'
145                    return
146                self.status = f'uploading {filepath.name}'
147                if self._upload_file(filepath, host_ip, http_port,
148                                     event_name, match_type, match_number):
149                    self._add_to_manifest(filepath)
150                    uploaded_count += 1
151
152            # Check previously uploaded files for changes (SHA-256 mismatch)
153            for filepath in manifest_files:
154                if self._stop_event.is_set():
155                    self.status = f'stopped ({uploaded_count}/{total} uploaded)'
156                    return
157                if not self._control.is_connected:
158                    self.status = 'host disconnected during upload'
159                    return
160                self.status = f'checking {filepath.name}'
161                if self._needs_reupload(filepath, host_ip, http_port,
162                                        event_name, match_type, match_number):
163                    self.status = f're-uploading {filepath.name}'
164                    self._upload_file(filepath, host_ip, http_port,
165                                      event_name, match_type, match_number,
166                                      overwrite=True)
167                    uploaded_count += 1
168
169            self.status = f'done ({uploaded_count}/{total} uploaded)'
170
171            # Notify host that uploads are complete
172            if not self._stop_event.is_set():
173                self._control.send_upload_complete(event_name, match_type, match_number)
174        except Exception:
175            logger.exception("Upload worker failed")
176            self.status = 'error'
177
178    def _find_log_dirs(self) -> List[Path]:
179        return [d for d in LOG_DIRS if d.is_dir()]
180
181    def _find_new_files(self, log_dir: Path) -> List[Path]:
182        manifest = self._load_manifest(log_dir)
183        files = []
184        for f in log_dir.iterdir():
185            if f.is_file() and f.suffix.lower() in LOG_EXTENSIONS and f.name not in manifest:
186                files.append(f)
187        files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
188        return files
189
190    def _find_manifest_files(self, log_dir: Path) -> List[Path]:
191        """Return files that are in the manifest and still exist on disk."""
192        manifest = self._load_manifest(log_dir)
193        files = []
194        for name in manifest:
195            f = log_dir / name
196            if f.is_file():
197                files.append(f)
198        files.sort(key=lambda f: f.stat().st_mtime, reverse=True)
199        return files
200
201    def _add_to_manifest(self, filepath: Path) -> None:
202        log_dir = filepath.parent
203        manifest = self._load_manifest(log_dir)
204        manifest.add(filepath.name)
205        self._save_manifest(log_dir, manifest)
206
207    def _load_manifest(self, log_dir: Path) -> Set[str]:
208        manifest_path = log_dir / MANIFEST_FILENAME
209        if not manifest_path.exists():
210            return set()
211        try:
212            return set(manifest_path.read_text().strip().splitlines())
213        except Exception:
214            logger.exception("Failed to read upload manifest")
215            return set()
216
217    def _save_manifest(self, log_dir: Path, manifest: Set[str]) -> None:
218        manifest_path = log_dir / MANIFEST_FILENAME
219        try:
220            manifest_path.write_text('\n'.join(sorted(manifest)) + '\n')
221        except Exception:
222            logger.exception("Failed to write upload manifest")
223
224    def _get_match_metadata(self) -> tuple:
225        try:
226            event_name = wpilib.DriverStation.getEventName()
227            match_type = wpilib.DriverStation.getMatchType()
228            match_number = wpilib.DriverStation.getMatchNumber()
229
230            type_names = {
231                wpilib.DriverStation.MatchType.kNone: '',
232                wpilib.DriverStation.MatchType.kPractice: 'practice',
233                wpilib.DriverStation.MatchType.kQualification: 'qual',
234                wpilib.DriverStation.MatchType.kElimination: 'elim',
235            }
236            match_type_str = type_names.get(match_type, '')
237            return event_name, match_type_str, str(match_number)
238        except Exception:
239            logger.exception("Failed to get match metadata")
240            return '', '', ''
241
242    def _check_receiver(self, host_ip: str, http_port: int) -> bool:
243        try:
244            conn = http.client.HTTPConnection(host_ip, http_port, timeout=3)
245            conn.request('GET', '/status')
246            resp = conn.getresponse()
247            conn.close()
248            return resp.status == 200
249        except Exception:
250            return False
251
252    def _needs_reupload(self, filepath: Path, host_ip: str, http_port: int,
253                        event_name: str, match_type: str,
254                        match_number: str) -> bool:
255        """Check if a previously uploaded file has changed by comparing SHA-256."""
256        try:
257            local_sha = hashlib.sha256(filepath.read_bytes()).hexdigest()
258
259            conn = http.client.HTTPConnection(host_ip, http_port, timeout=5)
260            headers = {
261                'X-Filename': filepath.name,
262                'X-Event-Name': event_name or '',
263                'X-Match-Type': match_type or '',
264                'X-Match-Number': match_number or '',
265            }
266            conn.request('GET', '/check', headers=headers)
267            resp = conn.getresponse()
268            body = json.loads(resp.read().decode('utf-8'))
269            conn.close()
270
271            if resp.status != 200:
272                return False
273
274            if not body.get('exists'):
275                return True
276
277            remote_sha = body.get('sha256', '')
278            if local_sha != remote_sha:
279                logger.info(f"SHA-256 mismatch for {filepath.name}, will re-upload")
280                return True
281
282            return False
283        except Exception:
284            logger.exception(f"Failed to check {filepath.name}")
285            return False
286
287    def _upload_file(self, filepath: Path, host_ip: str, http_port: int,
288                     event_name: str, match_type: str, match_number: str,
289                     overwrite: bool = False) -> bool:
290        try:
291            raw_data = filepath.read_bytes()
292            if len(raw_data) == 0:
293                logger.info(f"Skipping {filepath.name}: file is empty (0 bytes)")
294                self._control.send_message({
295                    'type': 'FILE_SKIPPED',
296                    'filename': filepath.name,
297                    'reason': 'empty (0 bytes)',
298                })
299                return True  # Treat as "handled" so it gets added to manifest
300            compressed_data = zlib.compress(raw_data)
301
302            conn = http.client.HTTPConnection(host_ip, http_port, timeout=30)
303            headers = {
304                'Content-Type': 'application/octet-stream',
305                'Content-Encoding': 'deflate',
306                'X-Filename': filepath.name,
307                'X-Event-Name': event_name or '',
308                'X-Match-Type': match_type or '',
309                'X-Match-Number': match_number or '',
310                'X-Uncompressed-Size': str(len(raw_data)),
311                'Content-Length': str(len(compressed_data)),
312            }
313            if overwrite:
314                headers['X-Overwrite'] = 'true'
315
316            start_time = time.monotonic()
317            conn.request('POST', '/upload', body=compressed_data, headers=headers)
318            resp = conn.getresponse()
319            conn.close()
320            elapsed = time.monotonic() - start_time
321            raw_mb = len(raw_data) / (1024 * 1024)
322            compressed_mb = len(compressed_data) / (1024 * 1024)
323            rate_mbps = compressed_mb / elapsed if elapsed > 0 else 0
324            savings = (1 - len(compressed_data) / len(raw_data)) * 100 if raw_data else 0
325
326            if resp.status == 200:
327                action = "Re-uploaded" if overwrite else "Uploaded"
328                logger.info(f"{action} {filepath.name} "
329                            f"({raw_mb:.1f} MB → {compressed_mb:.1f} MB compressed, "
330                            f"{savings:.0f}% savings, {elapsed:.2f}s, {rate_mbps:.1f} MB/s)")
331                return True
332            elif resp.status == 409:
333                logger.info(f"Already exists on receiver: {filepath.name}")
334                return True
335            else:
336                logger.warning(
337                    f"Upload failed for {filepath.name}: HTTP {resp.status}"
338                )
339                return False
340        except Exception:
341            logger.exception(f"Upload failed for {filepath.name}")
342            return False