Heartbeat

A heartbeat is a periodic signal between two components that indicates normal operation. When a lack of heartbeat messages is identified, it will be assumed that the originator has failed, shutdown, or is generally no longer available.

The heartbeat between the robot control and the application makes sure that in case of a crash or unexpected shutdown of the application, the robot control transitions into an error state and no longer executes motion instructions.

Warning

Please note, that the heartbeat functionality is no replacement for properly configured and certified safety components of robot axes or controllers. In the context of safety certification, the heartbeat functionality is considered unsafe software.

Note

Heartbeat is currently only available for robots of the Victor Robot Behavior Group.

General Concepts

Robot controllers, that conform to the victor behavior group, in general provide OPC UA endpoints that allow an application or external component, that interacts with the robot control, to register itself as a heartbeat provider.

Once activated, the robot control expects a heartbeat from this component within a configurable time interval. If the time interval between two beats is violated, the robot control will assume the provider or the connection to the provider did malfunction and stop the robot.

By default, it is possible, that multiple providers are registered concurrently. They are monitored independently from each other and distinguished from each other by a session id parameter of the heartbeat call.

It is also possible to request uniqueness of the heartbeat connection. This will disable the option to have multiple providers active concurrently and enforce, that there is only one heartbeat provider active at a time.

Importance of Thread Error Detection in Python Applications

Apart from obvious connection errors to the robot control, using the heartbeat within a Python application is only useful, if errors and undesired behavior inside the application are detected and lead to a stop of the heartbeat.

Consider a simple example where an application consists of three threads. One thread communicates in a cyclic manner with a high-level logic control by sending process data to the control and receiving and processing external signals, for example the state of a light barrier monitoring a handling output buffer conveyor. The second thread contains the actual application robot sequence program and the third thread is the internal worker thread of voraus-robot-arm.

Under normal operation, the robot moves along the sequence with full speed, but it should only move with half speed if the light barrier detects an object, because the buffer conveyor begins to fall behind its normal processing capacity. It should furthermore pause in its sequence at any time if another light barrier detects an object as well - e.g. the buffer conveyor is full.

If everything behaves normally, the communication thread processes the received signal and triggers for example one of the Time Override methods to slow the robot down. However, consider the case if the communication thread has a division by zero error during its data processing.

Unfortunately, Python by default ignores the crash of threads silently. The communication thread will die as a result, which leads to a loss of connection to the logic control. However, the other two threads still work as expected. There is still a valid connection to the robot control and the robot will still move along its sequence with full speed, even if the light barrier signals, would suggest it to slow down or pause in the current position. Instead of switching to a safe error state, the robot will continue its sequence as if everything is fine. However, the overall behavior is now undefined.

In order to solve the problem of the example above, voraus-robot-arm will by default monitor, that the application and all its threads are in a healthy state, while a heartbeat is active, such that the robot will always stop, if:

The network connection to the robot control is no longer available
The application was shut down
The application crashed or partly crashed by losing a thread with an uncaught exception

Note

On the implementation level, this is achieved by overwriting the threading.excepthook. Do not overwrite the excepthook while the heartbeat is active! Otherwise, the heartbeat will not work. If necessary, overwrite the excepthook before or after the heartbeat session.

As a result, the division by zero error in the communication thread of the example above would be seen by the handler of the voraus-robot-arm library. The heartbeat would be killed immediately, because something is obviously not working as expected. After the heartbeat interval times out in the robot control, the robot would stop with an error, even if the actual communication thread to the robot control still is valid and sends commands.

Use the Heartbeat in an Application

The heartbeat session of an application has one of the following states and can be retrieved with the method get_heartbeat_state():

class HeartbeatState(value)

An enum representing the heartbeat state.

INACTIVE = 'INACTIVE': The heartbeat is not active.

ALIVE = 'ALIVE': The heartbeat is active and in a healthy state.

DEAD = 'DEAD': The heartbeat was active but was killed due to a detected error.

The easiest and safest way to activate the heartbeat functionality is a context manager heartbeat_v().

with robot.heartbeat_v():

If a context manager does not fit the needs, use start_heartbeat_v() and stop_heartbeat_v() instead.

robot.start_heartbeat_v()
robot.move_ptp(HOME).result()
robot.stop_heartbeat_v()

Once the heartbeat is ALIVE, the robot instance will send the heartbeat signals automatically in one of its worker threads.

If an error in the application was detected and the heartbeat session is DEAD new instructions can no longer be registered. In order to register new instructions again, the error has to be reset (see resetting errors) and the heartbeat session must be closed and reopened (if desired), e.g. via exiting the heartbeat context and entering a new heartbeat context.

