API Integration

API Integration Skill

name: api-integration
version: 1.0.0
category: programming
tags: [api, integration, orcaflex, aqwa, wamit, mock-testing, automation, offshore-software]
created: 2026-01-06
updated: 2026-01-06
author: Claude
description: |
  Expert API integration for offshore engineering software (OrcaFlex, AQWA, WAMIT)
  with mock testing strategies, error handling, and automation workflows. Enables
  development and testing without requiring the actual commercial software licenses.

When to Use This Skill

Use this skill when you need to:

• Integrate with OrcaFlex Python API
• Integrate with ANSYS AQWA or WAMIT
• Create mock APIs for testing without software licenses
• Build automation workflows for marine analysis software
• Develop robust error handling for API calls
• Implement batch processing with external software APIs
• Create abstraction layers over multiple analysis tools

Core Knowledge Areas

1. OrcaFlex API Integration

Working with OrcaFlex Python API:

import os
from pathlib import Path
from typing import Optional, Dict, List, Any
from dataclasses import dataclass
from abc import ABC, abstractmethod

# Mock OrcaFlex API for testing without license
class MockOrcFxAPI:
    """Mock OrcaFlex API for development without license."""

    class Model:
        """Mock OrcaFlex Model class."""

        def __init__(self, file_path: str = None):
            self.file_path = file_path
            self._general_data = {}
            self._objects = []
            self._is_calculated = False

        def SaveData(self, path: str):
            """Mock save data file."""
            print(f"[MOCK] Saving data to: {path}")
            self.file_path = path

        def LoadData(self, path: str):
            """Mock load data file."""
            print(f"[MOCK] Loading data from: {path}")
            self.file_path = path

        def CalculateStatics(self):
            """Mock static calculation."""
            print("[MOCK] Calculating statics...")
            self._is_calculated = True

        def RunSimulation(self):
            """Mock simulation run."""
            print("[MOCK] Running simulation...")
            self._is_calculated = True

        def SaveSimulation(self, path: str):
            """Mock save simulation."""
            print(f"[MOCK] Saving simulation to: {path}")

    @staticmethod
    def otVessel():
        return "Vessel"

    @staticmethod
    def otLine():
        return "Line"

    @staticmethod
    def ot6DBuoy():
        return "6D Buoy"

# Try to import real OrcaFlex API, fall back to mock
try:
    import OrcFxAPI
    USING_MOCK_ORCAFLEX = False
except ImportError:
    OrcFxAPI = MockOrcFxAPI
    USING_MOCK_ORCAFLEX = True
    print("Warning: OrcaFlex not available, using mock API")

@dataclass
class OrcaFlexConfig:
    """Configuration for OrcaFlex analysis."""
    model_file: Path
    output_dir: Path
    simulation_duration: float
    time_step: float
    use_variable_timestep: bool = True
    thread_count: int = 4

