(interface: str, bitrate: int, channel: Optional[str] = None)
Entry point for script.
Entry point for script.
Entry point for script .
async def run(interface: str, bitrate: int, channel: Optional[str] = None) -> None:
    """Entry point for script."""
    log.info(f"Connecting to {interface} {bitrate} {channel}")
    driver = await CanDriver.build(
        bitrate=bitrate, interface=interface, channel=channel
    )
    messenger = CanMessenger(driver=driver)
    messenger.start()

    await messenger.send(
        node_id=NodeId.broadcast, message=SetupRequest(payload=EmptyPayload())
    )
    await messenger.send(
        node_id=NodeId.broadcast, message=EnableMotorRequest(payload=EmptyPayload())
    )

    # TODO (al, 2021-11-11): Allow creating groups from command line or config file.
    move_groups = [
        # Group 0
        [
            {
                NodeId.gantry_x: MoveGroupSingleAxisStep(
                    distance_mm=0, velocity_mm_sec=5000.5, duration_sec=3
                ),
            },
            {
                NodeId.gantry_y: MoveGroupSingleAxisStep(
                    distance_mm=0, velocity_mm_sec=5000.5, duration_sec=3
                ),
            },
        ],
        # Group 1
        [
            {
                NodeId.gantry_x: MoveGroupSingleAxisStep(
                    distance_mm=0, velocity_mm_sec=2000.25, duration_sec=3
                ),
            },
            {
                NodeId.gantry_y: MoveGroupSingleAxisStep(
                    distance_mm=0, velocity_mm_sec=1000.5, duration_sec=3
                ),
            },
        ],
    ]

    runner = MoveGroupRunner(move_groups=move_groups)

    try:
        await runner.run(can_messenger=messenger)
    except asyncio.CancelledError:
        pass
    finally:
        await messenger.stop()
        driver.shutdown()
async def run ( interface : str , bitrate : int , channel : Optional [ str ] = None ) -> None : log . info ( f"Connecting to {interface} {bitrate} {channel}" ) driver = await CanDriver . build ( bitrate = bitrate , interface = interface , channel = channel ) messenger = CanMessenger ( driver = driver ) messenger . start ( ) await messenger . send ( node_id = NodeId . broadcast , message = SetupRequest ( payload = EmptyPayload ( ) ) ) await messenger . send ( node_id = NodeId . broadcast , message = EnableMotorRequest ( payload = EmptyPayload ( ) ) ) # TODO (al, 2021-11-11): Allow creating groups from command line or config file. move_groups = [ # Group 0 [ { NodeId . gantry_x : MoveGroupSingleAxisStep ( distance_mm = 0 , velocity_mm_sec = 5000.5 , duration_sec = 3 ) , } , { NodeId . gantry_y : MoveGroupSingleAxisStep ( distance_mm = 0 , velocity_mm_sec = 5000.5 , duration_sec = 3 ) , } , ] , # Group 1 [ { NodeId . gantry_x : MoveGroupSingleAxisStep ( distance_mm = 0 , velocity_mm_sec = 2000.25 , duration_sec = 3 ) , } , { NodeId . gantry_y : MoveGroupSingleAxisStep ( distance_mm = 0 , velocity_mm_sec = 1000.5 , duration_sec = 3 ) , } , ] , ] runner = MoveGroupRunner ( move_groups = move_groups ) try : await runner . run ( can_messenger = messenger ) except asyncio . CancelledError : pass finally : await messenger . stop ( ) driver . shutdown ( )