Monitoring¶

Position, IPOC, motor currents, applied corrections, override and STATUS reads, as dicts, NumPy or pandas.

Usage¶

# Comprehensive snapshot
data = api.monitoring.get_live_data()
# {"position": {X,Y,Z,A,B,C}, "velocity": {...}, "force": {...}, "ipoc": 123456}

# Individual reads
pos = api.monitoring.get_position()      # {X, Y, Z, A, B, C}
ipoc = api.monitoring.get_ipoc()         # Interrupt point counter

# Motor currents (MACur, an INTERNAL tag: Max and Full contexts). Units are
# not stated by KUKA - treat as relative. Raises RSIVariableError if the
# context has no MACur; get_force() is the older name and returns zeros then.
currents = api.monitoring.get_motor_currents()   # {A1..A6}

# Max context only: what the controller actually applied, and $OV_PRO
api.monitoring.get_applied_correction()          # POSCORRMON {X..C}, {} if not wired
api.monitoring.get_applied_joint_correction()    # AXISCORRMON {A1..A6}
api.monitoring.get_override()                    # $OV_PRO %, None if not wired
api.monitoring.set_override(50)                  # 1-100; ValueError outside that
api.monitoring.get_correction_limit_status()     # POSCORR Stat decoded, if your context
# gets it to the PC (it cannot come back over Ethernet - see docs/rsi-objects.md);
# None on the shipped contexts
api.monitoring.get_robot_status(1, "Mode_Op")    # STATUS object n decoded ("T1", "AUT"...)

# NumPy/Pandas formats
arr = api.monitoring.get_live_data_as_numpy()        # shape (4, 6)
df = api.monitoring.get_live_data_as_dataframe()     # single-row DataFrame

# Console watch (blocking, Ctrl+C to stop)
api.monitoring.watch_network(duration=10, rate=0.2)

Reference¶

MonitoringAPI ¶

MonitoringAPI(client: RSIClient)

Real-time monitoring interface for KUKA RSI robot data.

Provides access to live position, velocity, force, and IPOC data in various formats for external processing and analysis.

Initialize MonitoringAPI namespace.

Parameters:

Name	Type	Description	Default
`client`	`RSIClient`	RSIClient instance for accessing receive variables	required

get_live_data ¶

get_live_data() -> Dict[str, Any]

Retrieve comprehensive real-time RSI data.

Returns:

Type Description

Dictionary containing

- position: TCP position (RIst) {X, Y, Z, A, B, C}
- velocity: TCP velocity {X, Y, Z} in mm/s
- acceleration: TCP acceleration {X, Y, Z} in mm/s^2
- force: Joint motor currents (MACur) {A1-A6}
- ipoc: Current interrupt point counter

Velocity and acceleration are taken from Velocity/Acceleration SEND variables when the config declares them (e.g. POSACT wired through D objects in the RSI context); otherwise they are derived from successive position samples - see _derive_motion() for the accuracy caveats.

Example

data = api.monitoring.get_live_data() print(f"Position: {data['position']}") Position: {'X': 600.5, 'Y': -200.3, 'Z': 1450.8, 'A': 0.0, 'B': 0.0, 'C': 0.0} print(f"IPOC: {data['ipoc']}") IPOC: 123456

get_velocity ¶

get_velocity() -> Dict[str, float]

Current TCP velocity in mm/s.

Returns:

Type	Description
`Dict[str, float]`	{X, Y, Z} in mm/s - from the config's Velocity variable if it declares one, otherwise differentiated from position.

get_acceleration ¶

get_acceleration() -> Dict[str, float]

Current TCP acceleration in mm/s^2.

Returns:

Type	Description
`Dict[str, float]`	{X, Y, Z} in mm/s^2 - a second difference of position unless the config declares an Acceleration variable, so treat as indicative.

get_live_data_as_numpy ¶

get_live_data_as_numpy() -> np.ndarray