class OrcaFlexWrapper:
    """
    Wrapper class for OrcaFlex API with error handling and utilities.

    Example:
        >>> config = OrcaFlexConfig(
        ...     model_file=Path('mooring.dat'),
        ...     output_dir=Path('results'),
        ...     simulation_duration=3600,
        ...     time_step=0.01
        ... )
        >>> wrapper = OrcaFlexWrapper(config)
        >>> wrapper.run_analysis()
    """

    def __init__(self, config: OrcaFlexConfig):
        self.config = config
        self.model: Optional[OrcFxAPI.Model] = None
        self.is_mock = USING_MOCK_ORCAFLEX

    def load_model(self) -> None:
        """Load OrcaFlex model from file."""
        if not self.config.model_file.exists():
            raise FileNotFoundError(f"Model file not found: {self.config.model_file}")

        try:
            self.model = OrcFxAPI.Model(str(self.config.model_file))
            print(f"Loaded model: {self.config.model_file}")
        except Exception as e:
            raise RuntimeError(f"Failed to load model: {e}")

    def configure_simulation(self) -> None:
        """Configure simulation parameters."""
        if self.model is None:
            raise ValueError("Model not loaded")

        if not self.is_mock:
            # Configure real model
            general = self.model.general
            general.ImplicitUseVariableTimeStep = 'Yes' if self.config.use_variable_timestep else 'No'
            general.TargetLogSampleInterval = self.config.time_step
            general.ThreadCount = self.config.thread_count

            # Set simulation stages
            general.StageCount = 2
            general.StageDuration[0] = 100  # Build-up
            general.StageDuration[1] = self.config.simulation_duration
        else:
            print(f"[MOCK] Configured simulation: duration={self.config.simulation_duration}s")

    def run_static_analysis(self) -> bool:
        """
        Run static analysis.

        Returns:
            True if successful, False otherwise
        """
        if self.model is None:
            raise ValueError("Model not loaded")

        try:
            self.model.CalculateStatics()
            print("Static analysis complete")
            return True
        except Exception as e:
            print(f"Static analysis failed: {e}")
            return False

    def run_dynamic_analysis(self, save_results: bool = True) -> Optional[Path]:
        """
        Run dynamic analysis.

        Args:
            save_results: Whether to save simulation results

        Returns:
            Path to saved simulation file if save_results=True, else None
        """
        if self.model is None:
            raise ValueError("Model not loaded")

        try:
            self.model.RunSimulation()
            print("Dynamic analysis complete")

            if save_results:
                self.config.output_dir.mkdir(parents=True, exist_ok=True)
                sim_file = self.config.output_dir / f"{self.config.model_file.stem}.sim"
                self.model.SaveSimulation(str(sim_file))
                print(f"Results saved: {sim_file}")
                return sim_file

            return None

        except Exception as e:
            print(f"Dynamic analysis failed: {e}")
            raise

    def extract_time_series(
        self,
        object_name: str,
        variable_name: str,
        object_extra: str = 'EndA'
    ) -> tuple:
        """
        Extract time series from results.

        Args:
            object_name: Name of object
            variable_name: Variable name
            object_extra: Object extra specification

        Returns:
            Tuple of (time, values) as numpy arrays
        """
        if self.model is None:
            raise ValueError("Model not loaded")

        if self.is_mock:
            # Return mock data
            import numpy as np
            time = np.linspace(0, self.config.simulation_duration, 1000)
            values = np.random.randn(1000) * 100 + 1000  # Mock tension data
            print(f"[MOCK] Extracted time series for {object_name}.{variable_name}")
            return time, values

        # Extract from real model
        import numpy as np
        obj = self.model[object_name]

        # Handle object extra parameter
        if not self.is_mock:
            # Map string to OrcaFlex enum
            if object_extra == 'EndA':
                oe = OrcFxAPI.oeEndA
            elif object_extra == 'EndB':
                oe = OrcFxAPI.oeEndB
            else:
                oe = OrcFxAPI.oeEndA  # Default

            time = np.array(obj.TimeHistory('Time', objectExtra=oe))
            values = np.array(obj.TimeHistory(variable_name, objectExtra=oe))
        else:
            time = np.linspace(0, 3600, 1000)
            values = np.random.randn(1000) * 100 + 1000

        return time, values

    def run_analysis(self) -> Dict[str, Any]:
        """
        Complete analysis workflow.

        Returns:
            Dictionary with analysis results
        """
        # Load model
        self.load_model()

        # Configure
        self.configure_simulation()

        # Run static
        static_success = self.run_static_analysis()
        if not static_success:
            return {'status': 'FAILED', 'stage': 'static'}

        # Run dynamic
        try:
            sim_file = self.run_dynamic_analysis(save_results=True)
            return {
                'status': 'SUCCESS',
                'simulation_file': sim_file,
                'is_mock': self.is_mock
            }
        except Exception as e:
            return {
                'status': 'FAILED',
                'stage': 'dynamic',
                'error': str(e)
            }

2. Abstract API Interface Pattern

Creating abstraction layer for multiple tools:

