Alternative examples
Basic COBOTTA usage examples and explanations of the functions used.
Writing config.yaml
Some naming conventions became inconsistent as functionality was extended.
When gRPC support was added, the port names were separated to distinguish FastAPI (ese774) from gRPC. When gRPC is enabled, the FastAPI port must be configured as FASTAPI_PORT rather than the former SERVER_PORT, because the server uses both GRPC_PORT and FASTAPI_PORT. Otherwise, an error occurs at server startup. Some older server configuration examples may still use the previous naming.
Users who only access an existing server do not need to worry about this. Administrators starting a device server should check the port names.
# Server
SERVER_IP: "10.170.10.33"
# SERVER_IP: "10.170.10.38"
FASTAPI_PORT: 8004
GRPC_PORT: 50051
SERVER_LOGGER_NAME: "AanddReaderServer"
# Device
DEVICE_PORT: "COM4"
CLIENT_LOGGER_NAMEandSERVER_LOGGER_NAMEmay be set to any preferred names.JOYPAD_PORTmust be configured when using a joypad.CAMERA_IPmust be configured when using a camera.
The COBOTTA server, the robot, and other servers (such as the joypad server) all read their settings from YAML. Be careful not to confuse their settings.
# --- cobotta-server
SERVER_IP: "10.170.10.41"
FASTAPI_PORT: 8000
CLIENT_LOGGER_NAME: "cobotta4_client"
SERVER_LOGGER_NAME: "cobotta4_server"
TIMEOUT: 5
LOGGER_LEVEL: "debug"
# --- joypad-server
JOYPAD_PORT: 50051
JOYPAD_LOGGER_NAME: "cobotta4_joypad"
# --- cobotta
COBOTTA_IP: "10.170.10.43"
COBOTTA_CAMERA_IP: None
STOP
Stop the operation.
Checking current errors
Almost all functions return a Boolean (True or False) indicating whether the requested operation completed successfully. Checking the return value is optional, but a False result means that the operation did not complete successfully. It is good practice to handle such failures and display the cause.
import asyncio
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
# Using an explicit configuration path avoids problems.
HERE = Path(__file__).parent
Config.load_yaml(HERE / "config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
try:
# Current error code
code = await client.error_code()
# Current error description
description = await client.error_description()
logger.info(f"cdoe={code}, description={description}")
except Exception as e:
print(e)
assert "result" == "result"
if __name__ == "__main__":
asyncio.run(main())
Basic command: take_arm
Acquire control of the arm. The robot cannot be moved without arm-control authority. This authority is acquired on the server side.
Take particular care when multiple users share a server.
Multi-user operation has not been fully implemented. Exclusive control when several terminals access the server concurrently has known problems; more precisely, the required behavior has not been fully specified, so this is a design limitation.
import asyncio
async def main():
# ...
await client.take_arm()
Basic command: give_arm (not implemented)
Release previously acquired arm-control authority. Since the server holds this authority, a client may continue operating after acquiring it without explicitly releasing it each time. In multi-user environments, however, take/release operations require care.
import asyncio
async def main():
# ...
await client.given_arm()
There is also a release_arm() method. It has the same purpose; only the name differs.
Basic command: turn_on_motor
Turn the motors on.
import asyncio
async def main():
# ...
await client.turn_on_motor()
Basic command: turn_off_motor
Turn the motors off. This also disables various functions, so it is an important operation.
import asyncio
async def main():
# ...
await client.turn_off_motor()
Basic command: set_coord_type
This is a global setting, not a per-method setting.
import asyncio
async def main():
# ...
await client.set_coord_type("J")
await client.set_coord_type("P")
Basic command: set_motion_mode
This is a global setting, not a per-method setting.
import asyncio
async def main():
# ...
await client.set_motion_mode(MotionMode.PTP)
await client.set_coord_type(MotionMode.LINE)
await client.set_coord_type(MotionMode.CIRCLE)
await client.set_coord_type(MotionMode.SPLINE)
Basic command: set_speed
This is a global setting, not a per-method setting.
import asyncio
async def main():
# ...
await client.set_speed(100)
Basic command: set_accel
This is a global setting, not a per-method setting.
import asyncio
async def main():
# ...
await client.set_accel(100)
Basic command: set_decel
This is a global setting, not a per-method setting.
import asyncio
async def main():
# ...
await client.set_decel(100)
Current arm position
import asyncio
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
try:
# Current position (XYZ)
positon = await client.get_current_position()
# Current position (joint angles)
angle = await client.get_current_angle()
logger.info(f"{positon}")
logger.info(f"{angle}")
except Exception as e:
print(e)
assert "result" == "result"
# send_command("unknown") # This produces an error
Checking whether an operation succeeded
Most robot-control functions return a value indicating whether the operation succeeded or ended with an error.
import asyncio
async def main():
# ...
ret = await client.take_arm()
if not ret:
# If arm control cannot be acquired, report the error and stop
error = await client.error_description()
logger.error(f"take_arm error: {error}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
The result of nearly every command, including movement commands, can be checked.
import asyncio
async def main():
# ...
# Movement command
ret = await client.move('P10')
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
if __name__ == "__main__":
asyncio.run(main())
Arm movement 0 (DENSO syntax)
The move() method also accepts a string resembling DENSO script syntax.
This form is not recommended.
import sys
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
# If an error exists, reset it before moving
await client.reset_error()
# Call take_arm before moving the arm
ret = await client.take_arm()
if not ret:
# If arm control cannot be acquired, report the error and stop
error = await client.error_description()
logger.error(f"take_arm error: {error}")
sys.exit(1)
logger.info("turn_on_motor")
await client.turn_on_motor()
client.motion_mode = MotionMode.LINE
# Move to P62
ret = await client.move("P62+(0,0,10)")
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
# Wait for the motion to finish
await client.wait_for_complete()
Arm movement 1: standard usage
Use coordinates already defined in the robot controller.
import sys
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
# If an error exists, reset it before moving
await client.reset_error()
# Call take_arm before moving the arm
ret = await client.take_arm()
if not ret:
# If arm control cannot be acquired, report the error and stop
error = await client.error_description()
logger.error(f"take_arm error: {error}")
sys.exit(1)
logger.info("turn_on_motor")
await client.turn_on_motor()
client.motion_mode = MotionMode.LINE
# Move to P62
ret = await client.move("P62")
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
# Wait for the motion to finish
await client.wait_for_complete()
Arm movement 2 (position)
XYZ coordinates (P)
Joint-angle coordinates (J), J1–J6
Move by specifying either coordinate representation.
import logging
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
Config.load_yaml(HERE / "../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Global setting: motion_mode PTP Specify
client.motion_mode = MotionMode.PTP
# client.motion_mode = MotionMode.LINE
# client.motion_mode = MotionMode.CIRCLE
# client.motion_mode = MotionMode.SPLINE
# Global setting: Move using XYZ coordinates (P)
await client.set_coord_type("P")
# Create a position (tuple or list)
pos = (
-29.0453511306835,
343.1440808456958,
115.31284305934872,
-178.28955954148657,
-16.619847808793967,
141.1028536689876,
)
# Move to the specified position
ret = await client.move(pos)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
assert ret is not None, "Connection failed"
await client.wait_for_complete()
Arm movement 3 (j2p, p2j)
Convert joint angles to a position before moving.
import logging
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
Config.load_yaml(HERE / "../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Global setting: PTP Specify
client.motion_mode = MotionMode.PTP
# client.motion_mode = MotionMode.LINE
# client.motion_mode = MotionMode.CIRCLE
# client.motion_mode = MotionMode.SPLINE
# Global setting: Move using XYZ coordinates (P)
await client.set_coord_type("P")
# Create a position (tuple or list)
new_pos = (
-29.0453511306835,
343.1440808456958,
115.31284305934872,
-178.28955954148657,
-16.619847808793967,
141.1028536689876,
)
# Convert P to J
next_angle = await client.p2j(new_pos)
# Change the angle of only the specified axis
next_angle[4] = next_angle[4] - 15
# Convert J to P
next_pos = await client.j2p(next2_angle)
# Move to the specified position
ret = await client.move(next_pos)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
assert ret is not None, "Connection failed"
await client.wait_for_complete()
Arm movement 4 (angle)
Move using specified joint angles. This is also useful for constructing coordinates through fine adjustments of position or angle.
import logging
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
Config.load_yaml(HERE / "../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Global setting: PTP Specify
client.motion_mode = MotionMode.PTP
# client.motion_mode = MotionMode.LINE
# client.motion_mode = MotionMode.CIRCLE
# client.motion_mode = MotionMode.SPLINE
# Create joint angles (tuple or list)
target_angle = [
56.04999367408907,
35.12319454476854,
114.33533666988626,
-4.0077312418009745,
0,
-25.962042331415812,
0.0,
0.0,
]
# Move to target_angle
# For angles, specify coord_type J
ret = await client.move(
pos,
coord_type="J",
)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
assert ret is not None, "Connection failed"
await client.wait_for_complete()
Arm movement 5 (coord_type)
Movement with an explicit coord_type
import logging
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
Config.load_yaml(HERE / "../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Create joint angles (tuple or list)
target_angle = [
56.04999367408907,
35.12319454476854,
114.33533666988626,
-4.0077312418009745,
0,
-25.962042331415812,
0.0,
0.0,
]
# Move to target_angle
# For angles, specify coord_type J
ret = await client.move(
pos,
coord_type="J",
)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
assert ret is not None, "Connection failed"
await client.wait_for_complete()
Arm movement 6 (path_blend)
A -> B -> C
When moving along this route, should accurate passage through B be prioritized, or should the speed at B be maintained? The path may instead be blended around B, treating it as an intermediate waypoint. Changes how path operations are combined when multiple paths are present. This parameter affects the path itself, not merely speed, acceleration, or deceleration. Terminology differs among manufacturers, but the parameter generally has a similar meaning.
import logging
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
Config.load_yaml(HERE / "../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Global setting: PTP Specify
client.motion_mode = MotionMode.PTP
# client.motion_mode = MotionMode.LINE
# client.motion_mode = MotionMode.CIRCLE
# client.motion_mode = MotionMode.SPLINE
# Move to target_angle
# For angles, specify coord_type J
ret = await client.move(
pos,
coord_type="P",
path_blend="@P",
)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
assert ret is not None, "Connection failed"
await client.wait_for_complete()
Arm movement 7 (speed)
Speed
import sys
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
# If an error exists, reset it before moving
await client.reset_error()
# Call take_arm before moving the arm
ret = await client.take_arm()
if not ret:
# If arm control cannot be acquired, report the error and stop
error = await client.error_description()
logger.error(f"take_arm error: {error}")
sys.exit(1)
logger.info("turn_on_motor")
await client.turn_on_motor()
client.motion_mode = MotionMode.LINE
# Move to P62
ret = await client.move(
"P62",
coord_type="P",
path_blend="@P",
speed=100,
)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
# Wait for the motion to finish
await client.wait_for_complete()
Arm movement 8 (accel/decel)
Acceleration and deceleration
import sys
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
# If an error exists, reset it before moving
await client.reset_error()
# Call take_arm before moving the arm
ret = await client.take_arm()
if not ret:
# If arm control cannot be acquired, report the error and stop
error = await client.error_description()
logger.error(f"take_arm error: {error}")
sys.exit(1)
logger.info("turn_on_motor")
await client.turn_on_motor()
client.motion_mode = MotionMode.LINE
# Move to P62
ret = await client.move(
"P62",
path_blend="@P",
coord_type="P",
speed=100,
accel=100,
decel=100,
)
if not ret:
error = await client.error_description()
logger.error(f"move error: {error}")
sys.exit(1)
# Wait for the motion to finish
await client.wait_for_complete()
Arm movement 9 (motion_mode)
Motion mode
import asyncio
# from http import client
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
angle = await client.get_current_angle()
angle[5] = angle[5] + -40
new_pos = await client.j2p(angle)
await client.move(
new_pos,
motion_mode=MotionMode.PTP,
# motion_mode = MotionMode.PTP,
# motion_mode = MotionMode.LINE,
# motion_mode = MotionMode.CIRCLE,
# motion_mode = MotionMode.SPLINE,
path_blend="@P",
coord_type="P",
speed=100,
accel=100,
decel=100,
)
if __name__ == "__main__":
asyncio.run(main())
Arm movement 10 (drive command)
The drive command can rotate only the specified joint directly.
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta3_server2.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Relative movement: rotate J6 by -90 degrees
await client.drive([6, -90], relative=True)
if __name__ == "__main__":
asyncio.run(main())
Arm movement 11 (offset command)
Offset movement
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta3_server2.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Relative movement with offset x=0, y=0, z=-10, rx=0, ry=0, rz=0
await client.move(
'P62',
offset=(0, 0, -10, 0, 0, 0)
)
if __name__ == "__main__":
asyncio.run(main())
Arm movement 12 (rotate)
Rotate around a specified coordinate axis. This function may ultimately be specific to COBOTTA.
The command implementation (including option syntax) is still under consideration and may change.
# import pytest
import asyncio
import sys
import json
from pathlib import Path
import logging
# Set httpx logging to WARNING or above (suppress INFO).
logging.getLogger("httpx").setLevel(logging.WARNING)
# import time
from cobotta2 import Config, MotionMode, RobotVarManager
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
HERE = Path(__file__).parent
config_cobotta2 = Config(HERE / "../config_cobotta2.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=config_cobotta2, logger=logger)
ret = await client.reset_error()
if ret is None or ret is False:
assert False, "Cobotta2 Connect Error"
await client.release_arm()
await client.turn_on_motor()
await client.take_arm()
await client.move(
"P123",
path_blend="@E",
motion_mode=MotionMode.PTP,
speed=100,
)
await client.wait_for_complete()
# Rotate 45 degrees about the origin (200, -50, 197).
await client.rotate("XY", 45, [200, -50, 197])
ret = await client.get_current_position()
print(json.dumps(ret, indent=2))
if __name__ == "__main__":
# asyncio.run(worker_cobotta2())
asyncio.run(main())
Gripper 1 (distance-controlled movement)
import asyncio
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Move to the 10 mm position.
await client.hand(pos=10, speed=100)
sleep(2)
# At present
# client.wait_for_complete() cannot be used for the gripper.
if __name__ == "__main__":
asyncio.run(main())
Gripper 2 (force-controlled closing)
import asyncio
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.reset_error()
await client.take_arm()
await client.turn_on_motor()
# Grip the target with 6 N of force.
await client.hand(force=6)
sleep(2)
# At present
# client.wait_for_complete() cannot be used for the gripper.
if __name__ == "__main__":
asyncio.run(main())
Changing the coordinate system (defining work coordinates)
Change the work coordinate system. By default, the robot coordinate system is also the work coordinate system, but for example, replacing the XYZ axes with tilted axes can make XYZ-based positioning easier. This also simplifies defining offset coordinates for relative movements.
import asyncio
from cobotta2 import Config, CobottaCtrl, MotionMode, RobotVarManager
from x_logger import XLogger
async def main():
Config.load_yaml("../config_cobotta1.yaml")
logger = XLogger(log_level="debug", logger_name=Config.CLIENT_LOGGER_NAME)
cobotta = AsysncCobottaClient(cobotta_ip=Config.COBOTTA_IP, arm_name="Arm", logger=logger)
await cobotta.reset_error()
try:
await cobotta.take_arm()
await cobotta.turn_on_motor()
current_pos = await cobotta.get_current_position()
logger.info(f"current_pos = {current_pos}")
pos = [
# 20.259341327288382,
0.0,
# 276.59469818653,
280,
# 38.52868750215305,
50.0,
# 178.72745234477128,
180.0,
# 4.521890180915519,
0.0,
# -89.24661909122011,
-90.0,
261.0,
]
data = await cobotta.get_work(0)
logger.info(f"get_work {data}")
logger.info(f"change_work(0)")
await cobotta.change_work(0)
logger.info("******")
logger.info("")
logger.info(f"set_work {pos}")
await cobotta.set_work(1, pos)
logger.info(f"change_work(1)")
await cobotta.change_work(1)
data = await cobotta.get_current_position()
logger.info(f"current_pos = {data}")
logger.info("")
logger.info("******")
logger.info("")
logger.info(f"change_work(0)")
await cobotta.change_work(0)
current_pos = await cobotta.get_current_position()
logger.info(f"current_pos = {current_pos}")
if __name__ == "__main__":
asyncio.run(main())
Retrieving COBOTTA hardware logs
import asyncio
async def main():
# ..
await client.start_log()
await client.start_servo_log()
Retrieve logs from the COBOTTA hardware (teaching pendant).
Reading and writing internal variables 1 (get_P)
Use this method to retrieve a specified internal P position from COBOTTA.
import asyncio
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.take_arm()
logger.info("== turn on motor()")
ret = await client.turn_on_motor()
assert ret is not None, "Connection failed"
logger.info("== run test()")
ret = await client.get_P(61)
assert "result" == "result"
Reading and writing internal variables 2 (property)
import asyncio
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger
async def main():
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.take_arm()
logger.info("== turn on motor()")
ret = await client.turn_on_motor()
logger.info("== run test()")
p61 = await client.P61
logger.info(f"ret = {p61}")
Writing internal variable values from a file to COBOTTA
import asyncio
import re
import json
from samples.cobotta2 import Config
from samples.cobotta2 import AsyncCobottaClient
from x_logger.x_logger import XLogger
async def main():
Config.load_yaml("config_server1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.take_arm()
await client.turn_on_motor()
start = "P60"
end = "P105"
filepath = f"COBOTTA1_{start}_{end}_pos.json"
# Extract the number after P and convert it to an integer.
start_num = int(re.findall(r"\d+", start)[0])
end_num = int(re.findall(r"\d+", end)[0])
# Read from a file.
with open(filepath, "r", encoding="utf-8") as f:
memory_loaded = json.load(f)
for i in range(start_num, end_num + 1):
key = f"P{i}"
value = memory_loaded.get(key)
if value is not None:
await client.set_P(i, value)
logger.info(f"set_P({i}, {value})")
else:
print(f"{key} does not exist")
if __name__ == "__main__":
asyncio.run(main())
Saving internal variables to a file
import asyncio
import time
import re
import json
from datetime import datetime
from cobotta2 import Config
from cobotta2 import AsyncCobottaClient
from x_logger.x_logger import XLogger
async def main():
Config.load_yaml("config_server1.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
client = AsyncCobottaClient(config=Config, logger=logger)
await client.take_arm()
await client.turn_on_motor()
start = "P120"
end = "P140"
# Extract the number after P and convert it to an integer.
start_num = int(re.findall(r"\d+", start)[0])
end_num = int(re.findall(r"\d+", end)[0])
memory = {}
for i in range(start_num, end_num + 1):
name_pos = await client.get_P(i)
logger.info(f"P{i}: {name_pos}")
target = list(name_pos)
memory[f"P{i}"] = target
now = datetime.now()
dt_str = now.strftime("%Y-%m-%d_%H%M") # Example: '2025-07-17_1632'
# Write to a file.
filepath = f"COBOTTA1_{start}_{end}_pos_{dt_str}.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(memory, f, ensure_ascii=True, indent=4)
filepath = f"COBOTTA1_{start}_{end}_pos.json"
with open(filepath, "w", encoding="utf-8") as f:
json.dump(memory, f, ensure_ascii=True, indent=4)
time.sleep(0.1)
if __name__ == "__main__":
asyncio.run(main())
Listing internal PacScript programs
Direct use of DENSO scripts is possible, although it is generally not recommended except in special cases. List COBOTTA-format PacScript files from DENSO WAVE. Display the scripts stored in COBOTTA.
from cobotta2 import Config, CobottaCtrl, MotionMode
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="debug", logger_name=Config.CLIENT_LOGGER_NAME)
cobotta = CobottaCtrl(cobotta_ip=Config.COBOTTA_IP, arm_name="Arm", logger=logger)
print(cobotta.get_cobotta_file_names())
Running an internal PacScript program
Direct use of DENSO scripts is possible, although it is generally not recommended except in special cases. Handle DENSO WAVE COBOTTA-format PacScript files. This can be used to test the control class.
*) When using CobottaCtrl directly in local mode, always disconnect on exit. Also disconnect on abnormal termination, including KeyboardInterrupt.
from cobotta2 import Config, CobottaCtrl, MotionMode
Config.load_yaml("config_cobotta2.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
cobotta = CobottaCtrl(cobotta_ip=Config.COBOTTA_IP, arm_name="Arm", logger=logger)
try:
cobotta.reset_error()
cobotta.turn_on_motor()
cobotta.release_arm()
name = "Pro1.pcs" # The file extension is optional.
# Note that this is mutually exclusive with take_script().
cobotta.take_script(name)
cobotta.run_script()
cobotta.wait_for_complete()
except KeyboardInterrupt as e:
logger.error(e)
except Exception as e:
loggger.error(e)
finally:
cobotta.disconnect()
Loading an internal PacScript program (upload)
Direct use of DENSO scripts is possible, although it is generally not recommended except in special cases. Handle DENSO WAVE COBOTTA-format PacScript files. This can be used to test the control class.
from cobotta2 import Config, CobottaCtrl, MotionMode
Config.load_yaml("config_cobotta2.yaml")
logger = XLogger(log_level="info", logger_name=Config.CLIENT_LOGGER_NAME)
cobotta = CobottaCtrl(cobotta_ip=Config.COBOTTA_IP, arm_name="Arm", logger=logger)
file_content = """
Sub Main
TakeArm Keep = 0
' Descending motion: move down to P[61]
Move L, @0 P[61]
End Sub
"""
# Upload file_content to Pro2.pcs.
cobotta.upload_content_to_cobotta(file_content, "Pro2.pcs")
print(cobotta.get_cobotta_file_names())
Saving an internal PacScript program (download)
Direct use of DENSO scripts is possible, although it is generally not recommended except in special cases. Handle DENSO WAVE COBOTTA-format PacScript files. This can be used to test the control class.
*) When using CobottaCtrl directly in local mode, always disconnect on exit. Also disconnect on abnormal termination, including KeyboardInterrupt.
from cobotta2 import Config, CobottaCtrl
from x_logger import XLogger
Config.load_yaml("config_cobotta1.yaml")
logger = XLogger(log_level="debug", logger_name=Config.CLIENT_LOGGER_NAME)
cobotta = CobottaCtrl(cobotta_ip=Config.COBOTTA_IP, arm_name="Arm", logger=logger)
print(cobotta.get_cobotta_file_names())
cobotta.download_file_from_cobotta("P62.pcs")