Python library and class API

English 한국어

The library offers two layers.

ClientInterfaceAPI

The high-level class for scripts, services, demos, and internal business logic.

URClient

The lower-level class for teams that want direct access to subscriptions, message history, and more detailed lifecycle control.

High-level class: ClientInterfaceAPI

The simplest way to embed the project is to pass the robot IP address directly to ClientInterfaceAPI.

from ur_client import ClientInterfaceAPI

with ClientInterfaceAPI(
    host="192.168.0.10",
    interface_name="primary_ro",
    error_db_path="ur_error_codes.sqlite3",
) as api:
    api.wait_until_ready(timeout=5.0, minimum_packet_count=1)

    values = api.get_values(
        "robot.mode",
        "robot.safety_mode",
        "tcp.pose.x",
        "tcp.pose.y",
        "tcp.pose.z",
        "errors.latest.error_code",
        "errors.latest.description",
    )
    print(values)

Reading single values or groups of values

Use read() for one path and read_many() or get_values() for several paths at once.

robot_mode = api.read("robot.mode")
z_value = api.read("tcp.pose.z")

selected = api.read_many(
    "robot.mode",
    "robot.safety_mode",
    "tcp.pose.x",
    "tcp.pose.y",
    "tcp.pose.z",
)

Watching values over time

When you want a simple iterator that produces refreshed values, use watch().

for item in api.watch(
    "robot.mode",
    "errors.latest.error_code",
    interval=0.5,
):
    print(item)

This is often enough for service tools, dashboards, or periodic data forwarding.

Snapshot-style access

If you want a single structured snapshot, use snapshot().

snapshot = api.snapshot()
print(snapshot["robot"])
print(snapshot["errors"])

Error lookup from live data

The API can resolve the latest detected error directly.

latest_code = api.latest_error_code()
details = api.latest_error_details()
print(latest_code)
print(details)

You can also look up a code explicitly.

info = api.lookup_error("C204A7")
print(info["error_code"])
print(info["description"])
print(info["explanation"])
print(info["suggestion"])

Lower-level class: URClient

Use URClient when you want more direct lifecycle control or custom subscription logic.

from ur_client import AppSettings, URClient

settings = AppSettings(
    host="192.168.0.10",
    interface_name="primary",
    error_db_path="ur_error_codes.sqlite3",
)

client = URClient(settings)
client.start()

try:
    client.wait_until_connected(5.0)
    client.wait_for_packets(minimum_packet_count=1, timeout=5.0)
    print(client.read("robot.mode"))
    print(client.messages(limit=5))
finally:
    client.close()

Convenience properties

Both the high-level and lower-level layers expose convenience properties for common needs.

• robot_mode

• safety_mode

• tcp_pose

• joint_positions

• global_variables

• latest_key_message

• latest_error

Public API reference

class ur_client.ClientInterfaceAPI(host: str, interface_name: str = 'primary_ro', *, port: int | None = None, auto_start: bool = True, wait_ready: bool = False, wait_timeout: float = 5.0, minimum_packet_count: int = 1, error_db_path: str = 'ur_error_codes.sqlite3', **settings_overrides: Any)

Bases: object

Simple class-based API for scripts and applications.

Pass the robot IP address directly to the constructor, then call read(), read_many(), or use the convenience properties.

classmethod connect(host: str, interface_name: str = 'primary_ro', *, port: int | None = None, wait_ready: bool = True, wait_timeout: float = 5.0, minimum_packet_count: int = 1, **settings_overrides: Any) → ClientInterfaceAPI

start() → ClientInterfaceAPI

stop() → None

close() → None

property running: bool

property connected: bool

property packet_count: int

property status: str

property robot_mode: Any

property safety_mode: Any

property tcp_pose: dict[str, Any]

property joint_positions: dict[str, Any]

property global_variables: dict[str, Any]

property latest_key_message: KeyMessage | None

property latest_error: dict[str, Any] | None

wait_until_ready(timeout: float = 5.0, minimum_packet_count: int = 1) → bool

wait_until_connected(timeout: float = 5.0) → bool

wait_for_packet(timeout: float = 5.0) → bool

wait_for_packets(minimum_packet_count: int = 1, timeout: float = 5.0) → bool

wait_for_update(last_packet_count: int | None = None, timeout: float = 5.0) → int | None

read(path: str, default: Any = None) → Any

get(path: str, default: Any = None) → Any

read_many(*paths: str) → dict[str, Any]

get_values(*paths: str) → dict[str, Any]

keys() → list[str]

values() → dict[str, Any]

flat_values() → dict[str, Any]

meta() → dict[str, Any]

snapshot() → dict[str, Any]

get_snapshot() → dict[str, Any]

messages(category: str | None = None, limit: int | None = None) → list[MessageRecord]

latest_message(category: str | None = None) → MessageRecord | None

latest_error_code() → str | None

lookup_error(code: str | None = None) → dict[str, Any] | None

latest_error_details() → dict[str, Any] | None

latest_message_error_details() → dict[str, Any] | None