class MarineAnalysisAPI(ABC):
    """Abstract base class for marine analysis software APIs."""

    @abstractmethod
    def create_model(self, config: dict) -> Any:
        """Create new analysis model."""
        pass

    @abstractmethod
    def load_model(self, file_path: Path) -> Any:
        """Load existing model from file."""
        pass

    @abstractmethod
    def run_static_analysis(self) -> bool:
        """Run static analysis."""
        pass

    @abstractmethod
    def run_dynamic_analysis(self) -> bool:
        """Run dynamic analysis."""
        pass

    @abstractmethod
    def extract_results(self, result_type: str) -> Dict[str, Any]:
        """Extract analysis results."""
        pass

    @abstractmethod
    def cleanup(self) -> None:
        """Cleanup resources."""
        pass

class OrcaFlexAPI(MarineAnalysisAPI):
    """OrcaFlex implementation of MarineAnalysisAPI."""

    def __init__(self):
        self.model = None
        self.is_mock = USING_MOCK_ORCAFLEX

    def create_model(self, config: dict) -> OrcFxAPI.Model:
        """Create new OrcaFlex model."""
        model = OrcFxAPI.Model()

        # Configure based on config dict
        if 'water_depth' in config:
            if not self.is_mock:
                model.environment.WaterDepth = config['water_depth']

        self.model = model
        return model

    def load_model(self, file_path: Path) -> OrcFxAPI.Model:
        """Load OrcaFlex model."""
        if not file_path.exists():
            raise FileNotFoundError(f"File not found: {file_path}")

        self.model = OrcFxAPI.Model(str(file_path))
        return self.model

    def run_static_analysis(self) -> bool:
        """Run statics."""
        if self.model is None:
            raise ValueError("No model loaded")

        try:
            self.model.CalculateStatics()
            return True
        except Exception as e:
            print(f"Static analysis error: {e}")
            return False

    def run_dynamic_analysis(self) -> bool:
        """Run dynamics."""
        if self.model is None:
            raise ValueError("No model loaded")

        try:
            self.model.RunSimulation()
            return True
        except Exception as e:
            print(f"Dynamic analysis error: {e}")
            return False

    def extract_results(self, result_type: str) -> Dict[str, Any]:
        """Extract results from model."""
        if self.model is None:
            raise ValueError("No model loaded")

        results = {}

        if result_type == 'mooring_tensions':
            # Extract mooring line tensions
            import numpy as np
            for i in range(1, 9):  # Assuming 8 mooring lines
                line_name = f"Mooring_{i}"
                try:
                    if self.is_mock:
                        time = np.linspace(0, 3600, 1000)
                        tension = np.random.randn(1000) * 200 + 1500
                    else:
                        line = self.model[line_name]
                        time = line.TimeHistory('Time')
                        tension = line.TimeHistory('Effective Tension', objectExtra=OrcFxAPI.oeEndA)

                    results[line_name] = {
                        'time': time,
                        'tension': tension,
                        'max': np.max(tension),
                        'mean': np.mean(tension)
                    }
                except:
                    pass  # Line doesn't exist

        return results

    def cleanup(self) -> None:
        """Cleanup OrcaFlex resources."""
        self.model = None

