_helpers.py

""" Helper functions for RapidCode Python samples.
"""
 
from _imports import RapidCode, RAPIDCODE_DIR, constants, platform
 
 
def get_enum_name(prefix: str, value: int) -> str:
    """Get enum name using reflection on SWIG-generated constants.
 
    Args:
        prefix: The enum prefix (e.g., "RSINetworkState_RSINetworkState")
        value: The enum value to look up
 
    Returns:
        The enum name with prefix stripped, or "UNKNOWN(value)" if not found
 
    Example:
        get_enum_name("RSINetworkState_RSINetworkState", controller.NetworkStateGet())
        # Returns "OPERATIONAL" for RSINetworkState_RSINetworkStateOPERATIONAL
    """
    for name in dir(RapidCode):
        if name.startswith(prefix):
            if getattr(RapidCode, name) == value:
                return name[len(prefix):]
    return f"UNKNOWN({value})"
 
 
def get_creation_parameters():
  # create a motion controller and return it. 
  # If any errors are found, raise an exception with the error log as the message.
  creation_params: RapidCode.CreationParameters = RapidCode.CreationParameters()
  creation_params.RmpPath = RAPIDCODE_DIR
  creation_params.NicPrimary = constants.RMP_NIC_PRIMARY
 
  if platform.system() == "Windows":
    creation_params.NodeName = constants.RMP_NODE_NAME
  elif platform.system() == "Linux":
    creation_params.CpuAffinity = constants.RMP_CPU_AFFINITY
  else:
    raise Exception("Unsupported platform")
 
  return creation_params
 
def check_errors(rsi_object):
  # check for errors in the given rsi_object and print any errors that are found. 
  # If the error log contains any errors (not just warnings), raises an exception with the error log as the message.
  # returns a tuple containing a boolean indicating whether the error log contained any errors and the error log string.
  error_string_builder = ""
  i = rsi_object.ErrorLogCountGet()
  while rsi_object.ErrorLogCountGet() > 0:
      error:RapidCode.RsiError = rsi_object.ErrorLogGet()
      error_type = "WARNING" if error.isWarning else "ERROR"
      error_string_builder += f"{error_type}: {error.text}\n"
      if len(error_string_builder) > 0:
          print(error_string_builder)
      if "ERROR" in error_string_builder:
          raise Exception(error_string_builder)
  return "ERROR" in error_string_builder, error_string_builder
 
def start_the_network(controller):
  # attempts to start the network using the given MotionController object. 
  # If the network fails to start, it reads and prints any log messages that may be helpful 
  # in determining the cause of the problem, and then raises an RsiError exception.
  if controller.NetworkStateGet() != RapidCode.RSINetworkState_RSINetworkStateOPERATIONAL:  # Check if network is started already.
    print("Starting Network..")
    controller.NetworkStart()                                                               # If not. Initialize The Network. (This can also be done from RapidSetup Tool)
 
  if controller.NetworkStateGet() != RapidCode.RSINetworkState_RSINetworkStateOPERATIONAL:  # Check if network is started again.
    messages_to_read = controller.NetworkLogMessageCountGet()                               # Some kind of error starting the network, read the network log messages
 
    for i in range(messages_to_read):
      print(controller.NetworkLogMessageGet(i))                                             # Print all the messages to help figure out the problem
    print("Expected OPERATIONAL state but the network did not get there.")
    #  raise Exception(Expected OPERATIONAL state but the network did not get there.)       # Uncomment if you want your application to exit when the network isn't operational. (Comment when using phantom axis)
  else:                                                                                     # Else, of network is operational.
    print("Network Started")
 
def abort_motion_object(motion_object):
  # Aborts motion on the given motion object (Axis or MultiAxis), waits for motion to complete,
  # clears faults, and verifies the object enters IDLE state.
  # If the object fails to enter IDLE state, raises an exception with the error source.
  motion_object.EStopAbort()
  motion_object.MotionDoneWait()
  motion_object.ClearFaults()
 
  # check for idle state
  verify_idle_state(motion_object)
 
def verify_idle_state(motion_object):
  # Verifies that the given motion object (Axis or MultiAxis) is in IDLE state.
  # If not, raises an exception with the error source.
  if motion_object.StateGet() != RapidCode.RSIState_RSIStateIDLE:
    source = motion_object.SourceGet()  # get state source enum
    error_msg = f"Axis or MultiAxis {motion_object.NumberGet()} is expected to be in IDLE state, but is in state {enum_to_name(motion_object.StateGet(), 'RSIState')}. " \
                f"\nError Source: {motion_object.SourceNameGet(source)}"
    raise Exception(error_msg)
 
def enum_to_name(value, prefix):
    """Reverse lookup: int value -> enum name"""
    for name in dir(RapidCode):
        if name.startswith(prefix + "_" + prefix) and getattr(RapidCode, name) == value:
            return name.split(prefix + "_" + prefix)[1]
    return str(value)