latest_raw_error_details() → dict[str, Any] | None

get_variable(name: str, default: Any = None) → Any

subscribe(callback: Any) → None

unsubscribe(callback: Any) → None

watch(*paths: str, interval: float = 0.5) → Iterator[dict[str, Any]]

watch_values(*paths: str, interval: float = 0.5, iterations: int | None = None) → Iterator[dict[str, Any]]

get_robot_state() → dict[str, Any]

class ur_client.AppSettings(host: 'str' = '192.168.163.128', interface_name: 'str' = 'primary_ro', port: 'int | None' = None, connect_timeout: 'float' = 5.0, read_timeout: 'float' = 5.0, reconnect_delay: 'float' = 2.0, max_packet_size: 'int' = 16777216, state_write_interval: 'float' = 0.5, save_sqlite: 'bool' = True, sqlite_path: 'str' = 'runtime/ur_monitor.sqlite3', save_ndjson: 'bool' = True, ndjson_path: 'str' = 'runtime/events.ndjson', save_snapshot_json: 'bool' = True, snapshot_json_path: 'str' = 'runtime/latest_snapshot.json', diagnostics_enabled: 'bool' = True, diagnostics_path: 'str' = 'runtime/diagnostics.ndjson', diagnostics_export_dir: 'str' = 'runtime/debug_exports', error_db_enabled: 'bool' = True, error_db_path: 'str' = 'ur_error_codes.sqlite3', recent_packet_limit: 'int' = 400, auto_start: 'bool' = True, window_refresh_ms: 'int' = 500, message_history_limit: 'int' = 200, favorite_keys: 'list[str]' = <factory>)

Bases: object

host: str = '192.168.163.128'

interface_name: str = 'primary_ro'

port: int | None = None

connect_timeout: float = 5.0

read_timeout: float = 5.0

reconnect_delay: float = 2.0

max_packet_size: int = 16777216

state_write_interval: float = 0.5

save_sqlite: bool = True

sqlite_path: str = 'runtime/ur_monitor.sqlite3'

save_ndjson: bool = True

ndjson_path: str = 'runtime/events.ndjson'

save_snapshot_json: bool = True

snapshot_json_path: str = 'runtime/latest_snapshot.json'

diagnostics_enabled: bool = True

diagnostics_path: str = 'runtime/diagnostics.ndjson'

diagnostics_export_dir: str = 'runtime/debug_exports'

error_db_enabled: bool = True

error_db_path: str = 'ur_error_codes.sqlite3'

recent_packet_limit: int = 400

auto_start: bool = True

window_refresh_ms: int = 500

message_history_limit: int = 200

favorite_keys: list[str]

class ur_client.URClient(settings: AppSettings | None = None, **settings_overrides: Any)

Bases: object

property running: bool

property connected: bool

property packet_count: int

property status: str

read(path: str, default: Any = None) → Any

read_many(*paths: str) → dict[str, Any]

select(*paths: str) → dict[str, Any]

wait_until_connected(timeout: float = 5.0) → bool

wait_for_packet(timeout: float = 5.0) → bool

wait_for_packets(minimum_packet_count: int = 1, timeout: float = 5.0) → bool

subscribe(callback: Any) → None

unsubscribe(callback: Any) → None

wait_for_update(last_packet_count: int | None = None, timeout: float = 5.0) → int | None

property robot_mode: Any

property safety_mode: Any

property tcp_pose: dict[str, Any]

property joint_positions: dict[str, Any]

property global_variables: dict[str, Any]

get_variable(name: str, default: Any = None) → Any

watch(*paths: str, interval: float = 0.5) → Iterator[dict[str, Any]]

messages(category: str | None = None, limit: int | None = None) → list[MessageRecord]

latest_message(category: str | None = None) → MessageRecord | None

property latest_key_message: KeyMessage | None

latest_key_message_dict() → dict[str, Any] | None

property latest_code_message: MessageRecord | None

latest_code_message_dict() → dict[str, Any] | None

property latest_nonzero_code_message: MessageRecord | None

latest_nonzero_code_message_dict() → dict[str, Any] | None

property latest_detected_error_message: MessageRecord | None

latest_detected_error_message_dict() → dict[str, Any] | None

property latest_popup_message: MessageRecord | None

latest_popup_message_dict() → dict[str, Any] | None

lookup_error(code: str | None) → dict[str, Any] | None

search_errors(query: str, limit: int = 20) → list[dict[str, Any]]

latest_error_details() → dict[str, Any] | None

latest_message_error_details() → dict[str, Any] | None

latest_raw_error_details() → dict[str, Any] | None

messages_with_code(nonzero_only: bool = False, limit: int | None = None) → list[MessageRecord]

keys() → list[str]

flat_values() → dict[str, Any]

values() → dict[str, Any]

meta() → dict[str, Any]

get(path: str, default: Any = None) → Any

snapshot() → dict[str, Any]

start() → None

stop() → None

close() → None

reset_storage() → None

export_debug_bundle(note: str = '') → str