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_NAME and SERVER_LOGGER_NAME may be set to any preferred names.

  • JOYPAD_PORT must be configured when using a joypad.

  • CAMERA_IP must 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")


Author

  • Kengo NAKADA