代用的なexample

コボッタの基本的使用例とそこで使っている関数の説明

config.yaml の書き方

機能拡張によって名前の方針が少しぶれているところがある。

途中で gRPC 対応にした時に FastAPI(ese774) 利用と gRPC 利用で PORTを分けるために用語を切り分けた。gRPC を読み込んでいる場合は、SERVER_PORT が FASTAPI_PORT に してないとだ正しく機能しない(サーバーが GRPC_PORT と FASTAPI_PORT の二つになる) サーバー起動時に文句を言われる。 これを区別してないサーバー設定サンプルがどこかに残っていることがある。

サーバーを使うだけのユーザーは気にしなくて良い。 デバイスサーバーを立ち上げる管理人は注意する必要がある。

# 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 と SERVER_LOGGER_NAME は好きな名前でいい。

• JOYPAD を使う時は JOYPAD_PORT の設定が必須

• CAMERA を使う時は CAMERA_IP の設定が必須

コボッタサーバーとコボッタと他のサーバ(joypad)など、 yaml から設定を読むので混乱しないように注意する

# --- 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

停止する

現在発生しているエラーの把握

ほぼ全ての関数は、関数で指定された動作が正しく終了したかどうか？ の判定を bool (True/False) で返す。その結果を参照しても しなくても問題ないが、False のときはその動作が正しく終了してない ので、ハンドリングした方が健全であり、エラーの原因を表示した方がいい。

import asyncio
from pathlib import Path
from cobotta2 import Config, MotionMode
from cobotta2.server import AsyncCobottaClient
from x_logger import XLogger

async def main():
    # config はパス情報も含めたほうがトラブルは少ない
    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:
        # 現在のエラーコード
        code = await client.error_code()
        # 現在のエラー
        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())

基本命令 (take_arm)

アーム権限の取得 アーム権限を取得してないと動かすことができない。 これは、本質的にはサーバー側で取得される。

つまり、サーバを複数人で使っているときは注意。

マルチユーザー環境でまだ作り組んでいないので、同時に別端末から 使うときの排他的扱いはバグがある（むしろ「仕様上」つめてないので、 設計上の不備がある）

import asyncio

async def main():
    # ...
    await client.take_arm()

基本命令 (give_arm) 未実装

取得したアーム権限を解放する。 サーバー側でアームを取得しているので、クライアントからは 一度アームを取得してしまえば、熱心にアーム解放をしなくても動くが、 マルチユーザー環境だと take/given は注意する。

import asyncio

async def main():
    # ...
    await client.given_arm()

実は release_arm() というメソッドもある。ないような同じで愛絵違いなだけ。

基本命令 (turn_on_motor)

モーターのon

import asyncio

async def main():
    # ...
    await client.turn_on_motor()

基本命令 (turn_off_motor)

モーターの off。 モーターを off したら色々切れるので割と重要

import asyncio

async def main():
    # ...
    await client.turn_off_motor()

基本命令 (set_coord_type)

メソッドごとではなくて、全体に設定。

import asyncio

async def main():
    # ...
    await client.set_coord_type("J")
    await client.set_coord_type("P")

基本命令 (set_motion_mode)

メソッドごとではなくて、全体に設定。

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)

基本命令 (set_speed)

メソッドごとではなくて、全体に設定。

import asyncio

async def main():
    # ...
    await client.set_speed(100)

基本命令 (set_accel)

メソッドごとではなくて、全体に設定。

import asyncio

async def main():
    # ...
    await client.set_accel(100)

基本命令 (set_decel)

メソッドごとではなくて、全体に設定。

import asyncio

async def main():
    # ...
    await client.set_decel(100)

現在のアームの位置

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:
        # 現在の位置(xyz)情報
        positon = await client.get_current_position()
        # 現在の位置(angle)情報
        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")  # これはエラーになる

動作に対してエラー判定する

大抵のロボット系関数は戻り値として動作したか？異常に終わったか？の判定ができる

import asyncio

async def main():
    # ...
    ret = await client.take_arm()
    if not ret:
        # アームが取得できないときにエラーの種類を用事して止める
        error = await client.error_description()
        logger.error(f"take_arm error: {error}")
        sys.exit(1)
        
if __name__ == "__main__":
    asyncio.run(main())

移動コマンド始め、ほぼあらゆるコマンドで成功の可否を取れる

import asyncio

async def main():
    # ...
    # 移動命令
    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())

アームの移動0 (デンソー形式)