class AQWAMockAPI(MarineAnalysisAPI):
    """
    Mock AQWA API for development.

    Note: Real ANSYS AQWA integration would use ANSYS Workbench scripting or
    ANSYS ACT (Application Customization Toolkit).
    """

    def __init__(self):
        self.project = None
        self.analysis_complete = False

    def create_model(self, config: dict) -> Any:
        """Create new AQWA model (mock)."""
        print("[MOCK AQWA] Creating new model...")
        self.project = {
            'config': config,
            'geometry': None,
            'mesh': None,
            'hydrodynamic_data': None
        }
        return self.project

    def load_model(self, file_path: Path) -> Any:
        """Load AQWA model (mock)."""
        print(f"[MOCK AQWA] Loading model from: {file_path}")
        self.project = {'loaded_from': str(file_path)}
        return self.project

    def run_static_analysis(self) -> bool:
        """Run hydrostatic analysis (mock)."""
        print("[MOCK AQWA] Running hydrostatic analysis...")
        import time
        time.sleep(0.5)  # Simulate computation
        return True

    def run_dynamic_analysis(self) -> bool:
        """Run hydrodynamic analysis (mock)."""
        print("[MOCK AQWA] Running hydrodynamic analysis...")
        import time
        time.sleep(1.0)  # Simulate computation
        self.analysis_complete = True
        return True

    def extract_results(self, result_type: str) -> Dict[str, Any]:
        """Extract AQWA results (mock)."""
        import numpy as np

        if result_type == 'rao':
            # Mock RAO data
            frequencies = np.linspace(0.1, 2.0, 50)
            rao_surge = np.exp(-((frequencies - 0.5)**2) / 0.1) * 2.0
            rao_heave = np.exp(-((frequencies - 0.8)**2) / 0.1) * 1.5

            return {
                'frequencies': frequencies,
                'rao_surge': rao_surge,
                'rao_heave': rao_heave
            }

        elif result_type == 'added_mass':
            # Mock added mass matrix
            return {
                'added_mass': np.random.rand(6, 6) * 1e6,
                'damping': np.random.rand(6, 6) * 1e5
            }

        return {}

    def cleanup(self) -> None:
        """Cleanup AQWA resources (mock)."""
        self.project = None
        self.analysis_complete = False

class APIFactory:
    """Factory for creating marine analysis API instances."""

    @staticmethod
    def create_api(api_type: str) -> MarineAnalysisAPI:
        """
        Create API instance.

        Args:
            api_type: 'orcaflex', 'aqwa', 'wamit', etc.

        Returns:
            API instance

        Example:
            >>> api = APIFactory.create_api('orcaflex')
            >>> api.load_model(Path('model.dat'))
            >>> api.run_dynamic_analysis()
        """
        if api_type.lower() == 'orcaflex':
            return OrcaFlexAPI()
        elif api_type.lower() == 'aqwa':
            return AQWAMockAPI()
        elif api_type.lower() == 'wamit':
            # Would implement WAMIT API here
            raise NotImplementedError("WAMIT API not implemented")
        else:
            raise ValueError(f"Unknown API type: {api_type}")

3. Mock Testing Strategy

Testing without software licenses:

import pytest
from unittest.mock import Mock, patch, MagicMock
from pathlib import Path

class TestOrcaFlexIntegration:
    """Test suite for OrcaFlex integration using mocks."""

    @pytest.fixture
    def mock_orcaflex_model(self):
        """Create mock OrcaFlex model for testing."""
        mock_model = Mock()
        mock_model.general = Mock()
        mock_model.environment = Mock()
        mock_model.CalculateStatics = Mock()
        mock_model.RunSimulation = Mock()
        mock_model.SaveSimulation = Mock()
        return mock_model

    def test_load_model(self, mock_orcaflex_model, tmp_path):
        """Test loading OrcaFlex model."""
        # Create dummy model file
        model_file = tmp_path / "test_model.dat"
        model_file.write_text("dummy model")

        config = OrcaFlexConfig(
            model_file=model_file,
            output_dir=tmp_path / "results",
            simulation_duration=3600,
            time_step=0.1
        )

        wrapper = OrcaFlexWrapper(config)

        with patch('OrcFxAPI.Model', return_value=mock_orcaflex_model):
            wrapper.load_model()
            assert wrapper.model is not None

    def test_run_static_analysis(self, mock_orcaflex_model, tmp_path):
        """Test static analysis execution."""
        config = OrcaFlexConfig(
            model_file=tmp_path / "test.dat",
            output_dir=tmp_path / "results",
            simulation_duration=3600,
            time_step=0.1
        )

        wrapper = OrcaFlexWrapper(config)
        wrapper.model = mock_orcaflex_model

        success = wrapper.run_static_analysis()

        assert success is True
        mock_orcaflex_model.CalculateStatics.assert_called_once()

    def test_run_dynamic_analysis(self, mock_orcaflex_model, tmp_path):
        """Test dynamic analysis execution."""
        config = OrcaFlexConfig(
            model_file=tmp_path / "test.dat",
            output_dir=tmp_path / "results",
            simulation_duration=3600,
            time_step=0.1
        )

        wrapper = OrcaFlexWrapper(config)
        wrapper.model = mock_orcaflex_model

        sim_file = wrapper.run_dynamic_analysis(save_results=True)

        mock_orcaflex_model.RunSimulation.assert_called_once()
        mock_orcaflex_model.SaveSimulation.assert_called_once()
        assert sim_file is not None

    def test_api_factory(self):
        """Test API factory pattern."""
        api = APIFactory.create_api('orcaflex')
        assert isinstance(api, OrcaFlexAPI)

        api_aqwa = APIFactory.create_api('aqwa')
        assert isinstance(api_aqwa, AQWAMockAPI)

        with pytest.raises(ValueError):
            APIFactory.create_api('unknown_api')

    def test_mock_time_series_extraction(self, tmp_path):
        """Test time series extraction with mock."""
        config = OrcaFlexConfig(
            model_file=tmp_path / "test.dat",
            output_dir=tmp_path / "results",
            simulation_duration=3600,
            time_step=0.1
        )

        wrapper = OrcaFlexWrapper(config)
        wrapper.model = Mock()  # Use mock model
        wrapper.is_mock = True

        time, values = wrapper.extract_time_series(
            'Mooring_1',
            'Effective Tension'
        )

        assert len(time) > 0
        assert len(values) > 0
        assert len(time) == len(values)

