Added trajectory planning to cli and api. Updated readme.

README.md

@@ -4,6 +4,31 @@ RSIPI is a high-performance, Python-based communication and control system desig
 
 ---
 
+🛡️ Safety Notice
+RSIPI is a powerful tool that directly interfaces with industrial robotic systems. Improper use can lead to dangerous movements, property damage, or personal injury.
+
+⚠️ Safety Guidelines
+Test in Simulation First
+Always verify your RSI communication and trajectories using simulation tools before deploying to a live robot.
+
+Enable Emergency Stops
+Ensure all safety hardware (E-Stop, fencing, light curtains) is active and functioning correctly.
+
+Supervised Operation Only
+Run RSIPI only in supervised environments with trained personnel present.
+
+Limit Movement Ranges
+Use KUKA Workspaces or software limits to constrain movement, especially when testing new code.
+
+Use Logging for Debugging
+Avoid debugging while RSI is active; instead, enable CSV logging and review logs post-run.
+
+Secure Network Configuration
+Ensure your RSI network is on a closed, isolated interface to avoid external interference or spoofing.
+
+Never Rely on RSIPI for Safety
+RSIPI is not a safety-rated system. Do not use it in applications where failure could result in harm.
+
 ## 📄 Description
 
 RSIPI allows users to:

src/RSIPI/config_parser.py

@@ -3,8 +3,11 @@ import xml.etree.ElementTree as ET
 class ConfigParser:
     """Handles parsing the RSI config file and generating structured send/receive variables."""
 
-    def __init__(self, config_file):
+    def __init__(self, config_file, rsi_limits_file=None):
         """Initialize with the configuration file and process send/receive structures."""
