Metadata-Version: 2.4
Name: vention-barcode-scanner
Version: 0.6.0
Summary: Async barcode scanner fleet manager for Vention industrial automation. Supports Keyence TCP, USB evdev, and mock scanners.
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Requires-Python: >=3.10
Requires-Dist: pydantic>=2.0
Requires-Dist: pyyaml>=6.0
Description-Content-Type: text/markdown

# 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

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.

```python
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.

```python
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:

```python
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

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

### Tuning and Diagnostics

```python
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

```python
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

```python
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

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

### Scanner ABC

```python
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

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