def test_abstract_api_interface():
    """Test abstract API interface pattern."""

    # Test OrcaFlex API
    api = OrcaFlexAPI()
    model = api.create_model({'water_depth': 1000})
    assert model is not None

    # Test AQWA Mock API
    api_aqwa = AQWAMockAPI()
    project = api_aqwa.create_model({'water_depth': 1000})
    assert project is not None

    success = api_aqwa.run_dynamic_analysis()
    assert success is True

    results = api_aqwa.extract_results('rao')
    assert 'frequencies' in results
    assert 'rao_surge' in results

4. Error Handling and Retry Logic

Robust error handling for API calls:

import time
from functools import wraps
from typing import Callable, Any, Optional

def retry_on_failure(
    max_retries: int = 3,
    delay: float = 1.0,
    backoff: float = 2.0,
    exceptions: tuple = (Exception,)
) -> Callable:
    """
    Decorator for retrying function calls on failure.

    Args:
        max_retries: Maximum number of retry attempts
        delay: Initial delay between retries [s]
        backoff: Backoff multiplier for delay
        exceptions: Tuple of exceptions to catch

    Example:
        >>> @retry_on_failure(max_retries=3, delay=1.0)
        ... def unstable_api_call():
        ...     # Make API call that might fail
        ...     pass
    """
    def decorator(func: Callable) -> Callable:
        @wraps(func)
        def wrapper(*args, **kwargs) -> Any:
            current_delay = delay

            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    if attempt == max_retries - 1:
                        # Last attempt failed
                        raise

                    print(f"Attempt {attempt + 1} failed: {e}")
                    print(f"Retrying in {current_delay}s...")
                    time.sleep(current_delay)
                    current_delay *= backoff

            return None

        return wrapper
    return decorator

class RobustAPIWrapper:
    """API wrapper with comprehensive error handling."""

    def __init__(self, api: MarineAnalysisAPI):
        self.api = api
        self.last_error: Optional[Exception] = None

    @retry_on_failure(max_retries=3, delay=2.0)
    def safe_run_static(self) -> bool:
        """Run static analysis with retry logic."""
        try:
            result = self.api.run_static_analysis()
            return result
        except Exception as e:
            self.last_error = e
            print(f"Static analysis error: {e}")
            raise

    @retry_on_failure(max_retries=3, delay=5.0)
    def safe_run_dynamic(self) -> bool:
        """Run dynamic analysis with retry logic."""
        try:
            result = self.api.run_dynamic_analysis()
            return result
        except Exception as e:
            self.last_error = e
            print(f"Dynamic analysis error: {e}")
            raise

    def safe_extract_results(
        self,
        result_type: str,
        fallback_value: Any = None
    ) -> Any:
        """Extract results with error handling."""
        try:
            return self.api.extract_results(result_type)
        except Exception as e:
            self.last_error = e
            print(f"Result extraction error: {e}")
            return fallback_value