+        from src.RSIPI.rsi_limit_parser import parse_rsi_limits
+        self.rsi_limits_file = rsi_limits_file
+        self.safety_limits = {}
         self.config_file = config_file
         self.internal_structure = {
             "ComStatus": "String",
@@ -44,6 +47,14 @@ class ConfigParser:
         if "IPOC" not in self.send_variables:
             self.send_variables.update({"IPOC": 0})
 
+        # Load safety limits from .rsi.xml if provided
+        if self.rsi_limits_file:
+            try:
+                self.safety_limits = parse_rsi_limits(self.rsi_limits_file)
+            except Exception as e:
+                print(f"[WARNING] Failed to load .rsi.xml safety limits: {e}")
+                self.safety_limits = {}
+
         print(f"✅ Final Send Variables: {self.send_variables}")
         print(f"✅ Final Receive Variables: {self.receive_variables}")
 

src/RSIPI/kuka_visualizer.py

@@ -5,8 +5,9 @@ import os
 
 
 class KukaRSIVisualizer:
-    def __init__(self, csv_file):
+    def __init__(self, csv_file, safety_limits=None):
         self.csv_file = csv_file
+        self.safety_limits = safety_limits or {}
         if not os.path.exists(csv_file):
             raise FileNotFoundError(f"CSV file {csv_file} not found.")
         self.df = pd.read_csv(csv_file)
@@ -32,7 +33,7 @@ class KukaRSIVisualizer:
         plt.show()
 
     def plot_joint_positions(self, save_path=None):
-        """Plots joint positions over time."""
+        """Plots joint positions over time with safety limit overlays."""
         plt.figure()
         time_series = range(len(self.df))
 
@@ -40,6 +41,11 @@ class KukaRSIVisualizer:
             if col in self.df.columns:
                 plt.plot(time_series, self.df[col], label=col)
 
+                # 🔴 Overlay safety band
+                if col in self.safety_limits:
+                    low, high = self.safety_limits[col]
+                    plt.axhspan(low, high, color='red', alpha=0.1, label=f"{col} Safe Zone")
+
         plt.xlabel("Time Steps")
         plt.ylabel("Joint Position (Degrees)")
         plt.title("Joint Positions Over Time")
@@ -59,6 +65,10 @@ class KukaRSIVisualizer:
             if col in self.df.columns:
                 plt.plot(time_series, self.df[col], label=col)
 
+                if col in self.safety_limits:
+                    low, high = self.safety_limits[col]
+                    plt.axhspan(low, high, color='red', alpha=0.1, label=f"{col} Safe Zone")
+
         plt.xlabel("Time Steps")
         plt.ylabel("Force Correction (N)")
         plt.title("Force Trends Over Time")
@@ -81,9 +91,16 @@ if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Visualize RSI data logs.")
     parser.add_argument("csv_file", type=str, help="Path to the RSI CSV log file.")
     parser.add_argument("--export", action="store_true", help="Export graphs as PNG/PDF.")
+    parser.add_argument("--limits", type=str, help="Optional .rsi.xml file to overlay safety bands")
 
     args = parser.parse_args()
-    visualizer = KukaRSIVisualizer(args.csv_file)
+
+    if args.limits:
+        from src.RSIPI.rsi_limit_parser import parse_rsi_limits
+        limits = parse_rsi_limits(args.limits)
+        visualizer = KukaRSIVisualizer(args.csv_file, safety_limits=limits)
+    else:
+        visualizer = KukaRSIVisualizer(args.csv_file)
 
     visualizer.plot_trajectory()
     visualizer.plot_joint_positions()

src/RSIPI/network_handler.py

@@ -5,7 +5,7 @@ import csv
 import logging
 import xml.etree.ElementTree as ET  # ✅ FIX: Import ElementTree
 from .xml_handler import XMLGenerator
-
+from .safety_manager import SafetyManager
 
 class NetworkProcess(multiprocessing.Process):
     """Handles UDP communication and optional CSV logging in a separate process."""
@@ -17,6 +17,7 @@ class NetworkProcess(multiprocessing.Process):
         self.stop_event = stop_event
         self.config_parser = config_parser
         self.udp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
+        self.safety_manager = SafetyManager(config_parser.safety_limits)
 
         self.client_address = (ip, port)
 
@@ -98,7 +99,7 @@ class NetworkProcess(multiprocessing.Process):
             print(f"[ERROR] Error parsing received message: {e}")
 
     def log_to_csv(self):
-        """Write send/receive variables to the CSV log."""
+        """Write send/receive variables to the CSV log with safety flags."""
         filename = self.log_filename.value.decode().strip()
         if not filename:
             return
@@ -112,14 +113,36 @@ class NetworkProcess(multiprocessing.Process):
                     headers = ["Timestamp", "IPOC"]
                     headers += [f"Send.{k}" for k in self.send_variables.keys()]
                     headers += [f"Receive.{k}" for k in self.receive_variables.keys()]
+                    headers += ["SafetyViolation"]
                     writer.writerow(headers)
 
-                # Write current data
+                # Gather values
                 timestamp = time.strftime("%d-%m-%Y %H:%M:%S")
                 ipoc = self.receive_variables.get("IPOC", 0)
                 send_data = [self.send_variables.get(k, "") for k in self.send_variables.keys()]
                 receive_data = [self.receive_variables.get(k, "") for k in self.receive_variables.keys()]
-                writer.writerow([timestamp, ipoc] + send_data + receive_data)
+
+                # 🔴 Check for safety violations
+                violation = False
+                for var in self.send_variables:
+                    value = self.send_variables[var]
+                    # Check structured variables
+                    if isinstance(value, dict):
+                        for subkey, subval in value.items():
+                            path = f"{var}.{subkey}"
+                            try:
+                                self.safety_manager.validate(path, subval)
+                            except Exception as e:
+                                violation = str(e)
+                                break
+                    else:
+                        try:
+                            self.safety_manager.validate(var, value)
+                        except Exception as e:
+                            violation = str(e)
+                            break
+
+                writer.writerow([timestamp, ipoc] + send_data + receive_data + [violation or "False"])
 
         except Exception as e:
             print(f"[ERROR] Failed to log data to CSV: {e}")

src/RSIPI/rsi_api.py

@@ -13,6 +13,16 @@ from threading import Thread
 from .trajectory_planner import generate_trajectory, execute_trajectory
 
 
+def compare_test_runs(file1, file2):
+    """Compare two movement logs."""
+    df1 = pd.read_csv(file1)
+    df2 = pd.read_csv(file2)
+
+    diff = abs(df1 - df2)
+    max_deviation = diff.max()
+    return f"📊 Max Deviation: {max_deviation}"
+
+
 class RSIAPI:
     """RSI API for programmatic control, including alerts, logging, graphing, and data retrieval."""
 
@@ -182,15 +192,6 @@ class RSIAPI:
         df.to_csv(filename, index=False)
         return f"✅ Data exported to {filename}"
 
-    def compare_test_runs(self, file1, file2):
-        """Compare two movement logs."""
-        df1 = pd.read_csv(file1)
-        df2 = pd.read_csv(file2)
-
-        diff = abs(df1 - df2)
-        max_deviation = diff.max()
-        return f"📊 Max Deviation: {max_deviation}"
-
     def generate_report(self, filename, format_type):
         """Generate a statistical report from movement data."""
         data = self.client.get_movement_data()

src/RSIPI/rsi_cli.py

@@ -160,6 +160,30 @@ class RSICommandLineInterface:
             output = parts[2] if len(parts) == 3 else "report.txt"
             result = self.client.generate_report(parts[1], output)
             print(result)
+        elif cmd == "safety-stop":
+            self.client.safety_manager.emergency_stop()
+            print("🛑 Safety: Emergency Stop activated.")
+
+        elif cmd == "safety-reset":
+            self.client.safety_manager.reset_stop()
+            print("✅ Safety: Emergency Stop reset. Motion allowed.")
+
+        elif cmd == "safety-status":
+            sm = self.client.safety_manager
+            print("🧱 Safety Status: " + ("STOPPED" if sm.is_stopped() else "ACTIVE"))
+            print("📏 Enforced Limits:")
+            for var, (lo, hi) in sm.get_limits().items():
+                print(f"  {var}: {lo} → {hi}")
+
+        elif cmd == "safety-set-limit" and len(parts) == 4:
+            var, lo, hi = parts[1], parts[2], parts[3]
+            try:
+                lo = float(lo)
+                hi = float(hi)
+                self.client.safety_manager.set_limit(var, lo, hi)
+                print(f"✅ Set limit for {var}: {lo} to {hi}")
+            except ValueError:
+                print("❌ Invalid numbers for limit. Usage: safety-set-limit RKorr.X -5 5")
 
         else:
             print("❌ Unknown command. Type 'help' for a list of commands.")

src/RSIPI/rsi_client.py

@@ -3,16 +3,17 @@ import multiprocessing
 import time
 from .config_parser import ConfigParser
 from .network_handler import NetworkProcess
+from .safety_manager import SafetyManager
 
 class RSIClient:
     """Main RSI API class that integrates network, config handling, and message processing."""
 
-    def __init__(self, config_file):
+    def __init__(self, config_file,  rsi_limits_file=None):
         """Initialize the RSI client and set up logging and networking."""
         logging.info(f"📂 Loading RSI configuration from {config_file}...")
 
         # Load configuration
-        self.config_parser = ConfigParser(config_file)
+        self.config_parser = ConfigParser(config_file, rsi_limits_file)
         network_settings = self.config_parser.get_network_settings()
 
         self.manager = multiprocessing.Manager()
@@ -20,6 +21,9 @@ class RSIClient:
         self.receive_variables = self.manager.dict(self.config_parser.receive_variables)
         self.stop_event = multiprocessing.Event()
 
+        # ✅ Initialise safety manager from limits
+        self.safety_manager = SafetyManager(self.config_parser.safety_limits)
+
         logging.info(f"🚀 Starting network process on {network_settings['ip']}:{network_settings['port']}...")
 
         # ✅ Corrected constructor call with all necessary parameters

src/RSIPI/rsi_config.py

@@ -1,5 +1,6 @@
 import xml.etree.ElementTree as ET
 import logging
+from src.RSIPI.rsi_limit_parser import parse_rsi_limits
 
 # ✅ Configure Logging (Allow it to be turned off)
 LOGGING_ENABLED = True  # Change to False to disable logging
@@ -35,12 +36,26 @@ class RSIConfig:
         "Tech.T2": ["T21", "T22", "T23", "T24", "T25", "T26", "T27", "T28", "T29", "T210"],
     }
 