move() では string で デンソーのスクリプト契機っぽい記述もできる。 使用は推奨しない。

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)
    
    # エラーがある時は rest しないと動かすことができない
    await client.reset_error()

    # アームを動かす前には take_arm をする必要がある
    ret = await client.take_arm()
    if not ret:
        # アームが取得できないときにエラーの種類を用事して止める
        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
   
    # 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)

    # モーションが終わるまで待つ
    await client.wait_for_complete()

アームの移動1 標準的な使い方

ロボット自体が内部で標準で持っている座標を使う

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)
    
    # エラーがある時は rest しないと動かすことができない
    await client.reset_error()

    # アームを動かす前には take_arm をする必要がある
    ret = await client.take_arm()
    if not ret:
        # アームが取得できないときにエラーの種類を用事して止める
        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
   
    # P62 へ移動する
    ret = await client.move("P62")
    if not ret:
        error = await client.error_description()
        logger.error(f"move error: {error}")
        sys.exit(1)

    # モーションが終わるまで待つ
    await client.wait_for_complete()

アームの移動2 (position)

• xyz 座標(P)

• angle 座標(J), J1-J6

を与えて移動する

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設定: motion_mode PTP を指定
    client.motion_mode = MotionMode.PTP
    # client.motion_mode = MotionMode.LINE
    # client.motion_mode = MotionMode.CIRCLE
    # client.motion_mode = MotionMode.SPLINE

    # global設定: xyz座標形式(P) で移動する
    await client.set_coord_type("P")
   
    # positon を作る (タプルでもリストでもどちらでもいい)
    pos = (
        -29.0453511306835,
        343.1440808456958,
        115.31284305934872,
        -178.28955954148657,
        -16.619847808793967,
        141.1028536689876,
    )
    # 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, "接続失敗"
    await client.wait_for_complete()

アームの移動3 (j2p, p2j)

angle から position 変更してから移動

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設定: PTP を指定
    client.motion_mode = MotionMode.PTP
    # client.motion_mode = MotionMode.LINE
    # client.motion_mode = MotionMode.CIRCLE
    # client.motion_mode = MotionMode.SPLINE

    # global設定: xyz座標形式(P) で移動する
    await client.set_coord_type("P")
   
    # positon を作る (タプルでもリストでもどちらでもいい)
    new_pos = (
        -29.0453511306835,
        343.1440808456958,
        115.31284305934872,
        -178.28955954148657,
        -16.619847808793967,
        141.1028536689876,
    )
    # P --> J 変換
    next_angle = await client.p2j(new_pos)
    
    # 指定の軸だけ 度を変える
    next_angle[4] = next_angle[4] - 15
    
    # J --> P 変換
    next_pos = await client.j2p(next2_angle)
    
    # 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, "接続失敗"
    await client.wait_for_complete()

アームの移動4 (angle)

指定Angle での move。 位置や角度の微調整をして、座標を作るのも使う

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設定: PTP を指定
    client.motion_mode = MotionMode.PTP
    # client.motion_mode = MotionMode.LINE
    # client.motion_mode = MotionMode.CIRCLE
    # client.motion_mode = MotionMode.SPLINE

    # angle を作る (タブプルでリストでもok)
    target_angle = [
        56.04999367408907,
        35.12319454476854,
        114.33533666988626,
        -4.0077312418009745,
        0,
        -25.962042331415812,
        0.0,
        0.0,
    ]
    # target_angle に移動する
    # angle だと 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, "接続失敗"
    await client.wait_for_complete()

アームの移動5 (coord_type)

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

    # angle を作る (タブプルでリストでもok)
    target_angle = [
        56.04999367408907,
        35.12319454476854,
        114.33533666988626,
        -4.0077312418009745,
        0,
        -25.962042331415812,
        0.0,
        0.0,
    ]
    # target_angle に移動する
    # angle だと 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, "接続失敗"
    await client.wait_for_complete()

アームの移動6 (path_blend)

A -> B -> C

と移動する時に、 B で位置保証性を重要視するか？ Bで速度を維持するか？ その代わりにパスをA-B混ぜて中間点のような一度経由して通るなど path が複数ある時に、そのパス操作の混ぜ方を変更する。 速度だけでも、加減速だけでもなくパス自体に対する影響パラメータ。 メーカーによって言い方が異なるが大抵は同じ意味のパラメータとなる

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設定: PTP を指定
    client.motion_mode = MotionMode.PTP
    # client.motion_mode = MotionMode.LINE
    # client.motion_mode = MotionMode.CIRCLE
    # client.motion_mode = MotionMode.SPLINE

    # target_angle に移動する
    # angle だと 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, "接続失敗"
    await client.wait_for_complete()

アームの移動7 (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)
    
    # エラーがある時は rest しないと動かすことができない
    await client.reset_error()

    # アームを動かす前には take_arm をする必要がある
    ret = await client.take_arm()
    if not ret:
        # アームが取得できないときにエラーの種類を用事して止める
        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
   
    # 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)

    # モーションが終わるまで待つ
    await client.wait_for_complete()