5. Batch Processing with APIs

Automate multiple analyses:

from concurrent.futures import ThreadPoolExecutor, as_completed
from typing import List, Dict
import pandas as pd

class BatchAnalysisRunner:
    """
    Run batch analyses using marine analysis APIs.

    Example:
        >>> parameter_sets = [
        ...     {'Hs': 5.0, 'Tp': 10.0},
        ...     {'Hs': 7.5, 'Tp': 12.0},
        ...     {'Hs': 10.0, 'Tp': 14.0}
        ... ]
        >>> runner = BatchAnalysisRunner(
        ...     api_type='orcaflex',
        ...     base_model=Path('base.dat')
        ... )
        >>> results = runner.run_batch(parameter_sets)
    """

    def __init__(
        self,
        api_type: str,
        base_model: Path,
        output_dir: Path = Path('batch_results')
    ):
        self.api_type = api_type
        self.base_model = base_model
        self.output_dir = output_dir
        self.output_dir.mkdir(parents=True, exist_ok=True)

    def run_single_analysis(
        self,
        run_id: int,
        parameters: Dict[str, Any]
    ) -> Dict[str, Any]:
        """
        Run single analysis with given parameters.

        Args:
            run_id: Unique run identifier
            parameters: Analysis parameters

        Returns:
            Results dictionary
        """
        print(f"Running analysis {run_id} with parameters: {parameters}")

        try:
            # Create API instance
            api = APIFactory.create_api(self.api_type)

            # Load base model
            api.load_model(self.base_model)

            # Apply parameters (implementation depends on API)
            # For OrcaFlex example:
            if self.api_type == 'orcaflex' and not USING_MOCK_ORCAFLEX:
                if 'Hs' in parameters:
                    api.model.environment.WaveHs = parameters['Hs']
                if 'Tp' in parameters:
                    api.model.environment.WaveTp = parameters['Tp']

            # Run analyses
            static_success = api.run_static_analysis()
            if not static_success:
                return {
                    'run_id': run_id,
                    'status': 'FAILED',
                    'stage': 'static',
                    'parameters': parameters
                }

            dynamic_success = api.run_dynamic_analysis()
            if not dynamic_success:
                return {
                    'run_id': run_id,
                    'status': 'FAILED',
                    'stage': 'dynamic',
                    'parameters': parameters
                }

            # Extract results
            results = api.extract_results('mooring_tensions')

            # Cleanup
            api.cleanup()

            return {
                'run_id': run_id,
                'status': 'SUCCESS',
                'parameters': parameters,
                'results': results
            }

        except Exception as e:
            return {
                'run_id': run_id,
                'status': 'FAILED',
                'error': str(e),
                'parameters': parameters
            }

    def run_batch(
        self,
        parameter_sets: List[Dict[str, Any]],
        parallel: bool = False,
        max_workers: int = 4
    ) -> List[Dict[str, Any]]:
        """
        Run batch analyses.

        Args:
            parameter_sets: List of parameter dictionaries
            parallel: Whether to run in parallel
            max_workers: Maximum parallel workers

        Returns:
            List of results dictionaries
        """
        results = []

        if parallel:
            # Run in parallel
            with ThreadPoolExecutor(max_workers=max_workers) as executor:
                futures = {
                    executor.submit(
                        self.run_single_analysis,
                        i + 1,
                        params
                    ): i
                    for i, params in enumerate(parameter_sets)
                }

                for future in as_completed(futures):
                    result = future.result()
                    results.append(result)
        else:
            # Run sequentially
            for i, params in enumerate(parameter_sets):
                result = self.run_single_analysis(i + 1, params)
                results.append(result)

        # Save results summary
        self._save_results_summary(results)

        return results

    def _save_results_summary(self, results: List[Dict[str, Any]]) -> None:
        """Save batch results summary to CSV."""
        summary_data = []

        for result in results:
            row = {
                'run_id': result['run_id'],
                'status': result['status'],
                **result.get('parameters', {})
            }

            # Add result metrics if available
            if 'results' in result:
                # Example: extract max tensions
                for line_name, line_results in result['results'].items():
                    row[f'{line_name}_max'] = line_results.get('max', None)

            summary_data.append(row)

        df = pd.DataFrame(summary_data)
        summary_file = self.output_dir / 'batch_results_summary.csv'
        df.to_csv(summary_file, index=False)
        print(f"Results summary saved: {summary_file}")