Retrieve live RSI data as a NumPy array.

Returns 2D array with rows: [position, velocity, acceleration, force] and columns padded to max length (6 for force axes).

Returns:

Type	Description
`ndarray`	NumPy array (4 x max_length) with robot state data

Example

arr = api.monitoring.get_live_data_as_numpy() print(arr.shape) (4, 6) print(arr[0]) # Position row [600.5 -200.3 1450.8 0.0 0.0 0.0]

get_live_data_as_dataframe ¶

get_live_data_as_dataframe() -> pd.DataFrame

Retrieve live RSI data as a Pandas DataFrame.

Returns:

Type	Description
`DataFrame`	DataFrame with single row containing current robot state

Example

df = api.monitoring.get_live_data_as_dataframe() print(df.columns) Index(['position', 'velocity', 'acceleration', 'force', 'ipoc']) print(df['ipoc'][0]) 123456

get_ipoc ¶

get_ipoc() -> Union[int, str]

Get current IPOC (Interrupt Point Counter) value.

The IPOC increments with each RSI cycle (typically every 4ms) and is used for synchronization between client and controller.

Returns:

Type	Description
`Union[int, str]`	Current IPOC value, or "N/A" if not available

Example

ipoc = api.monitoring.get_ipoc() print(ipoc) 123456

get_position ¶

get_position() -> Dict[str, float]

Get current TCP position in Cartesian coordinates.

Returns:

Type	Description
`Dict[str, float]`	Dictionary with X, Y, Z (mm) and A, B, C (degrees) orientation

Example

pos = api.monitoring.get_position() print(f"TCP at X={pos['X']}, Y={pos['Y']}, Z={pos['Z']}") TCP at X=600.5, Y=-200.3, Z=1450.8

get_force ¶

get_force() -> Dict[str, float]

Get current motor currents for all joints.

Motor current is a proxy for force/torque applied at each joint. Units depend on robot model and configuration.

Returns:

Type	Description
`Dict[str, float]`	Dictionary with A1-A6 motor current values

Example

force = api.monitoring.get_force() print(f"Joint A1 current: {force['A1']}") Joint A1 current: 12.5

get_motor_currents ¶

get_motor_currents() -> Dict[str, float]

Motor currents of the robot axes A1-A6, as the controller reports them through the DEF_MACur INTERNAL tag (no channel cost).

Unlike :meth:get_force (the older name for the same data) this raises when the context does not declare MACur instead of returning zeros: a zero that means "not wired" looks exactly like a real reading, and that kind of silent default has bitten this project before. Only the Max and Full contexts declare MACur; pass --internal DEF_MACur to RSIPI.config_builder to add it to a generated config.

The object reference does not state the units. Treat the values as relative until they have been compared against the pendant.

Returns:

Type	Description
`Dict[str, float]`	Dictionary with A1-A6 motor current values (a copy).

Raises:

Type	Description
`RSIVariableError`	if the config declares no `DEF_MACur`.

Example

currents = api.monitoring.get_motor_currents() print(f"A1: {currents['A1']:.2f}")

get_applied_correction ¶

get_applied_correction() -> Dict[str, float]

Cartesian correction the controller has actually applied (POSCORRMON).

This is the answer to "did the robot do what I asked?", which the commanded value cannot give you. A STOP object once silently stopped the controller applying any correction at all while RSI kept running perfectly - full packet rate, zero late packets, no error - and the only symptom was a pose that never changed. POSCORRMON would have shown that immediately.

Needs a context wiring POSCORRMON (max); returns an empty dict otherwise.

Returns:

Type	Description
`Dict[str, float]`	X, Y, Z (mm) and A, B, C (degrees) of applied correction

Example

api.motion.update_cartesian(X=5.0) api.monitoring.get_applied_correction()

get_applied_joint_correction ¶

get_applied_joint_correction() -> Dict[str, float]

Joint correction the controller has actually applied (AXISCORRMON).