Each heartbeat session has an id, which can be configured via the session_id argument. This id is used to identify the session of this application within the robot control. If no session id is provided, a random id will be used to register oneself at the robot control. An id can be reused after the old session was stopped or died.

The interval after which the robot control assumes the heartbeat is dead can also be configured. Use the interval_ms argument for that. By default an interval of 1000 ms is used.

Furthermore it might also be desired to request that the heartbeat session of your application is the only active heartbeat session of the robot control. To do so, use the unique_heartbeat_v() context manager or start_unique_heartbeat(). The stop method is the same for unique and not unique. If a new heartbeat is registered while a unique heartbeat is active, robot control will transition into an error. If a unique heartbeat is registered while any other heartbeat is active, robot control will transition into an error.

Advanced: Disabling the Automatic Thread Monitoring

Warning

The following section describes a highly advanced feature, which allows extensive customization. It should only be used if absolutely necessary. Configuring the proposed solutions incorrectly, may result in the heartbeat not working as desired.

It is possible to disable the automatic monitoring for thread exceptions as described in the section above. This can be done by starting the application with the environment variable HEARTBEAT_DISABLE_THREADING_EXCEPTHOOK_MONITORING=True.

Once the heartbeat is started, the threading.excepthook is left unchanged. In order to still kill the heartbeat from the custom implementation of a monitoring, the HeartbeatKiller class can be used. Calling the kill() method from any part of the code will kill the heartbeat if it is active.

If the automatic monitoring is not disabled, the HeartbeatKiller class will do nothing.

Full Example of Using a Heartbeat

Below you find an example where a thread dies due to an uncaught exception which causes the heartbeat signal to stop.

Heartbeat example

"""A simple example on how use the heartbeat functionality."""

import logging
import threading
import time
from math import radians
from typing import Protocol, runtime_checkable

from voraus_robot_arm import (
    HeartbeatState,
    HeartbeatVictorTrait,
    JointPose,
    LifecycleState,
    LifecycleTrait,
    MovePTPTrait,
    RobotArmError,
    VorausErrorHandler,
    VorausIndustrialRobotArm,
    configure_logging,
)

VORAUS_CORE_HOST = "localhost"
VORAUS_CORE_PORT = 48401
VORAUS_ERROR_HANDLER_PORT = 48404

HOME = JointPose().from_list(
    [radians(d) for d in [0, -90, 90, -90, -90, 0]]
)
VERTICAL = JointPose().from_list(
    [radians(d) for d in [0, -90, 0, -90, -90, 0]]
)

_logger = logging.getLogger(__name__)


@runtime_checkable
class _RequiredRobotTraits(
    HeartbeatVictorTrait, MovePTPTrait, LifecycleTrait, Protocol
): ...


def _wait_for_error_state(
    robot: _RequiredRobotTraits,
    timeout_s: float = 2,
) -> None:
    start_time = time.time()
    while time.time() - start_time < timeout_s:
        time.sleep(0.01)
        state = robot.get_lifecycle_state()
        if state == LifecycleState.ERROR:
            time.sleep(1)  # Allow the robot to slow down
            return

    msg = "Robot did not switch to error state!"
    raise RuntimeError(msg)


def start_faulty_thread() -> None:
    """Simulate a faulty thread with an uncaught exception."""

    def _raise() -> None:
        time.sleep(0.5)
        msg = "This is an example exception"
        raise RuntimeError(msg)

    t = threading.Thread(target=_raise, daemon=True, name="Example Thread")
    t.start()


def run_my_application(robot: _RequiredRobotTraits) -> None:
    """Example application loop."""
    robot.enable()
    while True:
        robot.move_ptp(HOME).result()
        robot.move_ptp(VERTICAL).result()


if __name__ == "__main__":
    configure_logging()

    robot = VorausIndustrialRobotArm()
    error_handler = VorausErrorHandler()
    with (
        robot.connect(host=VORAUS_CORE_HOST, port=VORAUS_CORE_PORT),
        error_handler.connect(
            host=VORAUS_CORE_HOST, port=VORAUS_ERROR_HANDLER_PORT
        ),
    ):
        assert robot.get_heartbeat_state_v() == HeartbeatState.INACTIVE
        with robot.heartbeat_v():
            current_state = robot.get_heartbeat_state_v()
            _logger.info("Current state: %s", current_state)
            assert current_state == HeartbeatState.ALIVE

            start_faulty_thread()
            try:
                run_my_application(robot)
            except RobotArmError:
                _logger.exception("Error caused by a faulty thread:")
            assert robot.get_heartbeat_state_v() == HeartbeatState.DEAD

            _wait_for_error_state(robot)
            error_handler.reset_error(robot)

            robot.enable()

            try:
                robot.move_ptp(HOME).result()
            except RobotArmError:
                _logger.exception(
                    "Cannot register new instruction if heartbeat is dead."
                )
        assert robot.get_heartbeat_state_v() == HeartbeatState.INACTIVE

        robot.start_heartbeat_v()
        robot.move_ptp(HOME).result()
        robot.stop_heartbeat_v()

