gitea
5a8527e96e
The bridge becomes pure transport: it opens the port, writes a poll string it is given at runtime, and streams raw lines. Protocol definitions live in Django and parsing happens in the browser, so a new scale type never means a new binary on the station PCs. The poll loop stays in the bridge because a browser cannot be the heartbeat - hidden tabs get throttled and a closed tab would stop the scale. Every payload now carries a hex field. Toledo Continuous sends STX and bit-packed status, which UTF-8 decoding silently corrupts. Fixed-column protocols need the opposite care: a space can mean balanced, so the trailing whitespace must survive. Parsers for the simulator, fixed-column, Toledo Continuous and SICS return an explicit error rather than a guess on a frame that does not match, since a silently wrong weight on a weighbridge is worse than no weight. Toledo and SICS are written from documented structure and have not been validated against real hardware. The admin monitor shows the raw stream as text and hex, sends one-off probe commands, and builds fixed-column protocols against captured lines. Spaces render as middots because in some protocols a space is the stability flag. The test writer now simulates a weighing cycle - idle, ramp, balanced, drop - alternating empty and loaded trucks, quantised to the 20kg and 50kg divisions real truck scales use. Bridge logging rotates and per-reading lines drop to DEBUG; the old config had grown two log files totalling 368MB. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
292 lines
10 KiB
Python
292 lines
10 KiB
Python
"""
|
|
COM Port Test Writer - Simulates data being sent to a COM port
|
|
"""
|
|
import serial
|
|
import time
|
|
import random
|
|
import logging
|
|
import sys
|
|
from typing import Optional
|
|
from config import (
|
|
COM_PORT, BAUD_RATE, TIMEOUT, WRITE_INTERVAL, TEST_DATA_TYPE,
|
|
EMPTY_MIN, EMPTY_MAX, FULL_MIN, FULL_MAX,
|
|
IDLE_SECONDS, RAMP_SECONDS, STABLE_SECONDS, DROP_SECONDS,
|
|
DIVISION_SMALL, DIVISION_LARGE, DIVISION_THRESHOLD, DEBUG
|
|
)
|
|
|
|
# Setup logging
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if DEBUG else logging.INFO,
|
|
# format='%(asctime)s - %(levelname)s - %(message)s'
|
|
format='%(message)s'
|
|
)
|
|
logger = logging.getLogger(__name__)
|
|
|
|
class ComPortTestWriter:
|
|
def __init__(self, port: str, baudrate: int = 9600, timeout: int = 1):
|
|
self.port = port
|
|
self.baudrate = baudrate
|
|
self.timeout = timeout
|
|
self.serial_conn: Optional[serial.Serial] = None
|
|
self.is_connected = False
|
|
self.is_running = False
|
|
self.data_counter = 0
|
|
# Weighing cycle state
|
|
self.phase = 'idle'
|
|
self.phase_started = time.monotonic()
|
|
self.loaded = False # alternates: empty truck, then full truck
|
|
self.target = self._pick_target()
|
|
self.phase_start_value = 0.0
|
|
|
|
def connect(self) -> bool:
|
|
"""Connect to the serial port"""
|
|
try:
|
|
self.serial_conn = serial.Serial(
|
|
port=self.port,
|
|
baudrate=self.baudrate,
|
|
timeout=self.timeout
|
|
)
|
|
self.is_connected = True
|
|
logger.info(f"✓ Connected to {self.port} at {self.baudrate} baud")
|
|
return True
|
|
except serial.SerialException as e:
|
|
logger.error(f"✗ Failed to connect to {self.port}: {e}")
|
|
logger.info("\nTip: Check that the COM port exists and is not in use.")
|
|
logger.info("You can list available ports with: python -m serial.tools.list_ports")
|
|
self.is_connected = False
|
|
return False
|
|
|
|
def disconnect(self):
|
|
"""Disconnect from the serial port"""
|
|
if self.serial_conn and self.serial_conn.is_open:
|
|
self.serial_conn.close()
|
|
self.is_connected = False
|
|
logger.info(f"✓ Disconnected from {self.port}")
|
|
|
|
def write_data(self, data: str) -> bool:
|
|
"""Write data to the serial port"""
|
|
try:
|
|
if self.is_connected and self.serial_conn:
|
|
# Add newline if not present
|
|
if not data.endswith('\n'):
|
|
data += '\n'
|
|
|
|
self.serial_conn.write(data.encode('utf-8'))
|
|
logger.info(f"→ Sent: {data.strip()}")
|
|
return True
|
|
except Exception as e:
|
|
logger.error(f"Error writing to {self.port}: {e}")
|
|
self.is_connected = False
|
|
return False
|
|
|
|
@staticmethod
|
|
def _quantise(value: float) -> int:
|
|
"""Snap a weight to the scale's division.
|
|
|
|
Truck scales are multi-interval: they read in DIVISION_SMALL steps up to
|
|
DIVISION_THRESHOLD and DIVISION_LARGE steps above it. Nothing in between
|
|
is displayable.
|
|
"""
|
|
if value <= 0:
|
|
return 0
|
|
division = DIVISION_SMALL if value < DIVISION_THRESHOLD else DIVISION_LARGE
|
|
# floor(x + 0.5), not round() - Python rounds halves to even, so a reading
|
|
# exactly between two divisions would sometimes snap down.
|
|
return int((value + division / 2) // division) * division
|
|
|
|
def _pick_target(self) -> int:
|
|
"""Pick the weight for the next truck on the scale."""
|
|
if self.loaded:
|
|
return self._quantise(random.randint(FULL_MIN, FULL_MAX))
|
|
return self._quantise(random.randint(EMPTY_MIN, EMPTY_MAX))
|
|
|
|
def _advance_phase(self):
|
|
"""Move to the next phase of the weighing cycle."""
|
|
if self.phase == 'idle':
|
|
self.phase = 'ramp'
|
|
elif self.phase == 'ramp':
|
|
self.phase = 'stable'
|
|
elif self.phase == 'stable':
|
|
self.phase = 'drop'
|
|
else:
|
|
# Truck left the scale - next one is the other kind
|
|
self.phase = 'idle'
|
|
self.loaded = not self.loaded
|
|
self.target = self._pick_target()
|
|
self.phase_started = time.monotonic()
|
|
|
|
def generate_scales_data(self) -> str:
|
|
"""Simulate a truck being weighed: idle at 0, drive on, settle, drive off.
|
|
|
|
Emits "<weight> kg ST" while the reading is stable (balanced), "US" otherwise,
|
|
mirroring how real indicators report stability.
|
|
"""
|
|
durations = {
|
|
'idle': IDLE_SECONDS,
|
|
'ramp': RAMP_SECONDS,
|
|
'stable': STABLE_SECONDS,
|
|
'drop': DROP_SECONDS,
|
|
}
|
|
# Roll over any fully elapsed phases first, so the value below is always
|
|
# computed from the phase it actually belongs to.
|
|
elapsed = time.monotonic() - self.phase_started
|
|
while elapsed >= durations[self.phase]:
|
|
self._advance_phase()
|
|
elapsed = 0.0
|
|
|
|
duration = durations[self.phase]
|
|
progress = min(1.0, elapsed / duration) if duration > 0 else 1.0
|
|
|
|
if self.phase == 'idle':
|
|
value, balanced = 0.0, False
|
|
elif self.phase == 'ramp':
|
|
# Climbing under load - noisy and unstable
|
|
value = self.target * min(1.0, progress + 0.05) + random.uniform(-40, 40)
|
|
value = max(0.0, value)
|
|
balanced = False
|
|
elif self.phase == 'stable':
|
|
# Settled on the target; drift is small enough that it usually rests
|
|
# on one division and only occasionally flickers to the next.
|
|
value = self.target + random.uniform(-12, 12)
|
|
balanced = True
|
|
else: # drop
|
|
value = self.target * (1.0 - progress) + random.uniform(-40, 40)
|
|
value = max(0.0, value)
|
|
balanced = False
|
|
|
|
return f"{self._quantise(value)} kg {'ST' if balanced else 'US'}"
|
|
|
|
def generate_counter_data(self) -> str:
|
|
"""Generate incrementing counter data"""
|
|
self.data_counter += 1
|
|
return f"Count: {self.data_counter}"
|
|
|
|
def generate_random_data(self) -> str:
|
|
"""Generate random numeric data"""
|
|
value = random.uniform(10000, 50000)
|
|
return f"Value: {value:.2f}"
|
|
|
|
def generate_mixed_data(self) -> str:
|
|
"""Generate mixed sensor data"""
|
|
temperature = random.uniform(20, 30)
|
|
humidity = random.uniform(30, 70)
|
|
pressure = random.uniform(1010, 1020)
|
|
return f"T:{temperature:.1f}C H:{humidity:.1f}% P:{pressure:.1f}hPa"
|
|
|
|
def get_next_data(self) -> str:
|
|
"""Get the next data to send based on test type"""
|
|
if TEST_DATA_TYPE == 'scales':
|
|
return self.generate_scales_data()
|
|
elif TEST_DATA_TYPE == 'counter':
|
|
return self.generate_counter_data()
|
|
elif TEST_DATA_TYPE == 'random':
|
|
return self.generate_random_data()
|
|
elif TEST_DATA_TYPE == 'mixed':
|
|
return self.generate_mixed_data()
|
|
else:
|
|
return self.generate_scales_data()
|
|
|
|
def run(self):
|
|
"""Run the test writer"""
|
|
logger.info(f"\n{'='*50}")
|
|
logger.info(f"COM Port Test Writer")
|
|
logger.info(f"{'='*50}")
|
|
logger.info(f"Port: {self.port}")
|
|
logger.info(f"Baud Rate: {self.baudrate}")
|
|
logger.info(f"Data Type: {TEST_DATA_TYPE}")
|
|
logger.info(f"Write Interval: {WRITE_INTERVAL}s")
|
|
if TEST_DATA_TYPE == 'scales':
|
|
logger.info(f"Cycle: idle {IDLE_SECONDS}s -> ramp {RAMP_SECONDS}s -> stable {STABLE_SECONDS}s -> drop {DROP_SECONDS}s")
|
|
logger.info(f"Empty truck: {EMPTY_MIN}-{EMPTY_MAX} kg | Full truck: {FULL_MIN}-{FULL_MAX} kg")
|
|
logger.info(f"{'='*50}\n")
|
|
|
|
if not self.connect():
|
|
logger.error("Failed to connect to COM port. Exiting.")
|
|
return
|
|
|
|
self.is_running = True
|
|
|
|
try:
|
|
logger.info("Press Ctrl+C to stop writing data\n")
|
|
|
|
while self.is_running:
|
|
try:
|
|
data = self.get_next_data()
|
|
self.write_data(data)
|
|
time.sleep(WRITE_INTERVAL)
|
|
|
|
except KeyboardInterrupt:
|
|
break
|
|
except Exception as e:
|
|
logger.error(f"Error in write loop: {e}")
|
|
break
|
|
|
|
except KeyboardInterrupt:
|
|
pass
|
|
finally:
|
|
self.is_running = False
|
|
self.disconnect()
|
|
logger.info("\n✓ Test writer stopped")
|
|
|
|
def list_available_ports():
|
|
"""List available COM ports"""
|
|
try:
|
|
from serial.tools import list_ports
|
|
ports = list_ports.comports()
|
|
if ports:
|
|
logger.info("\nAvailable COM ports:")
|
|
for port in ports:
|
|
logger.info(f" {port.device} - {port.description}")
|
|
else:
|
|
logger.info("No COM ports found")
|
|
except Exception as e:
|
|
logger.error(f"Error listing ports: {e}")
|
|
|
|
def main():
|
|
"""Main entry point"""
|
|
import argparse
|
|
|
|
parser = argparse.ArgumentParser(
|
|
description='Test COM port writer - continuously writes data to a COM port'
|
|
)
|
|
parser.add_argument(
|
|
'--port',
|
|
default=COM_PORT,
|
|
help=f'COM port to write to (default: {COM_PORT})'
|
|
)
|
|
parser.add_argument(
|
|
'--baud',
|
|
type=int,
|
|
default=BAUD_RATE,
|
|
help=f'Baud rate (default: {BAUD_RATE})'
|
|
)
|
|
parser.add_argument(
|
|
'--type',
|
|
choices=['scales', 'counter', 'random', 'mixed'],
|
|
default=TEST_DATA_TYPE,
|
|
help=f'Type of test data (default: {TEST_DATA_TYPE})'
|
|
)
|
|
parser.add_argument(
|
|
'--interval',
|
|
type=float,
|
|
default=WRITE_INTERVAL,
|
|
help=f'Write interval in seconds (default: {WRITE_INTERVAL})'
|
|
)
|
|
parser.add_argument(
|
|
'--list',
|
|
action='store_true',
|
|
help='List available COM ports and exit'
|
|
)
|
|
|
|
args = parser.parse_args()
|
|
|
|
if args.list:
|
|
list_available_ports()
|
|
return
|
|
|
|
writer = ComPortTestWriter(args.port, args.baud)
|
|
writer.run()
|
|
|
|
if __name__ == '__main__':
|
|
main()
|