""" Workflow Integration Helper for AutoEIS Hugging Face Space This module provides utilities for integrating with external workflow systems """ import base64 import json import requests from typing import Dict, Any, Optional from datetime import datetime, timedelta import jwt import hashlib class WorkflowClient: """Client for integrating AutoEIS HF Space with workflow systems""" def __init__(self, hf_space_url: str, secret_key: str = None): """ Initialize workflow client Args: hf_space_url: URL of your Hugging Face Space secret_key: Secret key for JWT generation (optional) """ self.hf_space_url = hf_space_url.rstrip('/') self.secret_key = secret_key or self._generate_secret() def _generate_secret(self) -> str: """Generate a secret key if none provided""" return hashlib.sha256( f"autoeis-{datetime.utcnow().isoformat()}".encode() ).hexdigest() def create_jwt_token(self, workflow_id: str, expires_in: int = 1800) -> str: """ Create a JWT token for authentication Args: workflow_id: ID of the workflow expires_in: Token expiration time in seconds (default 30 minutes) Returns: JWT token string """ payload = { 'workflow_id': workflow_id, 'exp': datetime.utcnow() + timedelta(seconds=expires_in), 'iat': datetime.utcnow(), 'iss': 'autoeis-workflow' } return jwt.encode(payload, self.secret_key, algorithm='HS256') def prepare_analysis_params( self, workflow_id: str, node_id: str, callback_url: str, csv_data: str, filename: str = "eis_data.csv", circuit_model: str = "auto", fitting_algorithm: str = "lm", max_iterations: int = 1000, tolerance: float = 1e-8, generate_plots: bool = True ) -> Dict[str, Any]: """ Prepare parameters for AutoEIS analysis Args: workflow_id: Unique workflow identifier node_id: Node ID in the workflow callback_url: URL to send results back to csv_data: CSV content as string filename: Name of the CSV file circuit_model: Circuit model to use ("auto" for automatic detection) fitting_algorithm: Algorithm for fitting ("lm", "trf", or "dogbox") max_iterations: Maximum fitting iterations tolerance: Fitting tolerance generate_plots: Whether to generate plots Returns: Dictionary of parameters ready for encoding """ # Encode CSV data to base64 csv_base64 = base64.b64encode(csv_data.encode()).decode() # Generate auth token auth_token = self.create_jwt_token(workflow_id) params = { "workflow_id": workflow_id, "node_id": node_id, "callback_url": callback_url, "input_data": { "csv_data": csv_base64, "filename": filename, "parameters": { "frequency_column": "frequency", "z_real_column": "z_real", "z_imag_column": "z_imag", "circuit_initial_guess": circuit_model, "fitting_algorithm": fitting_algorithm, "max_iterations": max_iterations, "tolerance": tolerance, "generate_plots": generate_plots, "save_circuit_diagram": True } }, "auth_token": auth_token } return params def generate_hf_url(self, params: Dict[str, Any]) -> str: """ Generate the Hugging Face Space URL with encoded parameters Args: params: Dictionary of parameters from prepare_analysis_params Returns: Complete URL to open the HF Space with parameters """ # Encode parameters to base64 params_json = json.dumps(params) encoded_params = base64.b64encode(params_json.encode()).decode() # Create URL url = f"{self.hf_space_url}?params={encoded_params}" return url def send_to_hf_space( self, workflow_id: str, node_id: str, callback_url: str, csv_data: str, **kwargs ) -> Dict[str, Any]: """ Send analysis request to HF Space Args: workflow_id: Unique workflow identifier node_id: Node ID in the workflow callback_url: URL to send results back to csv_data: CSV content as string **kwargs: Additional parameters for prepare_analysis_params Returns: Dictionary with action and URL """ # Prepare parameters params = self.prepare_analysis_params( workflow_id=workflow_id, node_id=node_id, callback_url=callback_url, csv_data=csv_data, **kwargs ) # Generate URL url = self.generate_hf_url(params) return { "action": "open_external", "url": url, "wait_for_callback": True, "params": params } def verify_callback_token(self, token: str) -> Dict[str, Any]: """ Verify a JWT token from callback Args: token: JWT token string Returns: Decoded payload if valid Raises: jwt.InvalidTokenError: If token is invalid """ return jwt.decode(token, self.secret_key, algorithms=['HS256']) def process_callback_results(self, callback_data: Dict[str, Any]) -> Dict[str, Any]: """ Process results received from HF Space callback Args: callback_data: Data received from callback Returns: Processed results """ results = { "workflow_id": callback_data.get("workflow_id"), "node_id": callback_data.get("node_id"), "status": callback_data.get("status"), "circuit_model": callback_data.get("results", {}).get("circuit_model"), "fit_parameters": callback_data.get("results", {}).get("fit_parameters"), "fit_error": callback_data.get("results", {}).get("fit_error"), "chi_squared": callback_data.get("results", {}).get("chi_squared"), "timestamp": callback_data.get("analysis_timestamp") } # Extract plots if available if "plots" in callback_data.get("results", {}): results["plots"] = callback_data["results"]["plots"] return results # Example usage if __name__ == "__main__": # Initialize client client = WorkflowClient( hf_space_url="https://huggingface.co/spaces/YOUR_USERNAME/autoeis-analyzer", secret_key="your-secret-key-here" ) # Sample CSV data sample_csv = """frequency,z_real,z_imag 100000,100.5,5.2 50000,102.3,8.7 10000,108.9,15.3 5000,115.2,22.1 1000,125.6,35.8 500,138.9,45.2 100,156.3,58.9 50,172.5,65.3 10,195.8,68.2 5,215.3,65.8 1,245.6,52.3 0.5,268.9,38.9 0.1,295.3,15.2""" # Prepare and send to HF Space result = client.send_to_hf_space( workflow_id="test-workflow-123", node_id="autoeis-node-1", callback_url="https://your-system.com/api/autoeis/callback", csv_data=sample_csv, circuit_model="auto", fitting_algorithm="lm" ) print("Generated URL:", result["url"]) print("\nTo integrate with your workflow system:") print("1. Open the URL in a new window/iframe") print("2. User performs analysis in HF Space") print("3. Results will be sent to your callback URL") print("4. Verify the JWT token in the callback") print("5. Process the results using process_callback_results()")