vention-barcode-scanner 0.1.0

Async barcode scanner fleet manager for Vention industrial automation. Supports Keyence TCP, USB evdev, and mock scanners.

vention-barcode-scanner

Async barcode scanner fleet manager for Vention industrial automation. Unified Scanner ABC over Keyence TCP, USB HID (evdev), and mock backends.

Scanner Types

Type	Protocol	Platform	Use Case
KeyenceScanner	Async TCP (port 9004)	Any	Keyence SR-series (SR-1000, SR-2000, SR-5000)
EvdevScanner	Linux USB (evdev)	Linux	USB keyboard-emulating barcode readers
MockScanner	In-memory	Any	Simulation and testing

Usage

from barcode_scanner.models import ScannerConfig, ScannerFleetConfig
from barcode_scanner.service import ScannerService

config = ScannerFleetConfig(
    scanners=[
        ScannerConfig(id="s1", host="192.168.7.100", location_id="station-a"),
        ScannerConfig(id="s2", host="192.168.7.101", location_id="station-b"),
    ],
)
service = ScannerService.from_config(config, mode="real", scanner_type="keyence")

await service.connect_all()
result = await service.scan("station-a")
await service.disconnect_all()

result.status          # ScanStatus.OK | NO_READ | TIMEOUT | ERROR
result.label_detected  # True / False
result.label_string    # "GT-205" or None

Label Validation

Optional callback to reject invalid labels at the scanner layer:

def validate_label(label: str) -> bool:
    return label.startswith("GT-") and len(label) <= 20

service = ScannerService.from_config(config, label_validator=validate_label)
# Labels that fail validation are downgraded to NO_READ

Metrics

metrics = service.get_metrics()["s1"]
metrics.total_scans      # 1234
metrics.success_rate     # 0.89
metrics.avg_scan_ms      # 45.2

Tuning and Diagnostics

await service.auto_tune("station-a")           # One-shot FTUNE + SAVE
await service.tune("station-a", bank=0)         # Interactive TUNE per bank + TQUIT
status = await service.get_scanner_status("station-a")
# {"busy": 0, "last_cmd": 0, "error": 0}       # BUSYSTAT / CMDSTAT / ERRSTAT

Structured Logging

from barcode_scanner.logger import set_log_callback

def on_scanner_log(code, source, level, message):
    mqtt_publish("scanner/logs", {"code": code, "source": source, "message": message})

set_log_callback(on_scanner_log)

Keyence Protocol

TCP port 9004. ASCII commands terminated with \r.

Scan sequence: LON\r (open scan window) -> dwell -> LOFF\r (close window) -> read response. The scanner only sends data after LOFF.

Command	Response	Purpose
LON\r	barcode or ERROR\r	Open scan window (default bank)
LON,01\r	barcode or ERROR\r	Open scan window (bank 1)
LOFF\r	--	Close scan window
BCLR\r	OK\r	Clear read buffer
RESET\r	OK\r	Reset scanner (reboots, drops TCP)
FTUNE\r	OK,FTUNE\r then result	One-shot auto-focus calibration
TUNE,00\r	OK\r	Start interactive tuning (bank 0)
TQUIT\r	OK\r	Stop tuning and save
SAVE\r	OK\r	Persist settings across power cycles
BUSYSTAT\r	0/1/2	Query busy state (idle/reading/tuning)
CMDSTAT\r	0/1/2	Query last command result
ERRSTAT\r	0/1/2	Query hardware error state

API

ScannerService

class ScannerService:
    @classmethod
    def from_config(cls, config, mode="real", scanner_type="keyence",
                    label_validator=None) -> ScannerService

    async def scan(self, location_id, timeout=None, bank=None) -> ScanResult
    async def connect_all(self) -> dict[str, bool]
    async def disconnect_all(self) -> None
    async def reset(self, location_id) -> bool
    async def reset_all(self) -> dict[str, bool]
    async def auto_tune(self, location_id) -> bool
    async def tune(self, location_id, bank=0) -> bool
    async def get_scanner_status(self, location_id) -> dict
    def get_metrics(self) -> dict[str, ScanMetrics]
    def get_metrics_summary(self) -> dict

ScanResult

class ScanResult:
    status: ScanStatus          # OK | NO_READ | TIMEOUT | ERROR
    label_detected: bool
    label_string: str | None
    scanner_id: str
    error: str | None

Scanner ABC

class Scanner(ABC):
    async def scan(self, timeout=None, bank=None) -> ScanResult
    async def connect(self) -> bool
    async def disconnect(self) -> None
    async def reset(self) -> bool
    async def auto_tune(self) -> bool
    async def tune(self, bank=0) -> bool
    async def get_status(self) -> dict
    connected: bool  # property

Development

cd barcode-scanner
uv sync
make test          # 115 tests
make lint          # ruff check + format
make type-check    # pyright