-    def __init__(self, config_file):
+    def __init__(self, config_file, rsi_limits_file=None):
         self.config_file = config_file
+        self.rsi_limits_file = rsi_limits_file
+        self.safety_limits = {}
+
         self.network_settings = {}
         self.send_variables = {}
         self.receive_variables = {}
+
         self.load_config()
+        self.load_safety_limits()  # ✅ NEW STEP
+
+    def load_safety_limits(self):
+        if self.rsi_limits_file:
+            try:
+                self.safety_limits = parse_rsi_limits(self.rsi_limits_file)
+                logging.info(f"✅ Loaded safety limits from {self.rsi_limits_file}")
+            except Exception as e:
+                logging.warning(f"⚠️ Failed to load RSI safety limits: {e}")
+                self.safety_limits = {}
 
     def strip_def_prefix(self, tag):
         """Remove 'DEF_' prefix from a variable name."""

src/RSIPI/rsi_limit_parser.py

@@ -0,0 +1,65 @@
+import xml.etree.ElementTree as ET
+
+def parse_rsi_limits(xml_path):
+    tree = ET.parse(xml_path)
+    root = tree.getroot()
+
+    raw_limits = {}
+
+    for rsi_object in root.findall("RSIObject"):
+        obj_type = rsi_object.attrib.get("ObjType", "")
+        params = rsi_object.find("Parameters")
+
+        if params is None:
+            continue
+
+        if obj_type == "POSCORR":
+            for param in params.findall("Parameter"):
+                name = param.attrib["Name"]
+                value = float(param.attrib["ParamValue"])
+                if name == "LowerLimX":
+                    raw_limits["RKorr.X_min"] = value
+                elif name == "UpperLimX":
+                    raw_limits["RKorr.X_max"] = value
+                elif name == "LowerLimY":
+                    raw_limits["RKorr.Y_min"] = value
+                elif name == "UpperLimY":
+                    raw_limits["RKorr.Y_max"] = value
+                elif name == "LowerLimZ":
+                    raw_limits["RKorr.Z_min"] = value
+                elif name == "UpperLimZ":
+                    raw_limits["RKorr.Z_max"] = value
+                elif name == "MaxRotAngle":
+                    for axis in ["A", "B", "C"]:
+                        raw_limits[f"RKorr.{axis}_min"] = -value
+                        raw_limits[f"RKorr.{axis}_max"] = value
+
+        elif obj_type == "AXISCORR":
+            for param in params.findall("Parameter"):
+                name = param.attrib["Name"]
+                value = float(param.attrib["ParamValue"])
+                if name.startswith("LowerLimA") or name.startswith("UpperLimA"):
+                    axis = name[-1]
+                    key = f"AKorr.A{axis}_{'min' if 'Lower' in name else 'max'}"
+                    raw_limits[key] = value
+
+        elif obj_type == "AXISCORREXT":
+            for param in params.findall("Parameter"):
+                name = param.attrib["Name"]
+                value = float(param.attrib["ParamValue"])
+                if name.startswith("LowerLimE") or name.startswith("UpperLimE"):
+                    axis = name[-1]
+                    key = f"AKorr.E{axis}_{'min' if 'Lower' in name else 'max'}"
+                    raw_limits[key] = value
+
+    # Build structured dictionary with only valid (min, max) pairs
+    structured_limits = {}
+    for key in list(raw_limits.keys()):
+        if key.endswith("_min"):
+            base = key[:-4]
+            min_val = raw_limits.get(f"{base}_min")
+            max_val = raw_limits.get(f"{base}_max")
+            if min_val is not None and max_val is not None:
+                structured_limits[base] = (min_val, max_val)
+
+    return structured_limits

