Ohara Lab

robo

There are primarily two ways to use it: synchronous and asynchronous programs.

A synchronous program can be written as a straightforward sequence of operations. Alternatively, synchronous code can be run asynchronously using threads or processes. The other approach is an asynchronous program using async.

Synchronous version

“A script that moves to the electronic balance (position P62) and measures the weight”

#!/usr/bin/env python
import logging
from pathlib import Path

# Set httpx logging to WARNING or above (suppress INFO).
logging.getLogger("httpx").setLevel(logging.WARNING)

def main():
    from cobotta2 import Config, MotionMode
    from cobotta2.server import AsyncCobottaClient
    from aandd_reader.server import AsyncAanddReaderClient
    from x_logger import XLogger

    HERE = Path(__file__).parent
    config_cobotta = Config(HERE / "config_cobotta1_server1.yaml")
    config_aandd_reader = Config(HERE / "config_cobotta1_server1.yaml")

    logger = XLogger(log_level="info", logger_name='test')
    cobotta = SyncCobottaClient(config=config_cobotta, logger=logger)
    aandd_reader = SyncAanddReaderClient(config=config_aandd_reader, logger=logger)
    cobotta.reset_error()
    cobotta.take_arm()
    cobotta.turn_on_motor()

    ### Robot and balance control
    ret = cobotta.move("P62")
    if not ret:
        logger.error(f"{cobotta.error_description()}': move failed")
        return False
    cobotta.wait_for_complete()
    mass = aandd_reader.meas()

    current_pos = cobotta.get_current_position()
    logger.info(f"mass = {mass}, current_position: {current_pos}")
    return True

if __name__ == "__main__":
    asyncio.run(main())

Asynchronous version

Error handling and similar details are omitted to keep this example as simple as possible. The basic procedure is the same as in the synchronous version. For modern programs, especially those with substantial I/O, it is worth becoming comfortable writing asynchronous code.

“A script that moves to the electronic balance (position P62) and measures the weight”

#!/usr/bin/env python
import asyncio
import logging
from pathlib import Path

# Set httpx logging to WARNING or above (suppress INFO).
logging.getLogger("httpx").setLevel(logging.WARNING)

async def main():
    from cobotta2 import Config, MotionMode
    from cobotta2.server import AsyncCobottaClient
    from aandd_reader.server import AsyncAanddReaderClient
    from x_logger import XLogger

    HERE = Path(__file__).parent
    config_cobotta = Config(HERE / "config_cobotta1_server1.yaml")
    config_aandd_reader = Config(HERE / "config_cobotta1_server1.yaml")

    logger = XLogger(log_level="info", logger_name='test')
    cobotta = AsyncCobottaClient(config=config_cobotta, logger=logger)
    aandd_reader = AsyncAanddReaderClient(config=config_aandd_reader, logger=logger)
    await cobotta.reset_error()
    await cobotta.take_arm()
    await cobotta.turn_on_motor()

    ### Robot and balance control
    ret = await cobotta.move("P62")
    if not ret:
        logger.error(f"{await cobotta.error_description()}': move failed")
        return False
    await cobotta.wait_for_complete()
    mass = await aandd_reader.meas()

    current_pos = await cobotta.get_current_position()
    logger.info(f"mass = {mass}, current_position: {current_pos}")
    return True

if __name__ == "__main__":
    asyncio.run(main())

Author