Complete Examples

Example 1: Multi-Tool Integration Workflow

from pathlib import Path
import numpy as np

def multi_tool_analysis_workflow(
    geometry_file: Path,
    analysis_config: dict,
    output_dir: Path
) -> dict:
    """
    Complete workflow using multiple marine analysis tools.

    Workflow:
    1. Generate hydrodynamic database with AQWA
    2. Import into OrcaFlex for dynamic analysis
    3. Extract and post-process results

    Example:
        >>> config = {
        ...     'water_depth': 1200,
        ...     'wave_height': 8.0,
        ...     'wave_period': 13.0,
        ...     'current_speed': 1.0
        ... }
        >>> results = multi_tool_analysis_workflow(
        ...     geometry_file=Path('vessel.igs'),
        ...     analysis_config=config,
        ...     output_dir=Path('integrated_analysis')
        ... )
    """
    output_dir.mkdir(parents=True, exist_ok=True)

    print("="*70)
    print("MULTI-TOOL ANALYSIS WORKFLOW")
    print("="*70)

    # Step 1: AQWA Hydrodynamic Analysis
    print("\n[Step 1/3] Running AQWA hydrodynamic analysis...")

    aqwa_api = APIFactory.create_api('aqwa')

    # Create AQWA model
    aqwa_config = {
        'geometry': geometry_file,
        'water_depth': analysis_config['water_depth'],
        'frequency_range': (0.1, 2.0),
        'n_frequencies': 50
    }

    aqwa_api.create_model(aqwa_config)

    # Run hydrodynamic analysis
    aqwa_api.run_dynamic_analysis()

    # Extract RAOs and hydrodynamic coefficients
    rao_data = aqwa_api.extract_results('rao')
    hydro_coeffs = aqwa_api.extract_results('added_mass')

    print("AQWA analysis complete")
    print(f"  RAO frequencies: {len(rao_data['frequencies'])}")

    # Step 2: Create OrcaFlex Model with AQWA Data
    print("\n[Step 2/3] Creating OrcaFlex model...")

    orcaflex_api = APIFactory.create_api('orcaflex')

    # Create model
    orcaflex_config = {
        'water_depth': analysis_config['water_depth'],
        'wave_height': analysis_config['wave_height'],
        'wave_period': analysis_config['wave_period'],
        'current_speed': analysis_config['current_speed']
    }

    orcaflex_api.create_model(orcaflex_config)

    # In real implementation, would import AQWA hydrodynamic database
    # For mock, we proceed with standard analysis

    # Run OrcaFlex analysis
    print("\n[Step 3/3] Running OrcaFlex dynamic analysis...")

    robust_wrapper = RobustAPIWrapper(orcaflex_api)

    static_success = robust_wrapper.safe_run_static()
    if not static_success:
        return {'status': 'FAILED', 'stage': 'static'}

    dynamic_success = robust_wrapper.safe_run_dynamic()
    if not dynamic_success:
        return {'status': 'FAILED', 'stage': 'dynamic'}

    # Extract results
    mooring_results = robust_wrapper.safe_extract_results(
        'mooring_tensions',
        fallback_value={}
    )

    print("\nOrcaFlex analysis complete")

    # Cleanup
    aqwa_api.cleanup()
    orcaflex_api.cleanup()

    # Combine results
    final_results = {
        'status': 'SUCCESS',
        'aqwa_results': {
            'rao': rao_data,
            'hydrodynamic_coefficients': hydro_coeffs
        },
        'orcaflex_results': {
            'mooring_tensions': mooring_results
        }
    }

    print("\n" + "="*70)
    print("WORKFLOW COMPLETE")
    print("="*70)

    return final_results