アームの移動8 (accel/decel)

加減速

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)
    
    # エラーがある時は rest しないと動かすことができない
    await client.reset_error()

    # アームを動かす前には take_arm をする必要がある
    ret = await client.take_arm()
    if not ret:
        # アームが取得できないときにエラーの種類を用事して止める
        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
   
    # 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)

    # モーションが終わるまで待つ
    await client.wait_for_complete()

アームの移動9 (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())

アームの移動10 (drive命令)

drive 命令で指定の軸だけ直接回すことができる

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

    # 相対移動 (J6 を -90) 回転
    await client.drive([6, -90], relative=True)

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

アームの移動11 (offset命令)

オフセット移動

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

    # 相対移動 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())

アームの移動12 (ratete)

ある軸座標を中心にして回転運動をする。 もしかして cobotta のみの関数になるかもしれない？

命令の実装(オプション形式など) まだ迷っているので変える可能性がある

# import pytest
import asyncio
import sys
import json

from pathlib import Path
import logging

# httpx のログを WARNING レベル以上にする（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()
    
    # (200,-50, 197) を原点として 45 度回転
    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())

ハンド1 (距離移動)

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

    # 位置 10mm に移動する
    await client.hand(pos=10, speed=100)
    sleep(2)
    # 現在のところ
    # ハンドは client.wait_for_complete() が使えない

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

ハンド2 (力移動、閉じる)

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

    # 6N でターゲットを掴む
    await client.hand(force=6)
    sleep(2)
    # 現在のところ
    # ハンドは client.wait_for_complete() が使えない

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

座標系を変える (work座標 定義)

work 座標系を変える。標準だとロボット座標系がそのまま work 座標系となるが、 たとえば、斜め方向に傾いたxyz座標軸を取り替えた方が、xyz指定で楽な倍がある。 相対的な移動などoffset 座標の取り方が楽になる。

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

コボッタのハードウエアログ取得

import asyncio

async def main():
    # ..
    await client.start_log()
    await client.start_servo_log() 

コボッタのハードウエア側(テーティングペンダント上）のログを取る

内部変数への読み書き1 (get_P)

メソッドを使って、コボッタ内部の指定の P を取得する

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, "接続失敗"

    logger.info("== run test()")
    ret = await client.get_P(61)

    assert "result" == "result"

内部変数への読み書き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}")

ファイルに保存してある内部変数値 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"

    # Pの後ろの数字を抽出して整数に変換
    start_num = int(re.findall(r"\d+", start)[0])
    end_num = int(re.findall(r"\d+", end)[0])

    # ファイルから読み込み
    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} は存在しません")

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

内部変数をファイルへ SAVE する

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"

    # Pの後ろの数字を抽出して整数に変換
    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")  # 例: '2025-07-17_1632'

    # ファイルに書き込む
    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())

内部の PacScript 一覧

基本的に特別な場合を除いて、推奨してないが、直接デンソーのスクリプトを使うことができる。 デンソーウェーブのコボッタ形式PacScript 一覧。 コボッタ内にあるものを表示する。

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

内部の PacScriptを run する

基本的に特別な場合を除いて、推奨してないが、直接デンソーのスクリプトを使うことができる。 デンソーウェーブのコボッタ形式PacScript 扱う。 制御クラスのテストに使える。

*) 必ず、local で直接 CobottaCtrl を使うときは、終了時には disconnect する KeyboardInterrupt を含めて異常終了するときは disconnect しておく

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"  # 拡張子はあってもなくてもいい

    # 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()

内部の PacScript を Load (Upload)

基本的に特別な場合を除いて、推奨してないが、直接デンソーのスクリプトを使うことができる。 デンソーウェーブのコボッタ形式PacScript 扱う。 制御クラスのテストに使える。

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
	
	' 下降動作    P[61]に下降
	Move L, @0 P[61]

End Sub
"""
# Pro2.pcs に file_content の内容をアップロードする
cobotta.upload_content_to_cobotta(file_content, "Pro2.pcs")
print(cobotta.get_cobotta_file_names())

内部の PacScript の Save (Download)

基本的に特別な場合を除いて、推奨してないが、直接デンソーのスクリプトを使うことができる。 デンソーウェーブのコボッタ形式PacScript 扱う。 制御クラスのテストに使える。

*) 必ず、local で直接 CobottaCtrl を使うときは、終了時には disconnect する KeyboardInterrupt を含めて異常終了するときは disconnect しておく

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")

---

作者

• Kengo NAKADA