If this example is run, the error below will pop up. At first, an error log message is shown that highlights that an uncaught exception was found and in which thread (Example Thread) followed by the traceback of that exception. Furthermore, the move_ptp instruction failed with RA-041 because the robot control transitioned into an error due to the lack of a heartbeat signal.

[ERROR]  Detected an uncaught exception in thread Example Thread
Traceback (most recent call last):
  File "/usr/lib/python3.11/threading.py", line 1045, in _bootstrap_inner
    self.run()
  File "/usr/lib/python3.11/threading.py", line 982, in run
    self._target(*self._args, **self._kwargs)
  File "/voraus-robot-arm-py/docs/docs_src/user_guide/traits/heartbeat_victor.py", line 64, in _raise
    raise RuntimeError(msg)
RuntimeError: This is an example exception
[ERROR]  Killing heartbeat due to unhandled thread exception.
[ERROR]  Error caused by a faulty thread:
Traceback (most recent call last):
  File "/voraus-robot-arm-py/docs/docs_src/user_guide/traits/heartbeat_victor.py", line 97, in <module>
    run_my_application(robot)
  File "/voraus-robot-arm-py/docs/docs_src/user_guide/traits/heartbeat_victor.py", line 74, in run_my_application
    robot.move_ptp(HOME).result()
  File "/voraus-robot-arm-py/src/voraus_robot_arm/_future.py", line 162, in result
    raise self._exception
voraus_robot_arm._errors.RobotArmError: RA-041 - Robot switched to error state.
Reason: 1516 - A heartbeat tracker timed out Heartbeat tracker 0 timed out.
Time since last triggered heartbeat: 1000037451ns.

Definition of the Heartbeat Methods

HeartbeatVictorTrait

protocol HeartbeatVictorTrait

Trait to initialize a heartbeat for robots with Victor behavior.

This protocol is runtime checkable.

Classes that implement this protocol must have the following methods / attributes:

abstractmethod start_heartbeat_v(session_id=None, interval_ms=1000)

Initiate a heartbeat session with the robot control.

Only one voraus-robot-arm heartbeat session can be active at once.

Parameters:

session_id (int | None) – The id of the heartbeat session to start. If None, a random id will be generated. Defaults to None.
interval_ms (int) – The amount of time in ms the robot control waits for a new heartbeat signal before transitioning into an error state. Defaults to 1000.

Return type:

None

abstractmethod start_unique_heartbeat_v(session_id=None, interval_ms=1000)

Initiate a unique heartbeat session with the robot control.

Unique means that this is the only active heartbeat session that exists globally for the robot control. Registering a new heartbeat while a unique heartbeat is active will be rejected and causes a robot control error. Calling this function while another heartbeat is active will cause a robot control error.

Parameters:

session_id (int | None) – The id of the heartbeat session to start. If None, a random id will be generated. Defaults to None.
interval_ms (int) – The amount of time in ms the robot control waits for a new heartbeat signal before transitioning into an error state. Defaults to 1000.

Return type:

None

abstractmethod stop_heartbeat_v()

Stop the currently active heartbeat session gracefully.

Requires an active heartbeat session else this raises an error.

Return type:: None

abstractmethod heartbeat_v(session_id=None, interval_ms=1000)

Context manager for a heartbeat session with the robot control.

Only one voraus-robot-arm heartbeat session can be active at once.

Parameters:

session_id (int | None) – The id of the heartbeat session to start. If None, a random id will be generated. Defaults to None.
interval_ms (int) – The amount of time in ms the robot control waits for a new heartbeat signal before transitioning into an error state. Defaults to 1000.

Return type:

Generator[None, None, None]

abstractmethod unique_heartbeat_v(session_id=None, interval_ms=1000)

Context manager for a unique heartbeat session with the robot control.

Parameters:

session_id (int | None) – The id of the heartbeat session to start. If None, a random id will be generated. Defaults to None.
interval_ms (int) – The amount of time in ms the robot control waits for a new heartbeat signal before transitioning into an error state. Defaults to 1000.

Return type:

Generator[None, None, None]

abstractmethod get_heartbeat_state_v()

Get the heartbeat state.

Return type:: HeartbeatState
Returns:: The current heartbeat state.

Limits

Tools