vention-popup-diverter

Async popup roller diverter fleet manager for Vention industrial automation. Unified PopupDiverter ABC over Flowsort X-Flow90 (ConveyLinx-Ai2 Modbus TCP) and mock backends.

Diverter Types

Usage

This is an async library: every 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, nothing connected. mode="mock" swaps in an in-memory diverter that completes the full sequence successfully.

import asyncio

from popup_diverter.models import DiverterConfig, DiverterFleetConfig
from popup_diverter.service import DiverterService

config = DiverterFleetConfig(
    diverters=[
        DiverterConfig(id="d1", host="192.168.1.100", location_id="zone-a"),
    ],
)
service = DiverterService.from_config(config, mode="mock")


async def main():
    await service.connect_all()
    result = await service.divert("zone-a")  # full raise→belt→lower sequence
    await service.disconnect_all()

    print(result.status)  # DivertStatus.OK | TIMEOUT | ERROR | BUSY
    print(result.lift_reached_top)  # True / False
    print(result.error)  # None or error string


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

divert() is the one-call easy path — it raises the lift, runs the belt for the configured time, lowers, and stops. Use the individual motor controls below only to drive the steps yourself.

Connecting real hardware

Same code — swap mode="mock" for mode="real". host is each diverter's IP; location_id is the name you address it by (in divert() / get_diverter()).

service = DiverterService.from_config(config, mode="real")

Divert Sequence

A single divert() call executes the full mechanical sequence:

1. Raise lift — left motor runs forward (PGD drives eccentric shafts up)
2. Wait for top sensor — polls inductive sensor until lift is fully raised (or timeout)
3. Start transport belt — right motor runs in configured direction
4. Run belt — tote transfers sideways for configured duration
5. Stop belt + lower lift — left motor reverses to lower the assembly
6. Wait for lower — timeout-based (no bottom sensor)
7. Stop all

On failure at any step, the diverter safely stops and lowers before returning an error.

Individual Motor Control

For testing, commissioning, or manual operation:

diverter = service.get_diverter("zone-a")
await diverter.raise_lift()
await diverter.start_belt(BeltDirection.FORWARD)
await diverter.stop_belt()
await diverter.lower_lift()
await diverter.stop()  # emergency stop all

Diagnostics

state = await service.read_state("zone-a")
state.voltage_mv  # 24136 (24.1V)
state.top_sensor  # True/False
state.lift_current_ma  # 450
state.belt_running  # True/False
state.lift_error  # None or "stalled, overloaded"

Metrics

metrics = service.get_metrics()["d1"]
metrics.total_diverts  # 1234
metrics.success_rate  # 0.95
metrics.avg_divert_ms  # 8200.0

Structured Logging

from popup_diverter.logger import set_log_callback


def on_diverter_log(code, source, level, message):
    mqtt_publish("diverter/logs", {"code": code, "source": source, "message": message})


set_log_callback(on_diverter_log)

Hardware: Flowsort X-Flow90

The X-Flow90 is a 90° pop-up roller transfer. Each unit has a ConveyLinx-Ai2 controller (by Pulseroller) that drives two motors and reads one sensor, all controlled via Modbus TCP.

Components

Modbus TCP Registers

The ConveyLinx-Ai2 must be in PLC I/O mode (configured via EasyRoll+). All communication uses assembled register block reads/writes.

Key output registers (offsets from base):

Key input registers (offsets from base):

Daisy Chaining

ConveyLinx-Ai2 modules can be daisy-chained via RJ-45 (Link Left / Link Right ports). One cable from the network switch to the first module, then chain the rest. Each module needs a unique static IP.

EasyRoll+ Setup (one-time per module)

1. Download EasyRoll+ (Windows only)
2. Connect via RJ-45, discover the module
3. Set static IP, subnet mask, gateway, disable DHCP
4. Set PLC I/O mode (Configuration tab → "Current Mode PLC")
5. Set "Outputs/Motors On PLC Disconnected" to "Stop All"

API

DiverterService

class DiverterService:
    @classmethod
    def from_config(cls, config, mode="real") -> DiverterService

    async def divert(self, location_id, direction=BeltDirection.FORWARD) -> DivertResult
    async def stop(self, location_id) -> None
    async def stop_all(self) -> None
    async def connect_all(self) -> dict[str, bool]
    async def disconnect_all(self) -> None
    async def read_state(self, location_id) -> DiverterState | None
    async def read_all_states(self) -> dict[str, DiverterState]
    def get_metrics(self) -> dict[str, DivertMetrics]
    def get_metrics_summary(self) -> dict

DivertResult

class DivertResult:
    status: DivertStatus  # OK | TIMEOUT | ERROR | BUSY
    diverter_id: str
    lift_reached_top: bool
    error: str | None

PopupDiverter ABC

class PopupDiverter(ABC):
    async def connect(self) -> bool
    async def disconnect(self) -> None
    async def divert(self, direction=BeltDirection.FORWARD) -> DivertResult
    async def stop(self) -> None
    async def raise_lift(self) -> None
    async def lower_lift(self) -> None
    async def start_belt(self, direction=BeltDirection.FORWARD) -> None
    async def stop_belt(self) -> None
    async def stop_lift(self) -> None
    async def read_state(self) -> DiverterState
    connected: bool  # property

Configuration

YAML config with optional config.local.yaml overlay and ${ENV_VAR:-default} substitution.

default_timeout: 5.0
reconnect_delay: 5.0
max_reconnect_attempts: 0   # 0 = unlimited
poll_interval: 0.05          # sensor polling during divert (seconds)

fleet:
  - id: diverter-1
    host: "192.168.1.100"
    location_id: zone-a
    belt_direction: forward  # or reverse
    lift_speed: 850          # RPM × 10
    belt_speed: 400          # mm/s
    lift_timeout: 5.0        # max seconds to raise lift
    belt_run_time: 3.0       # seconds belt runs after lift reaches top
    lower_timeout: 5.0       # max seconds to lower lift

Development

cd popup-diverter
uv sync
make test
make lint