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.
“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())
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())