# Run workflow
geometry_file = Path('fpso_geometry.igs')
config = {
    'water_depth': 1200,
    'wave_height': 8.0,
    'wave_period': 13.0,
    'current_speed': 1.0
}

workflow_results = multi_tool_analysis_workflow(
    geometry_file=geometry_file,
    analysis_config=config,
    output_dir=Path('integrated_analysis_results')
)

print(f"\nFinal status: {workflow_results['status']}")

Best Practices

1. API Availability Checking

def check_api_availability(api_type: str) -> tuple[bool, str]:
    """
    Check if API is available and return status.

    Args:
        api_type: Type of API to check

    Returns:
        Tuple of (is_available, message)

    Example:
        >>> available, msg = check_api_availability('orcaflex')
        >>> if available:
        ...     print("OrcaFlex API is ready")
        >>> else:
        ...     print(f"Using mock: {msg}")
    """
    if api_type == 'orcaflex':
        try:
            import OrcFxAPI
            return True, "OrcaFlex API available"
        except ImportError:
            return False, "OrcaFlex not installed, using mock API"

    elif api_type == 'aqwa':
        # AQWA typically accessed via ANSYS Workbench
        # Check if ANSYS is available
        return False, "AQWA integration via ANSYS Workbench (mock mode)"

    else:
        return False, f"Unknown API type: {api_type}"

2. Configuration Management

import yaml
from dataclasses import dataclass, asdict

@dataclass
class APIConfiguration:
    """Configuration for API integration."""
    api_type: str
    model_file: Optional[Path]
    output_dir: Path
    simulation_settings: dict
    retry_settings: dict

    def save_to_yaml(self, file_path: Path) -> None:
        """Save configuration to YAML file."""
        with open(file_path, 'w') as f:
            yaml.dump(asdict(self), f, default_flow_style=False)

    @classmethod
    def load_from_yaml(cls, file_path: Path) -> 'APIConfiguration':
        """Load configuration from YAML file."""
        with open(file_path) as f:
            data = yaml.safe_load(f)
        # Convert Path strings back to Path objects
        if 'model_file' in data and data['model_file']:
            data['model_file'] = Path(data['model_file'])
        data['output_dir'] = Path(data['output_dir'])
        return cls(**data)

3. Logging and Monitoring

import logging
from datetime import datetime

def setup_api_logging(
    log_dir: Path,
    api_type: str
) -> logging.Logger:
    """
    Setup logging for API operations.

    Args:
        log_dir: Directory for log files
        api_type: Type of API

    Returns:
        Configured logger
    """
    log_dir.mkdir(parents=True, exist_ok=True)

    timestamp = datetime.now().strftime('%Y%m%d_%H%M%S')
    log_file = log_dir / f"{api_type}_api_{timestamp}.log"

    logger = logging.getLogger(f"{api_type}_api")
    logger.setLevel(logging.DEBUG)

    # File handler
    fh = logging.FileHandler(log_file)
    fh.setLevel(logging.DEBUG)

    # Console handler
    ch = logging.StreamHandler()
    ch.setLevel(logging.INFO)

    # Formatter
    formatter = logging.Formatter(
        '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
    )
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)

    logger.addHandler(fh)
    logger.addHandler(ch)

    return logger

Resources

OrcaFlex API

• Documentation: OrcFxAPI Python documentation (in OrcaFlex installation)
• Examples: OrcaFlex → Examples → Python folder
• Support: support@orcina.com

ANSYS AQWA

• ANSYS ACT: Application Customization Toolkit for scripting
• PyAnsys: https://github.com/pyansys
• Documentation: ANSYS Help → AQWA → Programmer's Guide

Testing

• pytest: https://docs.pytest.org/
• unittest.mock: https://docs.python.org/3/library/unittest.mock.html
• Mock Testing Best Practices: Various online resources

Python API Design

• PEP 8: Python style guide
• Design Patterns: Gang of Four patterns
• Abstract Base Classes: Python ABC module documentation

Use this skill for: Expert API integration with marine analysis software, mock testing strategies, and automation workflows with robust error handling.