import multiprocessing import pandas as pd import numpy as np import json import matplotlib.pyplot as plt from .rsi_client import RSIClient from .rsi_graphing import RSIGraphing from .kuka_visualizer import KukaRSIVisualizer from .krl_to_csv_parser import KRLParser class RSIAPI: """RSI API for programmatic control, including alerts, logging, graphing, and data retrieval.""" def __init__(self, config_file="RSI_EthernetConfig.xml"): """Initialize RSIAPI with an RSI client instance.""" self.client = RSIClient(config_file) self.graph_process = None # Store graphing process def start_rsi(self): """Start the RSI client.""" self.client.start() return "✅ RSI started." def stop_rsi(self): """Stop the RSI client.""" self.client.stop() return "✅ RSI stopped." def update_variable(self, variable, value): """Dynamically update an RSI variable.""" try: if isinstance(value, str) and value.replace('.', '', 1).isdigit(): value = float(value) if '.' in value else int(value) self.client.update_send_variable(variable, value) return f"✅ Updated {variable} to {value}" except Exception as e: return f"❌ Failed to update {variable}: {e}" def get_variables(self): """Retrieve current send and receive variables.""" return { "send_variables": dict(self.client.send_variables), "receive_variables": dict(self.client.receive_variables) } def get_live_data(self): """Retrieve real-time RSI data for external processing.""" return { "position": self.client.receive_variables.get("RIst", {"X": 0, "Y": 0, "Z": 0}), "velocity": self.client.receive_variables.get("Velocity", {"X": 0, "Y": 0, "Z": 0}), "acceleration": self.client.receive_variables.get("Acceleration", {"X": 0, "Y": 0, "Z": 0}), "force": self.client.receive_variables.get("MaCur", {"A1": 0, "A2": 0, "A3": 0, "A4": 0, "A5": 0, "A6": 0}), "ipoc": self.client.receive_variables.get("IPOC", "N/A") } def get_live_data_as_numpy(self): """Retrieve live RSI data as a NumPy array.""" data = self.get_live_data() return np.array([ list(data["position"].values()), list(data["velocity"].values()), list(data["acceleration"].values()), list(data["force"].values()) ]) def get_live_data_as_dataframe(self): """Retrieve live RSI data as a Pandas DataFrame.""" data = self.get_live_data() return pd.DataFrame([data]) def get_ipoc(self): """Retrieve the latest IPOC value.""" return self.client.receive_variables.get("IPOC", "N/A") def reconnect(self): """Restart the network connection without stopping RSI.""" self.client.reconnect() return "✅ Network connection restarted." def toggle_digital_io(self, io, value): """Toggle digital I/O states.""" self.client.update_send_variable(io, int(value)) return f"✅ {io} set to {value}" def move_external_axis(self, axis, value): """Move an external axis.""" self.client.update_send_variable(f"ELPos.{axis}", float(value)) return f"✅ Moved {axis} to {value}" def correct_position(self, correction_type, axis, value): """Apply correction to RKorr or AKorr.""" self.client.update_send_variable(f"{correction_type}.{axis}", float(value)) return f"✅ Applied correction: {correction_type}.{axis} = {value}" def adjust_speed(self, tech_param, value): """Adjust speed settings.""" self.client.update_send_variable(tech_param, float(value)) return f"✅ Set {tech_param} to {value}" def override_safety(self, limit): """Override safety limits.""" return f"⚠️ Overriding safety limit: {limit}" def reset_variables(self): """Reset send variables to default values.""" self.client.reset_send_variables() return "✅ Send variables reset to default values." def get_status(self): """Retrieve full RSI system status.""" return { "network": self.client.config_parser.get_network_settings(), "send_variables": dict(self.client.send_variables), "receive_variables": dict(self.client.receive_variables) } # ✅ CSV LOGGING METHODS def start_logging(self, filename): """Start logging RSI data to CSV.""" self.client.start_logging(filename) return f"✅ CSV Logging started: {filename}" def stop_logging(self): """Stop logging RSI data.""" self.client.stop_logging() return "🛑 CSV Logging stopped." def is_logging_active(self): """Return logging status.""" return self.client.is_logging_active() # ✅ GRAPHING METHODS def start_graphing(self, mode="position"): """Start real-time graphing.""" if self.graph_process and self.graph_process.is_alive(): return "⚠️ Graphing is already running." self.graph_process = multiprocessing.Process(target=RSIGraphing, args=(self.client, mode)) self.graph_process.start() return f"✅ Graphing started in {mode} mode." def stop_graphing(self): """Stop live graphing.""" if self.graph_process and self.graph_process.is_alive(): self.graph_process.terminate() self.graph_process.join() return "🛑 Graphing stopped." return "⚠️ No active graphing process." # ✅ ALERT METHODS def enable_alerts(self, enable): """Enable or disable real-time alerts.""" self.client.enable_alerts(enable) return f"✅ Alerts {'enabled' if enable else 'disabled'}." def set_alert_threshold(self, alert_type, value): """Set threshold for deviation or force alerts.""" if alert_type in ["deviation", "force"]: self.client.set_alert_threshold(alert_type, value) return f"✅ {alert_type.capitalize()} alert threshold set to {value}" return "❌ Invalid alert type. Use 'deviation' or 'force'." # ✅ DATA EXPORT & ANALYSIS def export_movement_data(self, filename): """Export movement data to a CSV file.""" data = self.client.get_movement_data() df = pd.DataFrame(data) 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() df = pd.DataFrame(data) report = { "Max Position Deviation": df.iloc[:, 1:].max().to_dict(), "Mean Position Deviation": df.iloc[:, 1:].mean().to_dict(), } if format_type == "csv": df.to_csv(f"{filename}.csv", index=False) elif format_type == "json": with open(f"{filename}.json", "w") as f: json.dump(report, f) elif format_type == "pdf": fig, ax = plt.subplots() df.plot(ax=ax) plt.savefig(f"{filename}.pdf") return f"✅ Report saved as {filename}.{format_type}" def visualize_csv_log(self, csv_file, export=False): """ Visualize CSV log file directly via RSIAPI. Args: csv_file (str): Path to CSV log file. export (bool): Whether to export the plots. """ visualizer = KukaRSIVisualizer(csv_file) visualizer.plot_trajectory() visualizer.plot_joint_positions() visualizer.plot_force_trends() if export: visualizer.export_graphs() def parse_krl_to_csv(self, src_file, dat_file, output_file): """ Parses KRL files (.src, .dat) and exports coordinates to CSV. Args: src_file (str): Path to KRL .src file. dat_file (str): Path to KRL .dat file. output_file (str): Path for output CSV file. """ try: parser = KRLParser(src_file, dat_file) parser.parse_src() parser.parse_dat() parser.export_csv(output_file) return f"✅ KRL data successfully exported to {output_file}" except Exception as e: return f"❌ Error parsing KRL files: {e}"