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

Usage

This is an async library: every scanner call uses await, so it must run inside an async def you start with asyncio.run(...). (await at the top level of a script raises SyntaxError: 'await' outside function.)

Quickstart — no hardware (mock mode)

Runs as-is, no scanner connected. mode="mock" swaps in in-memory scanners; mock_label_probability=1.0 makes every scan return a label so the demo is deterministic.

import asyncio

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"),
    ],
    mock_label_probability=1.0,  # always "read" a label (drop to taste, e.g. 0.3)
    mock_labels=["GT-205"],
)
service = ScannerService.from_config(config, mode="mock")


async def main():
    await service.connect_all()
    result = await service.scan("station-a")  # one scan window open→dwell→close
    await service.disconnect_all()

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


asyncio.run(main())  # runs everything in main()

Connecting real hardware

Same code — swap mode="mock" for mode="real" and pick the backend. host is each scanner's IP; location_id is the name you scan by.

service = ScannerService.from_config(config, mode="real", scanner_type="keyence")
#                                          ^ "keyence" (TCP) | "evdev" (USB, Linux)

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.

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