#!/usr/bin/env python
# coding: utf-8
"""Dobot MG400 TCP/IP制御バックエンド。
Dashboard、Motion、Feedbackの3ソケットを管理し、MG400の基本制御、
状態取得、エラー取得、移動命令を一つのクラスにまとめる。
30004番ポートのリアルタイムステータスを別スレッドで受信し、最新フレームを
スレッドセーフに保持する。移動の同期/非同期は共通I/Fの意味に合わせ、DOBOT公式APIのSYNCパラメータを使う。
非同期移動を明示的に待つ場合はDOBOT公式APIのSync()を使う。
"""
import re
import threading
import time
from typing import Any, Dict, List, Optional, Sequence, Union
import numpy as np
from .dobot_api_ext import (
DobotApiDashboardExt,
DobotApiExt,
DobotApiMoveExt,
MyType,
)
from .dobot_error import DobotErrorManager
[docs]
class DobotTcp:
"""Dobot MG400のTCP/IP通信を管理する。
Dashboard API、Motion API、Feedback APIへの接続を保持し、
ロボットの制御と状態取得を提供する。
Args:
dobot_ip (str): MG400のIPアドレス。
instance_name (str): インスタンス識別名。
logger (Optional[Any]): 呼び出し側から渡されるロガー。
Noneの場合の処理は既存実装に従う。
log_level (Optional[str]): loggerがNoneの場合に使用するログレベル。
auto_connect (bool): Trueの場合、初期化時に接続する。
"""
DASHBOARD_PORT = 29999
MOVE_PORT = 30003
FEEDBACK_PORT = 30004
FEEDBACK_SIZE = 1440
FEEDBACK_MAGIC = 0x123456789ABCDEF
def __init__(
self,
dobot_ip: str = "192.168.1.6",
instance_name: str = "Ctrl",
logger: Optional[Any] = None,
log_level: Optional[str] = None,
auto_connect: bool = True,
):
if logger is None:
import logging
logging.basicConfig(level=(log_level or "INFO").upper())
logger = logging.getLogger(__name__)
self._logger = logger
self._dobot_ip = dobot_ip
self._instance_name = instance_name
self._dashboard: Optional[DobotApiDashboardExt] = None
self._move: Optional[DobotApiMoveExt] = None
self._feed: Optional[DobotApiExt] = None
self._connected = False
self._arm_taken = (
False # logical ownership only; MG400 API has no TakeArm equivalent
)
# Feedback受信スレッドと呼び出し側スレッドの間で最新状態を共有する。
self._state_lock = threading.Lock()
self._stop_event = threading.Event()
self._feed_thread: Optional[threading.Thread] = None
self._feedback: Dict[str, Any] = {}
self._feedback_ready = threading.Event()
self._last_status_summary = None
self._coord_type = "P"
self._motion_mode = "PTP"
self._path_blend = None
self._speed = 100
self._jspeed = 100
self._accel = 100.0
self._decel = None
self._current_work_no = 0
self._current_tool_no = 0
# MoveJog は開始後、MoveJog() を送るまで継続する。
self._jog_active = False
# エラー取得・アラーム解析・表示・実行時エラー判定は専用クラスへ委譲する。
self._error_manager = DobotErrorManager(
dashboard_getter=lambda: self._dashboard,
state_getter=self._state,
scalar_status=self._scalar_status,
logger=self._logger,
)
if auto_connect:
self.connect()
# ------------------------------------------------------------------
# connection / lifecycle
# ------------------------------------------------------------------
[docs]
def connect(self) -> bool:
"""MG400の3つのTCP/IPポートへ接続する。
Dashboard、Motion、Feedbackの順にソケットを生成し、接続成功後に
Feedback受信スレッドを開始する。
Returns:
bool: すべての接続と受信スレッド開始に成功した場合はTrue。
接続処理で例外が発生した場合はFalse。
"""
if self._connected:
self._logger.debug("[DobotTcp] Dobot is already connected: ip=%s", self._dobot_ip)
return True
self._logger.info("[DobotTcp] Connecting to Dobot: ip=%s", self._dobot_ip)
try:
self._dashboard = DobotApiDashboardExt(
self._dobot_ip, self.DASHBOARD_PORT, self._logger
)
self._move = DobotApiMoveExt(
self._dobot_ip, self.MOVE_PORT, self._logger
)
self._feed = DobotApiExt(
self._dobot_ip, self.FEEDBACK_PORT, self._logger
)
if MyType.itemsize != self.FEEDBACK_SIZE:
raise RuntimeError(
"Dobot feedback dtype size mismatch: "
f"MyType.itemsize={MyType.itemsize}, expected={self.FEEDBACK_SIZE}"
)
self._connected = True
self._stop_event.clear()
self._feedback_ready.clear()
self._feed_thread = threading.Thread(
target=self._feedback_loop, daemon=True
)
self._feed_thread.start()
self._logger.info(
"[DobotTcp] Connected to Dobot: dashboard=%d move=%d feedback=%d",
self.DASHBOARD_PORT,
self.MOVE_PORT,
self.FEEDBACK_PORT,
)
return True
except Exception:
self._logger.exception("[DobotTcp] Failed to connect to Dobot: ip=%s", self._dobot_ip)
self.disconnect()
return False
[docs]
def is_connected(self):
"""3本のDOBOT TCP接続が存在するかを返す。"""
if self._dashboard is None or self._move is None or self._feed is None:
return False
return True
[docs]
def reconnect(self):
"""既存接続を切断して3ポートへ再接続する。"""
self.disconnect()
return self.connect()
[docs]
def disconnect(self) -> bool:
"""MG400とのTCP/IP接続を閉じる。
Feedback受信スレッドへ停止を通知した後、Feedback、Motion、
Dashboardの各ソケットを閉じる。
Returns:
bool: 切断処理を実行した場合はTrue。
"""
self._logger.info("[DobotTcp] Disconnecting from Dobot: ip=%s", self._dobot_ip)
# MoveJog は停止命令を送るまで継続するため、Motion 接続を閉じる前に停止する。
if self._jog_active and self._move is not None:
try:
self.jog_stop()
except Exception:
self._logger.exception("[DobotTcp] Failed to stop jog before disconnect")
self._stop_event.set()
for obj in (self._feed, self._move, self._dashboard):
if obj is not None:
try:
obj.close()
except Exception:
self._logger.exception("[DobotTcp] Failed to close Dobot socket")
self._feed = self._move = self._dashboard = None
self._connected = False
self._arm_taken = False
self._feedback_ready.clear()
with self._state_lock:
self._feedback.clear()
return True
[docs]
def delete(self) -> bool:
return self.disconnect()
def __delete__(self):
return self.delete()
def _feedback_loop(self) -> None:
"""Feedbackポートから1440 byteフレームを継続受信する。
TCPでは1回のrecv()で1フレーム全体を受信できる保証がないため、
bytearrayへ蓄積してFEEDBACK_SIZEに達してから解析する。
test_valueが規定値と一致したフレームだけを最新状態として保存する。
"""
assert self._feed is not None
sock = self._feed.socket_dobot
buf = bytearray()
while not self._stop_event.is_set():
try:
chunk = sock.recv(self.FEEDBACK_SIZE - len(buf))
if not chunk:
if self._stop_event.is_set():
return
self._logger.warning(
"[DobotStatus][CONN] connection closed ip=%s port=%s",
self._dobot_ip,
self.FEEDBACK_PORT,
)
self._feed.reconnect()
sock = self._feed.socket_dobot
buf.clear()
continue
buf.extend(chunk)
if len(buf) < self.FEEDBACK_SIZE:
continue
raw = bytes(buf[: self.FEEDBACK_SIZE])
del buf[: self.FEEDBACK_SIZE]
info = np.frombuffer(raw, dtype=MyType)
if int(info["test_value"][0]) != self.FEEDBACK_MAGIC:
continue
# NumPy配列は参照のまま保持せずコピーする。
# スカラー値はそのまま保存する。
state = {}
for name in MyType.names:
value = info[name][0]
if hasattr(value, "copy"):
state[name] = value.copy()
else:
state[name] = value
with self._state_lock:
self._feedback = state
self._feedback_ready.set()
status_summary = (
self._scalar_status(state.get("robot_mode")),
self._scalar_status(state.get("EnableStatus")),
self._scalar_status(state.get("RunningStatus")),
self._scalar_status(state.get("isRunQueuedCmd")),
self._scalar_status(state.get("ErrorStatus")),
)
if status_summary != self._last_status_summary:
self._last_status_summary = status_summary
self._logger.info(
"[DobotStatus] mode=%s enable=%s running=%s queued=%s error=%s",
status_summary[0],
status_summary[1],
status_summary[2],
status_summary[3],
status_summary[4],
)
except Exception:
if not self._stop_event.is_set():
self._logger.exception("[DobotTcp] Feedback receive loop failed")
time.sleep(0.02)
@staticmethod
def _scalar_status(value):
"""NumPy配列をログ表示しやすい単一値へ変換する。"""
if value is None:
return None
try:
if hasattr(value, "__len__"):
return int(value[0])
return int(value)
except Exception:
return value
def _state(self, key: str, default=None, *, timeout: float = 2.0):
"""最新の有効Feedbackから指定項目を取得する。
接続直後の競合で未取得値を返さないため、最初の有効な1440 byte
Feedbackフレームが到着するまで最大 ``timeout`` 秒待つ。
Args:
key (str): ``MyType``で定義されるFeedback項目名。
default (Any): フレーム内に項目が存在しない場合の戻り値。
timeout (float): 初回Feedbackを待つ最大秒数。
Returns:
Any: 最新Feedbackの値、または項目が存在しない場合はdefault。
Raises:
RuntimeError: Feedbackが取得できない場合。
"""
if not self._feedback_ready.wait(timeout=timeout):
raise RuntimeError(
f"Dobot feedback is unavailable; cannot read state field: {key}"
)
with self._state_lock:
return self._feedback.get(key, default)
@staticmethod
def _numbers(reply: str) -> List[int]:
"""Dobot API応答文字列から整数列を抽出する。
Args:
reply (str): Dobot APIから返された応答文字列。
Returns:
List[int]: 応答中に含まれる整数を出現順に並べたリスト。
"""
return [int(x) for x in re.findall(r"-?\d+", reply or "")]
@staticmethod
def _reply_values(reply: str) -> List[float]:
"""Dashboard応答の最初の ``{...}`` から数値列を取り出す。
GetPose/GetAngle/RobotMode の応答
``ErrorID,{values...},Command();`` を対象とする。
"""
match = re.search(r"\{([^{}]*)\}", reply or "")
if match is None:
return []
return [
float(value)
for value in re.findall(
r"[-+]?(?:\d+(?:\.\d*)?|\.\d+)(?:[eE][-+]?\d+)?",
match.group(1),
)
]
def _dashboard_values(self, command_name: str, reply: str) -> List[float]:
"""Dashboardの状態取得応答を検証して数値列として返す。"""
if not self._ok(reply):
raise RuntimeError(f"{command_name} failed: {reply}")
values = self._reply_values(reply)
if not values:
raise RuntimeError(f"{command_name} returned no data: {reply}")
return values
@classmethod
def _ok(cls, reply: str) -> bool:
"""Dobot API応答が成功か判定する。
Args:
reply (str): Dobot APIから返された応答文字列。
Returns:
bool: 先頭のリターンコードが0ならTrue。
"""
nums = cls._numbers(reply)
return bool(nums) and nums[0] == 0
def _handle_command_reply(
self,
command_name: str,
reply: str,
*,
read_error_id: bool = True,
) -> bool:
"""Dobotコマンド応答を判定し、失敗時のログ処理を行う。
Args:
command_name (str): ログへ表示するDobotコマンド名。
reply (str): Dobot APIから返された応答文字列。
read_error_id (bool): 失敗時にGetErrorID()を呼ぶ場合はTrue。
Returns:
bool: Dobotのリターンコードが0の場合はTrue、それ以外はFalse。
"""
if self._ok(reply):
self._logger.debug("[DobotTcp] %s succeeded: %s", command_name, reply)
return True
self._logger.error("[DobotTcp] %s failed: response=%s", command_name, reply)
if read_error_id:
self._logger.error(
"[DobotTcp] %s failed; calling GetErrorID()",
command_name,
)
errors = self.get_error()
if errors:
for error_id in errors:
self._logger.error(
"[DobotTcp] %s: error_id=%s description=%s",
command_name,
error_id,
self.get_error_description(error_id),
)
else:
self._logger.error(
"[DobotTcp] %s: GetErrorID() returned no error ID",
command_name,
)
return False
def _motion_error_active(
self, command_name: str, *, query_error_id: bool = False
) -> bool:
"""実行時アラーム判定をエラー管理クラスへ委譲する。"""
return self._error_manager.motion_error_active(
command_name, query_error_id=query_error_id
)
def _wait_motion_start_or_error(
self, command_name: str, timeout: float = 0.25, interval: float = 0.01
) -> bool:
"""非同期モーション直後のエラー監視をエラー管理クラスへ委譲する。"""
return self._error_manager.wait_motion_start_or_error(
command_name, timeout=timeout, interval=interval
)
def _unsupported(self, name: str):
self._logger.error("[DobotTcp] Unsupported interface called: %s", name)
raise NotImplementedError(
f"{name} has no equivalent in the attached MG400 TCP/IP API"
)
# ------------------------------------------------------------------
# Arm ownership / motor
# ------------------------------------------------------------------
[docs]
def take_arm(self) -> bool:
"""アーム制御権取得I/Fを処理する。
MG400のTCP/IP APIには独立した制御権取得操作がないため、
ロボットへコマンドは送信せず成功として扱う。
EnableRobot()とは役割が異なるため、ここでは呼び出さない。
"""
self._logger.debug("[DobotTcp] take_arm(): no command required")
return True
[docs]
def give_arm(self) -> bool:
"""アーム制御権解放I/Fを処理する。
MG400のTCP/IP APIには独立した制御権解放操作がないため、
ロボットへコマンドは送信せず成功として扱う。
DisableRobot()とは役割が異なるため、ここでは呼び出さない。
"""
self._logger.debug("[DobotTcp] give_arm(): no command required")
return True
[docs]
def release_arm(self) -> bool:
"""give_arm()の別名。"""
return self.give_arm()
[docs]
def motor_on(self) -> bool:
"""駆動を有効にする。MG400ではEnableRobot()に対応する。"""
self._logger.info("[DobotTcp] Calling EnableRobot()")
reply = self._dashboard.EnableRobot()
return self._handle_command_reply("EnableRobot", reply)
[docs]
def motor_off(self) -> bool:
"""駆動を無効にする。MG400ではDisableRobot()に対応する。"""
self._logger.info("[DobotTcp] Calling DisableRobot()")
reply = self._dashboard.DisableRobot()
return self._handle_command_reply("DisableRobot", reply)
[docs]
def turn_on_motor(self) -> bool:
"""旧I/F互換。motor_on()へ委譲する。"""
return self.motor_on()
[docs]
def turn_off_motor(self) -> bool:
"""旧I/F互換。motor_off()へ委譲する。"""
return self.motor_off()
@property
def servo_on(self):
"""MG400の駆動Enable状態を返す。
Feedbackの ``EnableStatus`` を使用し、``EnableRobot()`` /
``DisableRobot()`` による現在のEnable状態を取得する。
``RobotMode`` や ``busy_status`` とは別の状態である。
"""
return bool(self._scalar_status(self._state("EnableStatus")))
# ------------------------------------------------------------------
# stop / reset
# ------------------------------------------------------------------
[docs]
def reset_error(self) -> bool:
"""現在発生しているアラームをクリアする。"""
self._logger.info("[DobotTcp] Calling ClearError()")
reply = self._dashboard.ClearError()
return self._handle_command_reply(
"ClearError",
reply,
read_error_id=False,
)
[docs]
def check_error(self) -> bool:
"""現在エラーがなければTrueを返す。"""
return self.get_error_count() == 0
[docs]
def halt(self) -> bool:
"""即時停止を行う。
現在利用しているMG400 APIでは、このI/Fに必要な停止動作と同義であることを
確認できるコマンドがないため未対応とする。ResetRobot()は代用しない。
Raises:
NotImplementedError: 対応するMG400実装がない。
"""
self._unsupported("halt")
[docs]
def stop(self):
"""ロボットの停止処理を行う。
このI/Fが要求する停止シーケンスをMG400で同じ意味のまま保証できないため
未対応とする。``halt()`` の単純な別名にはしない。
"""
self._unsupported("stop")
[docs]
def task_stop(self) -> bool:
"""タスク実行を停止する。
MG400の ``StopScript()`` とこのI/Fのタスク停止が同義であることを
確認できないため未対応とする。MG400固有のスクリプト停止は
``stop_script()`` として分離する。
"""
self._unsupported("task_stop")
[docs]
def hand_stop(self) -> bool:
self._unsupported("hand_stop")
# ------------------------------------------------------------------
# state / error
# ------------------------------------------------------------------
@property
def current_position(self):
return self.get_current_position()
@property
def current_angle(self):
return self.get_current_angle()
[docs]
def get_current_position(self):
"""現在TCP位置を6成分座標として同期取得して返す。
MG400のDashboard API ``GetPose()`` を直接使用する。4軸MG400の
[x, y, z, r] は公開I/Fで [x, y, z, 0.0, 0.0, rz] として返す。
"""
values = self._dashboard_values("GetPose", self._dashboard.GetPose())
if len(values) < 4:
raise RuntimeError(f"GetPose returned too few values: {values}")
rz_index = 5 if len(values) >= 6 else 3
return [
float(values[0]),
float(values[1]),
float(values[2]),
0.0,
0.0,
float(values[rz_index]),
]
[docs]
def get_current_pose(self):
"""get_current_position() の互換別名。"""
return self.get_current_position()
[docs]
def get_current_angle(self):
"""現在関節値を6成分として同期取得して返す。
MG400のDashboard API ``GetAngle()`` を直接使用する。MG400は4軸
なのでJ5/J6は0.0として公開する。
"""
values = self._dashboard_values("GetAngle", self._dashboard.GetAngle())
if len(values) < 4:
raise RuntimeError(f"GetAngle returned too few values: {values}")
return [
float(values[0]),
float(values[1]),
float(values[2]),
float(values[3]),
0.0,
0.0,
]
@property
def busy_status(self):
"""MG400が現在動作中かを真偽値で返す。
Feedbackの ``RunningStatus`` と ``JogStatus`` を参照し、
キュー動作中またはJOG中であればTrueを返す。
"""
running = bool(self._scalar_status(self._state("RunningStatus")))
jog = bool(self._scalar_status(self._state("JogStatus")))
return running or jog
[docs]
def is_running(self):
return self.busy_status
[docs]
def state(self):
"""現在の実行状態を返す。
MG400 Feedbackの ``RunningStatus`` を返す。
``busy_status`` はJOGも含む「ロボットが動作中」の判定であり、
``state()`` とは意図的に分ける。DOBOT固有のRobotMode番号が必要な場合は
``get_robot_mode()`` を使用する。
"""
return bool(self._scalar_status(self._state("RunningStatus")))
[docs]
def get_robot_mode(self):
"""DOBOT固有: Dashboard ``RobotMode()`` のモード番号を返す。"""
values = self._dashboard_values("RobotMode", self._dashboard.RobotMode())
return int(values[0])
def _set_motion_target(self, target: Sequence[float], *, joint: bool) -> None:
"""直近の移動目標を保存する。
``@E`` 相当の実位置到達確認が必要な場合にだけ使用する。
``get_motion_complete()`` の一般的な完了判定には使用しない。
"""
self._motion_target = [float(x) for x in target[:4]]
if joint:
self._motion_target_state_key = "q_actual"
else:
self._motion_target_state_key = "tool_vector_actual"
def _target_reached(self) -> bool:
"""保存済み目標へ実位置が到達したか確認する。
DOBOT公式サンプル ``WaitArrive()`` と同じく、実座標と目標座標の
各成分差が1.0以下かを確認する。
"""
target = getattr(self, "_motion_target", None)
state_key = getattr(self, "_motion_target_state_key", None)
if target is None or state_key is None:
return False
actual = self._state(state_key)
if actual is None:
return False
return all(abs(float(actual[i]) - target[i]) <= 1.0 for i in range(4))
[docs]
def get_motion_complete(self):
"""現在ロボットがモーション実行中でないかを返す。
モーション実行状態の判定には、DOBOTのリアルタイムステータス
``RunningStatus`` を使う。
"""
running = self.busy_status
if running is None:
return False
return not bool(running)
[docs]
def motion_complete(self):
"""``get_motion_complete()`` の別名。"""
return self.get_motion_complete()
[docs]
def wait_for_complete(self, timeout=None, interval=0.02) -> bool:
"""キューに入っているモーションがすべて完了するまで待つ。
MG400のSync()を使用する。Sync()は、先行してキューへ登録された
モーションの実行がすべて完了してから応答する。
``timeout`` と ``interval`` は共通I/Fのシグネチャを維持するため受け取るが、
MG400のSync()自体には対応する引数がないため使用しない。
Returns:
bool: Sync()が正常終了した場合True。
"""
self._logger.info("[MOTION WAIT] Sync() start")
reply = self._move.Sync()
result = self._handle_command_reply("Sync", reply)
if not result:
return False
# Sync()の応答コード0は待機コマンド自体の成功であり、
# 先行モーションがアラームなしで完了したこととは別である。
if self._motion_error_active("Sync", query_error_id=True):
return False
self._logger.info("[MOTION WAIT] Sync() complete")
return True
def _wait_encoder_target(self) -> bool:
"""``@E`` 相当として実位置が保存済み目標へ到達するまで待つ。"""
self._logger.info("[MOTION @E] encoder target wait start")
while not self._target_reached():
time.sleep(0.01)
self._logger.info("[MOTION @E] encoder target reached")
return True
[docs]
def get_error(self):
"""現在発生しているDobotエラーIDを取得する。"""
return self._error_manager.get_error()
[docs]
def get_error_code(self):
"""現在の先頭エラーコードを取得する。"""
return self._error_manager.get_error_code()
@property
def error_code(self):
return self._error_manager.get_error_code()
[docs]
def get_error_count(self):
return self._error_manager.get_error_count()
[docs]
def get_error_info(self, index: int):
return self._error_manager.get_error_info(index)
[docs]
def get_current_error_info(self):
return self._error_manager.get_current_error_info()
[docs]
def get_error_description(self, error_code: int) -> str:
return self._error_manager.get_error_description(error_code)
@property
def error_description(self):
return self._error_manager.error_description
[docs]
def print_error(self, e) -> bool:
return self._error_manager.print_error(e)
[docs]
def print_error_info(self, e) -> bool:
return self._error_manager.print_error_info(e)
# ------------------------------------------------------------------
# coordinate / tool selection
# ------------------------------------------------------------------
[docs]
def change_work(self, work_no: int) -> bool:
ok = self._ok(self._dashboard.User(int(work_no)))
if ok:
self._current_work_no = int(work_no)
return ok
[docs]
def get_current_work_no(self):
"""現在のユーザー座標系番号を取得する。
Returns:
int: フィードバックで取得したユーザー座標系番号。
未受信時は最後に設定した番号。
"""
value = self._state("userCoordinate")
if value is None:
return self._current_work_no
return int(value[0])
[docs]
def get_work(self, work_no=None):
"""ユーザー座標系定義を取得する。
MG400のUser座標系定義を、30004番ポートのリアルタイムステータス
``user`` から取得する。
``work_no`` がNoneの場合は現在選択中のUser座標系をそのまま返す。
番号を指定した場合、MG400の現在のPython APIには任意番号の
User座標系定義だけを直接取得する命令がないため、一時的に
User(work_no)で選択して ``user`` を取得し、その後で元のUserへ戻す。
Args:
work_no (int or None): 取得するWork/User座標系番号。
Noneの場合は現在選択中の座標系。
Returns:
list: [x, y, z, rx, ry, rz] の6成分座標系定義。
リアルタイムステータスを取得できない場合はNone。
"""
previous_work_no = self.get_current_work_no()
requested_work_no = work_no
if requested_work_no is not None:
requested_work_no = int(requested_work_no)
if requested_work_no < 0 or requested_work_no > 9:
raise ValueError("work_no must be in the range 0..9")
if previous_work_no != requested_work_no:
if not self.change_work(requested_work_no):
return None
started = time.monotonic()
while True:
current_work_no = self.get_current_work_no()
if current_work_no == requested_work_no:
break
if time.monotonic() - started >= 1.0:
break
time.sleep(0.01)
value = self._state("user")
result = None
if value is not None and len(value) >= 6:
result = [
float(value[0]),
float(value[1]),
float(value[2]),
float(value[3]),
float(value[4]),
float(value[5]),
]
if (
requested_work_no is not None
and previous_work_no is not None
and previous_work_no != requested_work_no
):
self.change_work(previous_work_no)
return result
[docs]
def set_work(self, work_no, work_define, work_attribute=0) -> bool:
"""ユーザー座標系を定義する。
MG400のUser座標系を ``SetUser(index,{x,y,z,r})`` で定義する。
``work_define`` は
[x, y, z, rx, ry, rz] として受け取り、MG400の4軸座標系
[x, y, z, r] へ変換する。MG400のrにはrzを対応させる。
rx/ryが0でない座標系はMG400では表現できないため拒否する。
``work_attribute`` は公開I/Fとの互換性のため受け取るが、
MG400の座標系定義には使用しない。
注意:
MG400マニュアル上、SetUserによる変更は現在の
プロジェクト実行中のみ有効で、プロジェクト停止後は
変更前の値へ戻る。
Args:
work_no (int): Work/User座標系番号。0..9。
work_define (list or tuple): [x,y,z,rx,ry,rz]。
work_attribute: 互換引数。現在は未使用。
Returns:
bool: SetUserが成功した場合True。
"""
work_no = int(work_no)
if work_no < 0 or work_no > 9:
raise ValueError("work_no must be in the range 0..9")
table = self._p_to_mg400(work_define, "work_define")
command = "SetUser(%d,{%.15g,%.15g,%.15g,%.15g})" % (
work_no,
table[0],
table[1],
table[2],
table[3],
)
self._logger.debug("[DobotTcp] Calling %s", command)
reply = self._dashboard.sendRecvMsg(command)
ok = self._ok(reply)
if not ok:
self._logger.error(
"[DobotTcp] SetUser() failed: response=%s",
reply,
)
return ok
@property
def coord_type(self):
return self._coord_type
@coord_type.setter
def coord_type(self, value):
self.set_coord_type(value)
[docs]
def set_coord_type(self, coord_type: str) -> bool:
value = str(coord_type).upper()
if value not in ("P", "J"):
return False
self._coord_type = value
return True
[docs]
def get_coord_type(self) -> str:
return self._coord_type
# ------------------------------------------------------------------
# motion mode / blend
# ------------------------------------------------------------------
@property
def motion_mode(self):
return self._motion_mode
@motion_mode.setter
def motion_mode(self, value):
self.set_motion_mode(value)
[docs]
def set_motion_mode(self, mode) -> bool:
mapping = {1: "PTP", 2: "LINE", 3: "CIRCLE", 4: "SPLINE"}
value = mapping.get(mode, str(getattr(mode, "value", mode)).upper())
if value not in ("PTP", "LINE", "CIRCLE", "SPLINE"):
return False
self._motion_mode = value
return True
[docs]
def get_motion_mode(self):
return self._motion_mode
@property
def path_blend(self):
return self._path_blend
@path_blend.setter
def path_blend(self, value):
self.set_path_blend(value)
[docs]
def set_path_blend(self, path_blend):
"""パス遷移条件を設定する。
現在の ``path_blend`` の値をDOBOTのCP値へ安全に変換する規則を
定義していないため、Noneだけを「指定なし」として受理する。
"""
if path_blend is None:
self._path_blend = None
return True
self._logger.error(
"[DobotTcp] path_blend=%s is not implemented: "
"path_blend cannot be mapped safely to DOBOT CP",
path_blend,
)
return False
[docs]
def get_path_blend(self):
return self._path_blend
# ------------------------------------------------------------------
# speed / acceleration
# ------------------------------------------------------------------
@property
def speed(self):
return self._speed
@property
def jspeed(self):
return self._jspeed
@property
def accel(self):
return self._accel
@property
def decel(self):
return self.get_decel()
@property
def extspeed(self):
return self.get_extspeed()
@property
def extaccel(self):
return self.get_extaccel()
@property
def extdecel(self):
return self.get_extdecel()
[docs]
def set_speed(self, speed):
"""MG400の直線移動速度率を設定する。
Dashboard APIの ``SpeedL`` を使用する。
グローバル速度倍率は ``SpeedFactor`` を使用する別I/Fで管理する。
"""
value = int(speed)
if not 1 <= value <= 100:
raise ValueError("speed must be in the range 1..100")
ok = self._handle_command_reply(
"SpeedL", self._dashboard.SpeedL(value), read_error_id=False
)
if ok:
self._speed = float(value)
return ok
[docs]
def get_speed(self):
"""最後に共通I/Fから設定した内部移動速度[%]を返す。
MG400 TCP/IPには ``SpeedL`` のreadback APIがないため、
DobotTcpが最後に設定した値を保持して返す。実速度計測値ではない。
"""
return self._speed
[docs]
def set_jspeed(self, speed: int) -> bool:
value = int(speed)
ok = self._ok(self._dashboard.SpeedJ(value))
if ok:
self._jspeed = value
return ok
[docs]
def get_jspeed(self):
return self._jspeed
[docs]
def set_accel(self, accel: float) -> bool:
value = int(accel)
ok1 = self._ok(self._dashboard.AccL(value))
ok2 = self._ok(self._dashboard.AccJ(value))
if ok1 and ok2:
self._accel = float(value)
return ok1 and ok2
[docs]
def get_accel(self):
return self._accel
[docs]
def set_decel(self, decel):
"""独立した減速度設定。
現在使用しているMG400 APIには独立した減速度指定を確認できないため、
Accへ読み替えず未対応とする。
"""
self._logger.error(
"[DobotTcp] independent decel is not implemented on MG400: %s",
decel,
)
return False
[docs]
def get_decel(self):
self._unsupported("get_decel")
[docs]
def set_extspeed(self, speed):
"""MG400のグローバル速度倍率を設定する。
Dashboard APIの ``SpeedFactor`` を使用する。
"""
value = int(speed)
if not 1 <= value <= 100:
raise ValueError("extspeed must be in the range 1..100")
return self._handle_command_reply(
"SpeedFactor", self._dashboard.SpeedFactor(value), read_error_id=False
)
[docs]
def get_extspeed(self):
"""MG400 Feedbackのglobal velocity ratio[%]を返す。"""
return float(self._scalar_status(self._state("velocityRatio")))
[docs]
def set_extaccel(self, accel):
self._unsupported("set_extaccel")
[docs]
def get_extaccel(self):
self._unsupported("get_extaccel")
[docs]
def set_extdecel(self, decel):
self._unsupported("set_extdecel")
[docs]
def get_extdecel(self):
self._unsupported("get_extdecel")
# ------------------------------------------------------------------
# motion
# ------------------------------------------------------------------
@staticmethod
def _vector6(values, name):
"""6成分数値列をfloat listへ正規化する。"""
if not isinstance(values, (list, tuple)):
raise ValueError("%s must be list or tuple" % name)
if len(values) < 6:
raise ValueError("%s must contain six numeric values" % name)
return [float(values[index]) for index in range(6)]
def _p_to_mg400(self, values, name="target"):
"""P6 [x,y,z,rx,ry,rz] をMG400 [x,y,z,r]へ変換する。"""
p = self._vector6(values, name)
if abs(p[3]) > 1.0e-9 or abs(p[4]) > 1.0e-9:
raise ValueError(
"%s contains rx/ry which MG400 cannot realize: rx=%s ry=%s"
% (name, p[3], p[4])
)
return [p[0], p[1], p[2], p[5]]
def _j_to_mg400(self, values, name="target"):
"""J6 [j1..j6] をMG400 [j1..j4]へ変換する。"""
j = self._vector6(values, name)
if abs(j[4]) > 1.0e-9 or abs(j[5]) > 1.0e-9:
raise ValueError(
"%s contains J5/J6 which MG400 cannot realize: J5=%s J6=%s"
% (name, j[4], j[5])
)
return [j[0], j[1], j[2], j[3]]
@staticmethod
def _mg400_p_to_public(values):
"""MG400 [x,y,z,r] を公開P6へ変換する。"""
return [
float(values[0]),
float(values[1]),
float(values[2]),
0.0,
0.0,
float(values[3]),
]
@staticmethod
def _mg400_j_to_public(values):
"""MG400 [j1..j4] を公開J6へ変換する。"""
return [
float(values[0]),
float(values[1]),
float(values[2]),
float(values[3]),
0.0,
0.0,
]
@staticmethod
def _parse_reply_values(reply):
"""DOBOT応答 0,{...},Command(...); の数値部を取得する。"""
if not isinstance(reply, str):
return False
match = re.search(r"\{([^}]*)\}", reply)
if match is None:
return False
text = match.group(1).strip()
if not text:
return []
values = []
for item in text.split(","):
values.append(float(item.strip()))
return values
def _normalize_offset(self, offset):
"""offsetを6成分へ正規化する。
list/tupleの3成分は [dx,dy,dz,0,0,0] とし、6成分はそのまま使う。
``P(...)`` または ``(...)`` の文字列形式も受け付ける。
"""
values = offset
if isinstance(offset, str):
text = offset.strip()
if text.upper().startswith("P(") and text.endswith(")"):
text = text[2:-1]
elif text.startswith("(") and text.endswith(")"):
text = text[1:-1]
values = [item.strip() for item in text.split(",") if item.strip()]
if not isinstance(values, (list, tuple)):
raise ValueError("offset must be str, list, or tuple")
if len(values) == 3:
return [
float(values[0]), float(values[1]), float(values[2]),
0.0, 0.0, 0.0,
]
if len(values) >= 6:
return [float(values[index]) for index in range(6)]
raise ValueError("offset must contain three or six numeric values")
[docs]
def dev(self, base, offset):
"""MG400で実現可能な範囲のDev(P姿勢変位)を計算する。
MG400はP=[x,y,z,rx,ry,rz]のうちrx/ryを実現できない。
したがってbase/offset双方のrx/ryが0の場合だけ、
x,y,z,rzの変位を合成する。
rx/ryを黙って捨てることはしない。
"""
try:
base6 = self._vector6(base, "base")
offset6 = self._vector6(offset, "offset")
if abs(base6[3]) > 1.0e-9:
raise ValueError(
"base rx is not realizable on MG400"
)
if abs(base6[4]) > 1.0e-9:
raise ValueError(
"base ry is not realizable on MG400"
)
if abs(offset6[3]) > 1.0e-9:
raise ValueError(
"offset rx is not realizable on MG400"
)
if abs(offset6[4]) > 1.0e-9:
raise ValueError(
"offset ry is not realizable on MG400"
)
return [
base6[0] + offset6[0],
base6[1] + offset6[1],
base6[2] + offset6[2],
0.0,
0.0,
base6[5] + offset6[5],
]
except Exception as exc:
self._logger.error(
"[DobotTcp] dev failed: %s",
exc,
)
return False
[docs]
def p2j(self, p_type_pos):
"""P6をMG400逆運動学へ渡し、J6で返す。"""
try:
p = self._p_to_mg400(p_type_pos, "p_type_pos")
reply = self._dashboard.InverseSolution(
p[0], p[1], p[2], p[3], 0, 0
)
if not self._ok(reply):
return False
values = self._parse_reply_values(reply)
if values is False or len(values) < 4:
self._logger.error(
"[DobotTcp] InverseSolution parse failed: %s",
reply,
)
return False
return self._mg400_j_to_public(values)
except Exception as exc:
self._logger.error("[DobotTcp] p2j failed: %s", exc)
return False
[docs]
def j2p(self, j_type_pos):
"""J6をMG400順運動学へ渡し、P6で返す。"""
try:
j = self._j_to_mg400(j_type_pos, "j_type_pos")
reply = self._dashboard.PositiveSolution(
j[0], j[1], j[2], j[3], 0, 0
)
if not self._ok(reply):
return False
values = self._parse_reply_values(reply)
if values is False or len(values) < 4:
self._logger.error(
"[DobotTcp] PositiveSolution parse failed: %s",
reply,
)
return False
return self._mg400_p_to_public(values)
except Exception as exc:
self._logger.error("[DobotTcp] j2p failed: %s", exc)
return False
[docs]
def move(
self,
target,
*,
motion_mode=None,
coord_type=None,
path_blend=None,
offset=None,
pose=None,
speed=None,
accel=None,
decel=None,
is_async=True,
):
"""公開move I/FをMG400の移動コマンドへ変換する。
公開座標:
P = [x, y, z, rx, ry, rz]
J = [j1, j2, j3, j4, j5, j6]
MG400内部:
P = [x, y, z, r] (r <- rz)
J = [j1, j2, j3, j4]
同期契約:
is_async=True:
TCP/IPのMovJ/MovL/JointMovJを送信し、受付応答後に戻る。
is_async=False:
TCP/IPのMovJ/MovL/JointMovJを送信した後、Sync()を送信して
投入済みモーションの実行完了まで待ってから戻る。
DobotStudio ProスクリプトのMovJ/MovLにあるSYNCオプションは、
TCP/IP 30003ポートのMovJ/MovLオプションではないため送信しない。
"""
if not isinstance(target, (list, tuple)):
self._logger.error(
"[DobotTcp] target must be list or tuple"
)
return False
if path_blend is not None:
if not self.set_path_blend(path_blend):
return False
if pose is not None:
self._logger.error(
"[DobotTcp] pose/fig=%s is not implemented on MG400",
pose,
)
return False
if accel is not None:
if not self.set_accel(accel):
return False
if decel is not None:
if not self.set_decel(decel):
return False
if motion_mode is None:
mode = self._motion_mode
else:
mode = str(
getattr(motion_mode, "value", motion_mode)
).upper()
if coord_type is None:
ctype = self._coord_type
else:
ctype = str(coord_type).upper()
try:
public_target = self._vector6(
target,
"target",
)
if ctype == "P":
if offset is not None:
public_target = self.dev(
public_target,
offset,
)
if public_target is False:
return False
mg_target = self._p_to_mg400(
public_target,
"target",
)
elif ctype == "J":
if offset is not None:
self._logger.error(
"[DobotTcp] offset is valid only for P coordinates"
)
return False
if mode != "PTP":
self._logger.error(
"[DobotTcp] MG400 cannot preserve coord_type=J "
"with motion_mode=%s",
mode,
)
return False
mg_target = self._j_to_mg400(
public_target,
"target",
)
else:
self._logger.error(
"[DobotTcp] coord_type must be P or J: %s",
ctype,
)
return False
except Exception as exc:
self._logger.error(
"[DobotTcp] invalid move target: %s",
exc,
)
return False
if speed is not None:
# PTP/J動作の速度率はSpeedJ、LINE動作はSpeedLへ写像する。
if ctype == "J" or mode == "PTP":
if not self.set_jspeed(speed):
return False
else:
if not self.set_speed(speed):
return False
# TCP/IP 30003ポートのMovJ/MovL/JointMovJには、
# DobotStudio ProスクリプトのSYNCオプションを付加しない。
# blockingは、移動コマンド受付後に別コマンドSync()で実現する。
dyn_params = []
if ctype == "J":
command_name = "JointMovJ"
self._set_motion_target(
mg_target,
joint=True,
)
reply = self._move.JointMovJ(
*(mg_target + dyn_params)
)
elif mode == "PTP":
command_name = "MovJ"
self._set_motion_target(
mg_target,
joint=False,
)
reply = self._move.MovJ(
*(mg_target + dyn_params)
)
elif mode == "LINE":
command_name = "MovL"
self._set_motion_target(
mg_target,
joint=False,
)
reply = self._move.MovL(
*(mg_target + dyn_params)
)
elif mode == "CIRCLE":
self._logger.error(
"[DobotTcp] move(CIRCLE) cannot be represented by "
"a single target on MG400 Arc/Circle"
)
return False
elif mode == "SPLINE":
self._logger.error(
"[DobotTcp] move(SPLINE) is not implemented on MG400"
)
return False
else:
self._logger.error(
"[DobotTcp] unsupported motion_mode: %s",
mode,
)
return False
self._logger.info(
"[MOTION] command=%s public_target=%s "
"mg400_target=%s coord=%s async=%s",
command_name,
public_target,
mg_target,
ctype,
bool(is_async),
)
result = self._handle_command_reply(
command_name,
reply,
)
if not result:
return False
if bool(is_async):
# TCP応答コード0は受付成功に過ぎない。実行開始前にリミット等で
# ErrorStatusへ落ちるケースを短時間監視し、失敗ならFalseを返す。
return self._wait_motion_start_or_error(command_name)
return self.wait_for_complete()
[docs]
def jog(self, axis, direction, speed, step=None):
"""共通JOG I/FをMG400へ変換する。
Args:
axis (int): 関節番号。MG400では1..4。
direction (int): 正方向は1、負方向は-1。
speed (float): DOBOTの速度倍率[%]。1..100。
step (Optional[float]): Noneなら連続JOG。値を指定した場合は
その絶対値[deg]だけ指定関節を相対移動する。
Returns:
bool: コマンド受付または有限ステップ移動に成功した場合True。
Notes:
連続JOGにはDOBOT公式TCP/IP APIの ``MoveJog`` を使用する。
``MoveJog("J1+")`` 等で開始し、``MoveJog()`` で停止する。
MoveJogには個別速度引数がないため、speedはグローバル速度倍率
``SpeedFactor`` に写像する。
"""
try:
axis = int(axis)
if axis < 1 or axis > 4:
raise ValueError("MG400 jog axis must be in the range 1..4")
direction = int(direction)
if direction not in (-1, 1):
raise ValueError("direction must be 1 or -1")
speed = abs(float(speed))
if speed < 1.0 or speed > 100.0:
raise ValueError("speed must be in the range 1..100")
if step is not None:
step = abs(float(step))
if step == 0.0:
raise ValueError("step must be greater than 0")
# step指定時は連続Jogではなく、同じ共通I/Fの有限相対移動として扱う。
# DOBOT側の速度指定は既存drive()のSpeedJ[%]に委譲する。
return self.drive(
(axis, direction * step),
relative=True,
speed=speed,
is_async=False,
)
if self._move is None:
self._logger.error("[DobotTcp] jog failed: motion connection is not available")
return False
# 連続JOG中に別のJOGを開始する場合は、先に現在のJOGを停止する。
if self._jog_active:
if not self.jog_stop():
return False
if not self.set_extspeed(speed):
return False
axis_id = "J%d%s" % (axis, "+" if direction > 0 else "-")
reply = self._move.MoveJog(axis_id)
ok = self._handle_command_reply("MoveJog", reply)
if not ok:
return False
self._jog_active = True
self._logger.info(
"[MOTION] command=MoveJog axis=%d direction=%d speed=%s",
axis,
direction,
speed,
)
return True
except Exception as exc:
self._logger.error("[DobotTcp] jog failed: %s", exc)
return False
[docs]
def jog_stop(self):
"""連続JOGを停止する。
DOBOT公式TCP/IP APIでは引数なしの ``MoveJog()`` がJOG停止命令。
"""
if self._move is None:
self._logger.error("[DobotTcp] jog_stop failed: motion connection is not available")
return False
try:
reply = self._move.MoveJog()
ok = self._handle_command_reply("MoveJog", reply)
self._jog_active = False
if ok:
self._logger.info("[MOTION] command=MoveJog stop")
return bool(ok)
except Exception as exc:
self._logger.error("[DobotTcp] jog_stop failed: %s", exc)
return False
[docs]
def drive(
self,
target,
*,
relative=False,
path_blend=None,
speed=None,
is_async=False,
):
"""軸番号と値によるDrive共通I/F。"""
try:
if path_blend is not None:
if not self.set_path_blend(path_blend):
return False
if isinstance(target, str):
pairs = []
for match in re.finditer(
r"\(\s*([+-]?\d+)\s*,\s*"
r"([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s*\)",
target,
):
pairs.append(
(
int(match.group(1)),
float(match.group(2)),
)
)
if len(pairs) == 0:
raise ValueError(
"target must contain Drive-style axis pairs"
)
elif isinstance(target, (list, tuple)):
if len(target) == 2:
if isinstance(target[0], (int, float)):
if isinstance(target[1], (int, float)):
pairs = [
(
int(target[0]),
float(target[1]),
)
]
else:
raise ValueError(
"target must be an axis/value pair"
)
else:
pairs = []
for item in target:
if not isinstance(item, (list, tuple)):
raise ValueError(
"target items must be axis/value pairs"
)
if len(item) != 2:
raise ValueError(
"target items must be axis/value pairs"
)
pairs.append(
(
int(item[0]),
float(item[1]),
)
)
else:
pairs = []
for item in target:
if not isinstance(item, (list, tuple)):
raise ValueError(
"target items must be axis/value pairs"
)
if len(item) != 2:
raise ValueError(
"target items must be axis/value pairs"
)
pairs.append(
(
int(item[0]),
float(item[1]),
)
)
else:
raise ValueError(
"target must be str, tuple, or list of tuples"
)
current = self.get_current_angle()
if current is False:
return False
joints = list(current)
for axis, value in pairs:
index = int(axis) - 1
if index < 0 or index >= 6:
raise ValueError(
"axis must be 1..6: %s" % axis
)
if index >= 4:
if abs(float(value)) > 1.0e-9:
raise ValueError(
"MG400 cannot drive axis %s" % axis
)
if relative:
joints[index] = (
joints[index] + float(value)
)
else:
joints[index] = float(value)
return self.move(
joints,
motion_mode="PTP",
coord_type="J",
path_blend=None,
speed=speed,
is_async=is_async,
)
except Exception as exc:
self._logger.error(
"[DobotTcp] drive failed: %s",
exc,
)
return False
[docs]
def approach(
self,
lComp,
vntPoseBase,
vntPoseLen,
bstrOpt=None,
):
"""基準P座標へオフセットを加えて移動する。"""
if bstrOpt is not None:
self._unsupported("approach(bstrOpt)")
return self.move(
vntPoseBase,
motion_mode=lComp,
coord_type="P",
offset=vntPoseLen,
)
[docs]
def get_motion_preparation(self):
"""モーション準備状態を取得する。MG400では未対応。"""
self._unsupported("get_motion_preparation")
[docs]
def rotate(
self,
plane,
angle,
center,
*,
path_blend=None,
speed=None,
is_async=True,
):
del (
plane,
angle,
center,
path_blend,
speed,
is_async,
)
self._unsupported("rotate")
# ------------------------------------------------------------------
# script / IO / variables
# ------------------------------------------------------------------
[docs]
def stop_script(self):
"""DOBOT固有: 実行中のスクリプトを停止する。
ベンダーDashboard APIの StopScript() をそのまま使用する。
共通Ctrl I/Fには公開しない。
"""
self._logger.info("[DobotTcp] Calling StopScript()")
reply = self._dashboard.StopScript()
return self._handle_command_reply(
"StopScript",
reply,
)
[docs]
def run_project(self, project_name):
"""DOBOT 固有: 指定プロジェクトを RunScript で実行する。"""
if project_name is None:
self._logger.error(
"[DobotTcp] run_project: project_name is required"
)
return False
name = str(project_name).strip()
if not name:
self._logger.error(
"[DobotTcp] run_project: project_name is empty"
)
return False
reply = self._dashboard.RunScript(name)
return self._handle_command_reply(
"RunScript",
reply,
)
[docs]
def run_script(self):
"""共通 I/F のスクリプト実行。
公開I/Fの ``run_script()`` は引数なしだが、MG400の
``RunScript(projectName)`` はプロジェクト名を必要とするため、
このメソッドでは別の意味を割り当てず未対応とする。
"""
return self._unsupported("run_script")
[docs]
def get_IO(self, id_num: int):
"""共通 ``IO*`` の出力readbackを返す。
``set_IO()`` がDOを書き込むため、同じI/Fの ``get_IO()`` もDO状態を
読む。入力は ``get_digital_input()`` で明示的に取得する。
"""
index = int(id_num)
if index < 1:
raise ValueError("id_num must be >= 1")
output_bits = int(self._scalar_status(self._state("digital_outputs")))
return 1 if output_bits & (1 << (index - 1)) else 0
[docs]
def set_IO(self, id_num: int, value: float) -> bool:
return self._ok(self._dashboard.DO(int(id_num), int(bool(value))))
@property
def lock(self):
"""ロック状態を取得する。MG400では対応状態を確認できないため未対応。"""
self._unsupported("lock")
@property
def task(self):
"""実行タスクを取得する。MG400では対応するタスクハンドルがないため未対応。"""
self._unsupported("task")
@property
def task_state(self):
"""タスク状態を取得する。MG400では対応する状態ハンドルがないため未対応。"""
self._unsupported("task_state")
[docs]
def get_task_status(self):
"""現在のタスク状態値を取得する。MG400では未対応。"""
self._unsupported("get_task_status")
@property
def direct_mode(self):
return self.get_direct_mode()
[docs]
def get_direct_mode(self):
"""ダイレクトモード状態を取得する。MG400では未対応。"""
self._unsupported("get_direct_mode")
[docs]
def get_version(self, rel_toml_path=None):
"""コントローラのバージョン情報を取得する。MG400では未対応。"""
del rel_toml_path
self._unsupported("get_version")
# Hand/log/autocal: I/Fだけを揃え、MG400側にない意味を捏造しない。
[docs]
def init_hand(self):
"""ハンドを初期化する。MG400標準APIでは未対応。"""
self._unsupported("init_hand")
[docs]
def open_hand(self, speed=None):
"""ハンドを開く。MG400標準APIでは未対応。"""
self._unsupported("open_hand")
[docs]
def hand_open(self, speed=None):
"""ハンドを開放する。MG400標準APIでは未対応。"""
self._unsupported("hand_open")
[docs]
def hand(
self,
pos=None,
*,
position=None,
force=None,
direction=True,
relative=False,
speed=None,
is_async=False,
):
del (
pos,
position,
force,
direction,
relative,
speed,
is_async,
)
self._unsupported("hand")
[docs]
def autocal(self, origin: bool = False) -> bool:
"""自動キャリブレーションを実行する。MG400で対応処理を確認できないため未対応。"""
self._unsupported("autocal")
[docs]
def start_log(self):
self._unsupported("start_log")
[docs]
def stop_log(self):
self._unsupported("stop_log")
[docs]
def clear_log(self):
self._unsupported("clear_log")
[docs]
def start_servo_log(self):
self._unsupported("start_servo_log")
[docs]
def stop_servo_log(self):
self._unsupported("stop_servo_log")
[docs]
def clear_servo_log(self):
self._unsupported("clear_servo_log")
[docs]
def get_log_count(self):
self._unsupported("get_log_count")
[docs]
def get_log_record(self, index):
self._unsupported("get_log_record")