Spaces:
Sleeping
Sleeping
| """ | |
| 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()") |