The per-axis counterpart of :meth:get_applied_correction. Needs a context wiring AXISCORRMON (max); returns an empty dict otherwise.

Returns:

Type	Description
`Dict[str, float]`	A1-A6 applied correction in degrees

Example

api.monitoring.get_applied_joint_correction()['A6'] 0.02

get_correction_limit_status ¶

get_correction_limit_status(raw: bool = False)

Is the controller clamping the Cartesian correction, and where?

POSCORR limits the cumulative correction and, per the RSI reference, "if an input exceeds the valid range, the corresponding maximum value is used" - it clamps and reports nothing. Its Stat output is the only signal that this is happening, which is why a move can stop dead at exactly the limit with no error anywhere.

Needs a context wiring POSCORR's Stat output (RSIPI_Max); returns None otherwise.

Parameters:

Name	Type	Description	Default
`raw`	`bool`	return the integer instead of the decoded dict	`False`

Returns:

Type	Description
`dict \| int \| None`	`{"active": bool, "limited": bool, "at_limit": [str, ...], "raw": int}`, or the int if raw, or None if not wired.

Example

api.monitoring.get_correction_limit_status()

get_robot_status ¶

get_robot_status(index: int = 1, meaning: Optional[str] = None) -> Optional[Any]

Read a STATUS object's value, optionally decoded.

A STATUS object reports one controller status chosen by its Type parameter, so a context may wire several. They appear as Status1, Status2 and so on.

No shipped context includes a STATUS object - its Type is an enum whose numeric value must come from RSIVisual rather than be guessed - so this returns None unless you have added one.

Parameters:

Name	Type	Description	Default
`index`	`int`	which Status channel to read	`1`
`meaning`	`Optional[str]`	decode the number using one of STATUS_MEANINGS, e.g. "Sensor" or "Mode_Op". Omit for the raw integer.	`None`

Returns:

Type	Description
`Optional[Any]`	The raw int, the decoded string, or None if not wired.

Example

api.monitoring.get_robot_status(1, "Sensor") 'CYCLE' api.monitoring.get_robot_status(2, "Mode_Op") 'T1'

get_override ¶

get_override() -> Optional[int]

Program override ($OV_PRO) as a percentage, or None if not wired.

Needs a context with an OV_PRO object (max).

Example

api.monitoring.get_override() 100

set_override ¶

set_override(percent: int) -> str

Set the program override ($OV_PRO) from Python.

Slows or speeds the robot's programmed motion while RSI runs, which is a gentler lever than an E-stop when something looks wrong.

Parameters:

Name	Type	Description	Default
`percent`	`int`	1-100. Values outside that are rejected rather than clamped - a silently clamped speed request is the kind of thing you only notice on the robot.	required

Raises:

Type	Description
`RSIVariableError`	if the config declares no OvProW channel, i.e. the context has no MAP2OV_PRO object
`ValueError`	if percent is outside 1-100

Example

api.monitoring.set_override(30) 'Override set to 30%'

watch_network ¶

watch_network(duration: Optional[float] = None, rate: float = 0.2) -> None

Continuously print live position and IPOC data to console.

Useful for monitoring network communication health and robot movement during testing and debugging.

Parameters:

Name	Type	Description	Default
`duration`	`Optional[float]`	Watch duration in seconds (None = until Ctrl+C)	`None`
`rate`	`float`	Update rate in seconds (default: 0.2 = 5 Hz)	`0.2`

Example

Watch for 10 seconds at 5Hz¶

api.monitoring.watch_network(duration=10) [14:32:01] IPOC: 123456 | RIst: {'X': 600.5, 'Y': -200.3, 'Z': 1450.8} [14:32:01] IPOC: 123506 | RIst: {'X': 600.6, 'Y': -200.3, 'Z': 1450.8} ...

Watch indefinitely (Ctrl+C to stop)¶

api.monitoring.watch_network()