src/RSIPI/safety_manager.py

@@ -0,0 +1,66 @@
+import logging
+
+
+class SafetyManager:
+    def __init__(self, limits=None):
+        """
+        limits: Optional dictionary of variable limits in the form:
+            {
+                'RKorr.X': (-5.0, 5.0),
+                'AKorr.A1': (-6.0, 6.0),
+                ...
+            }
+        """
+        self.limits = limits if limits is not None else {}
+        self.e_stop = False
+        self.last_values = {}
+
+    def validate(self, path: str, value: float) -> float:
+        """
+        Validate a single variable update against safety rules.
+        Raises RuntimeError if E-Stop is active.
+        Raises ValueError if out of allowed bounds.
+        """
+        if self.e_stop:
+            raise RuntimeError(f"SafetyManager: E-STOP active, motion blocked for {path}.")
+
+        if path in self.limits:
+            min_val, max_val = self.limits[path]
+            if not (min_val <= value <= max_val):
+                raise ValueError(
+                    f"SafetyManager: {path}={value} is out of bounds ({min_val} to {max_val})"
+                )
+
+        return value
+
+    def emergency_stop(self):
+        """Activate emergency stop mode (blocks all motion)."""
+        self.e_stop = True
+
+    def reset_stop(self):
+        """Reset emergency stop (motion allowed again)."""
+        self.e_stop = False
+
+    def set_limit(self, path: str, min_val: float, max_val: float):
+        """Manually set or override a safety limit at runtime."""
+        self.limits[path] = (min_val, max_val)
+
+    def get_limits(self):
+        """Return a copy of current limits."""
+        return self.limits.copy()
+
+    def is_stopped(self):
+        return self.e_stop
+
+    def validate(self, path: str, value: float) -> float:
+        if self.e_stop:
+            logging.warning(f"Safety violation: {path} update blocked (E-STOP active)")
+            raise RuntimeError(f"E-STOP active. Motion blocked for {path}.")
+
+        if path in self.limits:
+            min_val, max_val = self.limits[path]
+            if not (min_val <= value <= max_val):
+                logging.warning(f"Safety violation: {path}={value} blocked (out of bounds {min_val} to {max_val})")
+                raise ValueError(f"{path}={value} is out of bounds ({min_val} to {max_val})")
+
+        return value