Intis/coin-counter-simulators
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Better logging

Browse Source
This commit is contained in:
Eden Kirin
2025-10-14 08:20:19 +02:00
parent daa7e53d6d
commit 9d0557f96d
5 changed files with 143 additions and 141 deletions
Download Patch File Download Diff File Expand all files Collapse all files

117
source/pelican.py
View File

@ -4,13 +4,12 @@ Pelican Device Simulator
Simulates a Pelican coin counting device for development/testing purposes.
"""
import argparse
import logging
import random
import time
from typing import Optional
import serial
from loguru import logger
# Protocol constants
STX = 0x02
@ -42,7 +41,7 @@ PASSWORD = "69390274"
class PelicanSimulator:
def __init__(self, logger: logging.Logger, port: str, baudrate: int = 9600):
def __init__(self, port: str, baudrate: int = 9600):
self.port = port
self.baudrate = baudrate
self.serial_conn: Optional[serial.Serial] = None
@ -79,8 +78,6 @@ class PelicanSimulator:
200, # 2.00
]
self.logger = logger
def calc_crc(self, data: bytes) -> int:
"""Calculate CRC-16 using the algorithm from the spec"""
crc = 0
@ -107,29 +104,29 @@ class PelicanSimulator:
def parse_message(self, message: bytes) -> Optional[bytes]:
"""Parse and validate a received message"""
if len(message) < 5:
self.logger.error("Message too short")
logger.error("Message too short")
return None
if message[0] != STX:
self.logger.error("Invalid STX")
logger.error("Invalid STX")
return None
if message[-1] != ETX:
self.logger.error("Invalid ETX")
logger.error("Invalid ETX")
return None
length = message[1]
data = message[2 : 2 + length]
if len(data) != length:
self.logger.error("Length mismatch")
logger.error("Length mismatch")
return None
crc_received = (message[2 + length] << 8) | message[2 + length + 1]
crc_calculated = self.calc_crc(data)
if crc_received != crc_calculated:
self.logger.error(
logger.error(
f"CRC mismatch: received {crc_received:04X}, calculated {crc_calculated:04X}"
)
return None
@ -145,17 +142,17 @@ class PelicanSimulator:
if password == PASSWORD:
self.link_established = True
self.logger.info("Link established successfully")
logger.info("Link established successfully")
return self.create_message(bytes([CMD_RESPONSE_CONSTRUCT_LINK, 0x00]))
else:
self.logger.warning(f"Invalid password: {password}")
logger.warning(f"Invalid password: {password}")
return self.create_message(bytes([CMD_RESPONSE_CONSTRUCT_LINK, 0x01]))
def handle_destruct_link(self, data: bytes) -> bytes:
"""Handle destruct link command (0x03)"""
if self.link_established:
self.link_established = False
self.logger.info("Link destructed")
logger.info("Link destructed")
return self.create_message(bytes([CMD_RESPONSE_DESTRUCT_LINK, 0x00]))
else:
return self.create_message(bytes([CMD_RESPONSE_DESTRUCT_LINK, 0x01]))
@ -163,19 +160,19 @@ class PelicanSimulator:
def handle_get_value(self, data: bytes) -> bytes:
"""Handle get value command (0x11)"""
if not self.link_established:
self.logger.warning("Get value request rejected - link not established")
logger.warning("Get value request rejected - link not established")
return self.create_message(bytes([CMD_VALUE_RETURNED, 0x01]))
if len(data) < 2:
self.logger.error("Get value request missing variable number")
logger.error("Get value request missing variable number")
return self.create_message(bytes([CMD_VALUE_RETURNED, 0x01]))
var_num = data[1]
self.logger.info(f"Get value request for variable 0x{var_num:02X}")
logger.info(f"Get value request for variable 0x{var_num:02X}")
# Variable 0x16 - Current counting result
if var_num == 0x16:
self.logger.info("Responding with current counting result")
logger.info("Responding with current counting result")
# Generate random current counts for simulation
random_counts = []
@ -218,7 +215,7 @@ class PelicanSimulator:
total_coins = sum(random_counts)
# Log detailed count information
self.logger.info(
logger.info(
f"Current count: {total_coins} total coins, {random_rejected} rejected"
)
@ -239,13 +236,13 @@ class PelicanSimulator:
count_details.append(f"{standard_denoms[i]}: {random_counts[i]}")
if count_details:
self.logger.info(f"Coin counts: {', '.join(count_details)}")
logger.info(f"Coin counts: {', '.join(count_details)}")
return self.create_message(bytes(response))
# Variable 0x1C - Total counting result
elif var_num == 0x1C:
self.logger.info("Responding with total counting result")
logger.info("Responding with total counting result")
response = [CMD_VALUE_RETURNED, 0x00, 0x1C]
for count in self.total_count:
response.extend(
@ -257,12 +254,12 @@ class PelicanSimulator:
]
)
total_coins = sum(self.total_count)
self.logger.info(f"Total count: {total_coins} total coins since last reset")
logger.info(f"Total count: {total_coins} total coins since last reset")
return self.create_message(bytes(response))
# Variable 0x31 - Software version
elif var_num == 0x31:
self.logger.info("Responding with software version information")
logger.info("Responding with software version information")
response = [CMD_VALUE_RETURNED, 0x00, 0x31]
# SW version
response.extend(
@ -291,14 +288,14 @@ class PelicanSimulator:
self.host_version & 0xFF,
]
)
self.logger.info(
logger.info(
f"SW Version: {self.sw_version:08X}, SW Code: {self.sw_code:08X}, Host Version: {self.host_version:08X}"
)
return self.create_message(bytes(response))
# Variable 0x33 - Machine status
elif var_num == 0x33:
self.logger.info("Responding with machine status")
logger.info("Responding with machine status")
status_flags_1 = 0x00
if self.motor_running:
status_flags_1 |= 0x01
@ -346,12 +343,12 @@ class PelicanSimulator:
status_desc.append("keyboard locked")
status_str = ", ".join(status_desc) if status_desc else "all unlocked"
self.logger.info(f"Machine status: {state_desc}, {status_str}")
logger.info(f"Machine status: {state_desc}, {status_str}")
return self.create_message(bytes(response))
# Variable 0x1F - Get denomination values
elif var_num == 0x1F:
self.logger.info("Responding with coin denomination values")
logger.info("Responding with coin denomination values")
response = [CMD_VALUE_RETURNED, 0x00, 0x1F]
# Add denomination values for all 20 coin types (4 bytes each)
for denomination in self.denominations:
@ -372,18 +369,18 @@ class PelicanSimulator:
else:
formatted_denoms.append(f"Coin{i+1}: 0.{denom:02d}")
self.logger.info(
logger.info(
f"Denomination values (20 coins): {', '.join(formatted_denoms[:8])}"
)
if len(formatted_denoms) > 8:
self.logger.info(
logger.info(
f"Additional denominations: {', '.join(formatted_denoms[8:])}"
)
return self.create_message(bytes(response))
# Variable 0x22 - Coin exit counters (similar to 0x16 but for sorted coins)
elif var_num == 0x22:
self.logger.info("Responding with coin exit counters")
logger.info("Responding with coin exit counters")
# Generate random exit counts for simulation
exit_counts = []
@ -407,7 +404,7 @@ class PelicanSimulator:
)
total_exits = sum(exit_counts)
self.logger.info(f"Coin exit counters: {total_exits} total coins sorted")
logger.info(f"Coin exit counters: {total_exits} total coins sorted")
# Log counts for standard denominations
standard_denoms = [
@ -426,7 +423,7 @@ class PelicanSimulator:
exit_details.append(f"{standard_denoms[i]}: {exit_counts[i]}")
if exit_details:
self.logger.info(f"Exit counts: {', '.join(exit_details)}")
logger.info(f"Exit counts: {', '.join(exit_details)}")
return self.create_message(bytes(response))
@ -434,13 +431,13 @@ class PelicanSimulator:
elif var_num == 0x23:
# Need transaction number (2 bytes) after variable number
if len(data) < 4:
self.logger.error(
logger.error(
"Get transaction data request missing transaction number"
)
return self.create_message(bytes([CMD_VALUE_RETURNED, 0x01]))
transaction_num = (data[2] << 8) | data[3]
self.logger.info(
logger.info(
f"Responding with transaction data for transaction #{transaction_num}"
)
@ -498,7 +495,7 @@ class PelicanSimulator:
# Note amount - currency 1 (4 bytes) / Account number low for CDS
response.extend([0x00, 0x00, 0x00, 0x00])
self.logger.info(
logger.info(
f"Transaction #{transaction_num}: Cashier {cashier_num}, "
f"Date {current_time.tm_year}/{current_time.tm_mon}/{current_time.tm_mday} "
f"{current_time.tm_hour:02d}:{current_time.tm_min:02d}, "
@ -508,7 +505,7 @@ class PelicanSimulator:
return self.create_message(bytes(response))
else:
self.logger.warning(f"Unknown variable number: 0x{var_num:02X}")
logger.warning(f"Unknown variable number: 0x{var_num:02X}")
return self.create_message(bytes([CMD_VALUE_RETURNED, 0x01]))
def handle_get_display(self, data: bytes) -> bytes:
@ -554,7 +551,7 @@ class PelicanSimulator:
new_line[position + i] = c
self.display_line2 = "".join(new_line)
self.logger.info(
logger.info(
f"Display updated: L1='{self.display_line1}' L2='{self.display_line2}'"
)
@ -569,7 +566,7 @@ class PelicanSimulator:
return self.create_message(bytes([CMD_RESPONSE_LOCK_DISPLAY, 0x01]))
self.display_locked = data[1] == 0x01
self.logger.info(f"Display {'locked' if self.display_locked else 'unlocked'}")
logger.info(f"Display {'locked' if self.display_locked else 'unlocked'}")
return self.create_message(bytes([CMD_RESPONSE_LOCK_DISPLAY, 0x00]))
@ -579,7 +576,7 @@ class PelicanSimulator:
return self.create_message(bytes([CMD_RESPONSE_LOCK_KEYBOARD, 0x01]))
self.keyboard_locked = data[1] == 0x01
self.logger.info(f"Keyboard {'locked' if self.keyboard_locked else 'unlocked'}")
logger.info(f"Keyboard {'locked' if self.keyboard_locked else 'unlocked'}")
return self.create_message(bytes([CMD_RESPONSE_LOCK_KEYBOARD, 0x00]))
@ -590,7 +587,7 @@ class PelicanSimulator:
cmd = data[0]
self.logger.info(f"Received command: 0x{cmd:02X}")
logger.info(f"Received command: 0x{cmd:02X}")
if cmd == CMD_CONSTRUCT_LINK:
return self.handle_construct_link(data)
@ -607,12 +604,12 @@ class PelicanSimulator:
elif cmd == CMD_LOCK_KEYBOARD:
return self.handle_lock_keyboard(data)
else:
self.logger.warning(f"Unknown command: 0x{cmd:02X}")
logger.warning(f"Unknown command: 0x{cmd:02X}")
return None
def run(self):
"""Main simulator loop"""
self.logger.info(
logger.info(
f"Starting Pelican simulator on {self.port} at {self.baudrate} baud"
)
@ -626,7 +623,7 @@ class PelicanSimulator:
timeout=1,
)
self.logger.info("Serial port opened successfully")
logger.info("Serial port opened successfully")
buffer = bytearray()
@ -643,14 +640,18 @@ class PelicanSimulator:
message = bytes(buffer[stx_idx : etx_idx + 1])
buffer = buffer[etx_idx + 1 :]
self.logger.debug(f"RX: {' '.join(f'{b:02X}' for b in message)}")
logger.debug(
f"RX: {' '.join(f'{b:02X}' for b in message)}"
)
# Parse and handle message
parsed_data = self.parse_message(message)
if parsed_data:
response = self.handle_command(parsed_data)
if response:
self.logger.debug(f"TX: {' '.join(f'{b:02X}' for b in response)}")
logger.debug(
f"TX: {' '.join(f'{b:02X}' for b in response)}"
)
self.serial_conn.write(response)
except ValueError:
pass # ETX not found yet
@ -658,32 +659,10 @@ class PelicanSimulator:
time.sleep(0.01)
except KeyboardInterrupt:
self.logger.info("Simulator stopped by user")
logger.info("Simulator stopped by user")
except Exception as e:
self.logger.error(f"Error: {e}", exc_info=True)
logger.error(f"Error: {e}", exc_info=True)
finally:
if self.serial_conn and self.serial_conn.is_open:
self.serial_conn.close()
self.logger.info("Serial port closed")
def main():
parser = argparse.ArgumentParser(description="Pelican Device Simulator")
parser.add_argument(
"--port",
"-p",
default="/dev/ttyUSB0",
help="Serial port (default: /dev/ttyUSB0)",
)
parser.add_argument(
"--baudrate", "-b", type=int, default=9600, help="Baud rate (default: 9600)"
)
args = parser.parse_args()
simulator = PelicanSimulator(args.port, args.baudrate)
simulator.run()
if __name__ == "__main__":
main()
logger.info("Serial port closed")