""" Serial Bridge - SSE Server Reads from COM port and streams data to React frontend via Server-Sent Events No backend storage - direct streaming only """ import os import logging from logging.handlers import RotatingFileHandler import serial import serial.tools.list_ports from dotenv import load_dotenv from threading import Thread, Lock from datetime import datetime from flask import Flask, Response, request from flask_cors import CORS, cross_origin from queue import Queue import json import time # Load environment variables load_dotenv() # Configure logging # Log next to this file, not the working directory, so starting the bridge from # different places does not scatter logs around. Rotate so it cannot grow without # bound - individual readings are DEBUG (see [RX] below), events are INFO. _LOG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'serial_bridge.log') logging.basicConfig( level=os.getenv('LOG_LEVEL', 'INFO'), format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', handlers=[ RotatingFileHandler(_LOG_PATH, maxBytes=2 * 1024 * 1024, backupCount=3, encoding='utf-8'), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) app = Flask(__name__) CORS(app) # Enable CORS for all routes # Bridge version - reported via /status so stale stations are visible centrally BRIDGE_VERSION = '1.1.0' # Where the last poll setting is remembered, so a bridge restarted with no # browser open resumes polling instead of sitting idle. POLL_CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'poll_config.json') # Global state data_queue = Queue() connected_clients = [] is_running = False reader_thread = None # Poll settings pushed by the UI. An empty command means the scale streams on # its own and nothing should be written to it. The bridge never interprets the # command - it only writes the bytes. poll_config = {'command': '', 'interval': 0.2} poll_lock = Lock() serial_write_lock = Lock() active_reader = None def load_poll_config(): """Restore the last poll setting saved by save_poll_config().""" global poll_config try: if os.path.exists(POLL_CONFIG_PATH): with open(POLL_CONFIG_PATH, 'r', encoding='utf-8') as fh: stored = json.load(fh) poll_config = { 'command': str(stored.get('command', '')), 'interval': float(stored.get('interval', 0.2)), } logger.info(f"[POLL] Restored: {poll_config}") except Exception as e: logger.error(f"Could not read {POLL_CONFIG_PATH}: {e}") def save_poll_config(): try: with open(POLL_CONFIG_PATH, 'w', encoding='utf-8') as fh: json.dump(poll_config, fh) except Exception as e: logger.error(f"Could not write {POLL_CONFIG_PATH}: {e}") class SerialReader: def __init__(self, port, baudrate, timeout): self.port = port self.baudrate = baudrate self.timeout = timeout self.ser = None self.is_connected = False def connect(self): try: self.ser = serial.Serial( port=self.port, baudrate=self.baudrate, timeout=self.timeout ) self.is_connected = True logger.info(f"[OK] Connected to {self.port} at {self.baudrate} baud") return True except Exception as e: logger.error(f"[ERROR] Failed to connect to {self.port}: {e}") self.is_connected = False return False def read_data(self): """Return the most recent reading, discarding any older buffered ones. The scale streams faster than this loop runs, so lines accumulate in the OS buffer. Only the latest weight is meaningful - returning the oldest would make the display fall further behind the scale over time. """ if not self.is_connected: return None try: raw = None while self.ser.in_waiting: candidate = self.ser.readline() if candidate.strip(): raw = candidate if raw: # 'data' is a lossy text view for convenience; 'hex' is the real # bytes. Binary protocols (Toledo sends STX and bit-packed status) # must be parsed from 'hex' - decoding them as UTF-8 corrupts them. return { 'port': self.port, 'data': raw.decode('utf-8', errors='ignore').rstrip('\r\n'), 'hex': raw.hex(), 'timestamp': datetime.now().isoformat(), 'received_at': datetime.now().strftime('%Y-%m-%d %H:%M:%S') } except Exception as e: logger.error(f"Error reading from {self.port}: {e}") return None def write_command(self, command: str) -> bool: """Write a command verbatim - no terminator is appended. The original Delphi used TComPort.WriteStr('%p'), which sends exactly those bytes, so anything extra would not be recognised by the scale. """ if not self.is_connected or not self.ser: return False try: with serial_write_lock: self.ser.write(command.encode('utf-8')) return True except Exception as e: logger.error(f"Error writing to {self.port}: {e}") return False def disconnect(self): if self.ser: self.ser.close() self.is_connected = False logger.info(f"[OK] Disconnected from {self.port}") def read_serial_data(): """Continuously read from serial port and queue data""" global is_running global active_reader reader = SerialReader( port=os.getenv('COM_PORT', 'COM3'), baudrate=int(os.getenv('BAUD_RATE', '9600')), timeout=float(os.getenv('TIMEOUT', '1')) ) if not reader.connect(): logger.error("Failed to connect to COM port") return active_reader = reader read_interval = float(os.getenv('READ_INTERVAL', '0.5')) while is_running: try: with poll_lock: command = poll_config['command'] interval = poll_config['interval'] if command: # Polled scale: it stays silent until asked. Give it a moment to # answer before reading, or the reply lands on the next pass. reader.write_command(command) time.sleep(min(0.1, interval)) data = reader.read_data() if data: # Keep only the newest reading - a queued older weight is stale # by the time a client reads it. while not data_queue.empty(): try: data_queue.get_nowait() except Exception: break data_queue.put(data) # Every reading, 5x a second - only useful when probing a scale. logger.debug(f"[RX] {data['port']}: {data['data']}") time.sleep(interval if command else read_interval) except KeyboardInterrupt: logger.info("Serial reader interrupted") break except Exception as e: logger.error(f"Error in serial reader: {e}") reader.disconnect() @app.route('/events') @cross_origin() def events(): """SSE endpoint - streams serial data to connected clients""" def generate(): client_id = id(object()) connected_clients.append(client_id) logger.info(f"[CONNECT] Client connected. Total: {len(connected_clients)}") try: # Send connection message yield f"data: {json.dumps({'status': 'connected', 'message': 'Connected to serial bridge'})}\n\n" # Stream data from queue while is_running: try: # Get data with timeout to allow graceful shutdown data = data_queue.get(timeout=1) yield f"data: {json.dumps(data)}\n\n" logger.debug(f"[TX] Sent: {data['data']}") except: # Queue timeout - send heartbeat yield f": heartbeat\n\n" except GeneratorExit: # Client disconnected pass finally: if client_id in connected_clients: connected_clients.remove(client_id) logger.info(f"[DISCONNECT] Client disconnected. Total: {len(connected_clients)}") return Response( generate(), mimetype='text/event-stream', headers={ 'Cache-Control': 'no-cache', 'X-Accel-Buffering': 'no', 'Connection': 'keep-alive' } ) @app.route('/poll', methods=['GET', 'POST']) @cross_origin() def poll(): """Get or set the poll command the bridge sends on its own timer. The UI pushes this from the station's protocol definition, so a scale with a different command needs no change to this program. An empty command puts the bridge in continuous mode, where it only listens. """ global poll_config if request.method == 'GET': with poll_lock: return dict(poll_config) payload = request.get_json(silent=True) or {} command = payload.get('command', '') interval = payload.get('interval', 0.2) if not isinstance(command, str): return {'error': 'command must be a string'}, 400 try: interval = float(interval) except (TypeError, ValueError): return {'error': 'interval must be a number'}, 400 if interval <= 0: return {'error': 'interval must be greater than zero'}, 400 with poll_lock: poll_config = {'command': command, 'interval': interval} save_poll_config() logger.info(f"[POLL] Set to {poll_config}") return dict(poll_config) @app.route('/send', methods=['POST']) @cross_origin() def send(): """Write one command to the port without changing the poll loop. Used by the admin monitor to probe an unknown scale. """ payload = request.get_json(silent=True) or {} command = payload.get('command', '') if not isinstance(command, str) or not command: return {'error': 'command is required'}, 400 if active_reader is None or not active_reader.is_connected: return {'error': 'serial port not connected'}, 503 if not active_reader.write_command(command): return {'error': 'write failed'}, 500 logger.info(f"[TX] Sent command: {command!r}") return {'sent': command} @app.route('/status') def status(): """Get current status""" with poll_lock: current_poll = dict(poll_config) return { 'status': 'running' if is_running else 'stopped', 'version': BRIDGE_VERSION, 'connected_clients': len(connected_clients), 'com_port': os.getenv('COM_PORT', 'COM3'), 'baud_rate': int(os.getenv('BAUD_RATE', '9600')), 'serial_connected': bool(active_reader and active_reader.is_connected), 'poll': current_poll } @app.route('/ports') def list_ports(): """List available COM ports""" ports = [] for port, desc, hwid in serial.tools.list_ports.comports(): ports.append({ 'port': port, 'description': desc, 'hwid': hwid }) return {'ports': ports} def start_app(): """Start the application""" global is_running, reader_thread logger.info("=" * 60) logger.info("ScalesApp Serial Bridge - SSE Streaming") logger.info("=" * 60) logger.info(f"COM Port: {os.getenv('COM_PORT', 'COM3')}") logger.info(f"Baud Rate: {os.getenv('BAUD_RATE', '9600')}") logger.info(f"SSE Endpoint: http://127.0.0.1:5000/events") logger.info("=" * 60) load_poll_config() is_running = True # Start serial reader thread reader_thread = Thread(target=read_serial_data, daemon=True) reader_thread.start() logger.info("[OK] Serial reader thread started\n") # Start Flask app debug = os.getenv('DEBUG', 'False').lower() == 'true' host = os.getenv('HOST', '127.0.0.1') app.run(host=host, port=5000, debug=debug, use_reloader=False) def stop_app(): """Stop the application""" global is_running is_running = False logger.info("\n[OK] Application stopped") if __name__ == '__main__': try: start_app() except KeyboardInterrupt: stop_app()