PHYSICS_MODULE_STANDARDIZATION.md

Physics Module Standardization Guide

Status: Implementation in progress
Version: 1.0
Last Updated: 2026-02-23

0️⃣ GLOBAL CONTRACT (MANDATORY FOR ALL MODULES)

Every physics module MUST conform to this interface:

class PhysicsModule(ABC):
    
    def initialize(self, config: dict) -> None:
        """Load constants, precompute tables, allocate memory."""
    
    def compute_derivatives(
        self,
        state: SpacecraftState,
        environment: EnvironmentState,
        t: float
    ) -> StateDerivative:
        """Return contribution to state derivative."""
    
    def compute_outputs(
        self,
        state: SpacecraftState,
        environment: EnvironmentState,
        t: float
    ) -> dict:
        """Return diagnostic metrics only (no state mutation)."""
    
    def validate(self) -> bool:
        """Check module internal state is consistent."""
    
    def diagnostics(self) -> dict:
        """Return diagnostic data for logging."""
    
    def get_required_state_fields(self) -> list[str]:
        """Declare which state fields this module reads."""
    
    def get_updated_state_fields(self) -> list[str]:
        """Declare which derivative fields this module contributes."""
    
    def get_module_type(self) -> PhysicsModuleType:
        """Return the type of physics this module computes."""

1️⃣ COMPLIANCE CHECKLIST

✅ Methods (7 REQUIRED)

• initialize(config) - Setup, no exceptions
• compute_derivatives(state, env, t) - Pure, deterministic, returns StateDerivative
• compute_outputs(state, env, t) - Diagnostic output, pure function
• validate() - Boolean state check
• diagnostics() - Return dict (never None)
• get_required_state_fields() - Return list of field names
• get_updated_state_fields() - Return list of derivative field names

✅ Error Handling

• No exceptions in compute_derivatives()
• No exceptions in compute_outputs()
• Return zero-derivatives on compute failure
• Set is_valid = False on init failure
• Store errors in self.last_error for diagnostics

✅ Type Safety

• All inputs are immutable (frozen dataclasses)
• All outputs are properly typed (dict, StateDerivative, bool, list)
• Never return None for dict/list outputs (return empty instead)

✅ Documentation

• Class docstring explains physics/purpose
• Method docstrings document parameters and returns
• Specify which state/env fields are read/written
• Add example usage in docstring

2️⃣ IMPLEMENTATION TEMPLATES

Template A: Utility-Based Module (Simple Computations)

Use when: Module wraps existing utility functions that compute physics

from models.core import PhysicsModule, PhysicsModuleType, StateDerivative
from typing import Dict, List, Any

class MyUtilityModule(PhysicsModule):
    """Module that wraps utility functions.
    
    Reads: state fields X, Y, Z
    Writes: acceleration
    
    Example:
        module = MyUtilityModule()
        module.initialize({"param1": 1.0})
        derivatives = module.compute_derivatives(state, env, t)
    """
    
    def __init__(self):
        self.config: Dict[str, Any] = {}
        self.is_valid = False
        self.last_error: str = None
    
    def initialize(self, config: Dict[str, Any]) -> None:
        try:
            self.config = config
            # Validate parameters
            assert "param1" in config
            self.is_valid = True
            self.last_error = None
        except Exception as e:
            self.is_valid = False
            self.last_error = str(e)
    
    def compute_derivatives(self, state, environment, t: float) -> StateDerivative:
        try:
            if not self.is_valid:
                return StateDerivative(
                    velocity_derivative=(0, 0, 0),
                    acceleration=(0, 0, 0),
                    angular_acceleration=(0, 0, 0)
                )
            
            # Extract from state/environment
            position = state.position
            velocity = state.velocity
            
            # Call utility function
            accel = compute_force(position, velocity, self.config["param1"])
            
            return StateDerivative(
                velocity_derivative=velocity,          # dr/dt = v
                acceleration=accel,
                angular_acceleration=(0, 0, 0)
            )
        except Exception as e:
            self.last_error = str(e)
            return StateDerivative(
                velocity_derivative=(0, 0, 0),
                acceleration=(0, 0, 0),
                angular_acceleration=(0, 0, 0)
            )
    
    def compute_outputs(self, state, environment, t: float) -> Dict[str, Any]:
        try:
            if not self.is_valid:
                return {"error": self.last_error}
            
            # Compute diagnostic quantities
            magnitude = compute_magnitude(state.position)
            
            return {
                "magnitude": magnitude,
                "timestamp": t,
                "is_valid": self.is_valid
            }
        except Exception as e:
            self.last_error = str(e)
            return {"error": str(e)}
    
    def validate(self) -> bool:
        return self.is_valid
    
    def diagnostics(self) -> Dict[str, Any]:
        return {
            "is_valid": self.is_valid,
            "last_error": self.last_error,
            "config": self.config
        }
    
    def get_required_state_fields(self) -> List[str]:
        return ["position", "velocity"]
    
    def get_updated_state_fields(self) -> List[str]:
        return ["acceleration"]
    
    def get_module_type(self):
        return PhysicsModuleType.ORBITAL_MECHANICS
    
    def compute(self, physics_input):
        """Legacy interface - deprecated."""
        try:
            derivatives = self.compute_derivatives(
                physics_input.state,
                physics_input.environment,
                physics_input.t_epoch
            )
            diags = self.diagnostics()
            outputs = self.compute_outputs(
                physics_input.state,
                physics_input.environment,
                physics_input.t_epoch
            )
            diags.update(outputs)
            
            from models.core import PhysicsOutput
            return PhysicsOutput(
                state_derivative=derivatives,
                diagnostics=diags,
                events=[],
                valid=self.is_valid
            )
        except Exception as e:
            self.last_error = str(e)
            from models.core import PhysicsOutput, StateDerivative
            return PhysicsOutput(
                state_derivative=StateDerivative(
                    velocity_derivative=(0, 0, 0),
                    acceleration=(0, 0, 0),
                    angular_acceleration=(0, 0, 0)
                ),
                diagnostics={"error": str(e)},
                events=[],
                valid=False
            )

Template B: Environment Module (Provides Data Only)

Use when: Module computes environment state, not spacecraft derivatives

from models.core import PhysicsModule, PhysicsModuleType, StateDerivative
from typing import Dict, List, Any

class MyEnvironmentModule(PhysicsModule):
    """Environment module - provides data, no state updates.
    
    Reads: (environment data, position for distance calcs)
    Writes: (nothing - returns zero derivatives)
    
    Example:
        module = MyEnvironmentModule()
        module.initialize({})
        outputs = module.compute_outputs(state, env, t)
        # outputs contains: solar_wind_speed, solar_wind_density, ...
    """
    
    def __init__(self):
        self.is_valid = False
        self.last_error: str = None
    
    def initialize(self, config: Dict[str, Any]) -> None:
        try:
            # Load/validate environment parameters
            self.is_valid = True
            self.last_error = None
        except Exception as e:
            self.is_valid = False
            self.last_error = str(e)
    
    def compute_derivatives(self, state, environment, t: float) -> StateDerivative:
        """Environment modules return zero derivatives."""
        return StateDerivative(
            velocity_derivative=(0, 0, 0),
            acceleration=(0, 0, 0),
            angular_acceleration=(0, 0, 0)
        )
    
    def compute_outputs(self, state, environment, t: float) -> Dict[str, Any]:
        """Return environmental data at given time."""
        try:
            # Compute environment quantities
            solar_wind_speed = compute_solar_wind_speed(t)
            solar_wind_density = compute_solar_wind_density(t)
            
            return {
                "solar_wind_speed": solar_wind_speed,
                "solar_wind_density": solar_wind_density,
                "timestamp": t
            }
        except Exception as e:
            self.last_error = str(e)
            return {"error": str(e), "timestamp": t}
    
    def validate(self) -> bool:
        return self.is_valid
    
    def diagnostics(self) -> Dict[str, Any]:
        return {
            "is_valid": self.is_valid,
            "last_error": self.last_error
        }
    
    def get_required_state_fields(self) -> List[str]:
        return []  # Environment modules don't read spacecraft state
    
    def get_updated_state_fields(self) -> List[str]:
        return []  # Environment modules don't update state
    
    def get_module_type(self):
        return PhysicsModuleType.ENVIRONMENT

Template C: Complex Module (State + Telemetry)

Use when: Module has complex internal state and multiple outputs

from models.core import PhysicsModule, PhysicsModuleType, StateDerivative
from typing import Dict, List, Any

class MyComplexModule(PhysicsModule):
    """Complex module with internal state tracking.
    
    Example:
        module = MyComplexModule()
        module.initialize({"param": 1.0})
        for t in times:
            deriv = module.compute_derivatives(state, env, t)
            output = module.compute_outputs(state, env, t)
    """
    
    def __init__(self):
        self.config = {}
        self.is_valid = False
        self.last_error = None
        self.internal_state = {}  # Track any internal computations
    
    def initialize(self, config: Dict[str, Any]) -> None:
        try:
            self.config = config
            self.internal_state = {"iterations": 0}
            self.is_valid = True
            self.last_error = None
        except Exception as e:
            self.is_valid = False
            self.last_error = str(e)
    
    def compute_derivatives(self, state, environment, t: float) -> StateDerivative:
        try:
            if not self.is_valid:
                return zero_derivatives()
            
            # Main physics computation
            accel = self._compute_acceleration(state, environment, t)
            
            # Update internal tracking (no side effects on input)
            self.internal_state["last_accel"] = accel
            
            return StateDerivative(
                velocity_derivative=state.velocity,
                acceleration=accel,
                angular_acceleration=(0, 0, 0)
            )
        except Exception as e:
            self.last_error = str(e)
            return zero_derivatives()
    
    def compute_outputs(self, state, environment, t: float) -> Dict[str, Any]:
        try:
            if not self.is_valid:
                return {"error": self.last_error}
            
            # Return telemetry/diagnostics
            outputs = {
                "position_magnitude": magnitude(state.position),
                "velocity_magnitude": magnitude(state.velocity),
                "timestamp": t
            }
            
            # Include internal state info
            if "last_accel" in self.internal_state:
                outputs["accel_magnitude"] = magnitude(
                    self.internal_state["last_accel"]
                )
            
            return outputs
        except Exception as e:
            self.last_error = str(e)
            return {"error": str(e)}
    
    def validate(self) -> bool:
        return self.is_valid
    
    def diagnostics(self) -> Dict[str, Any]:
        return {
            "is_valid": self.is_valid,
            "last_error": self.last_error,
            "iterations": self.internal_state.get("iterations", 0)
        }
    
    def get_required_state_fields(self) -> List[str]:
        return ["position", "velocity", "mass"]
    
    def get_updated_state_fields(self) -> List[str]:
        return ["acceleration"]
    
    def get_module_type(self):
        return PhysicsModuleType.ORBITAL_MECHANICS
    
    def _compute_acceleration(self, state, environment, t):
        """Internal computation - can raise exceptions."""
        return (0, 0, 0)

3️⃣ PHYSICS MODULES TO REFACTOR

Environment Models (8 modules)

• cme.py ✅ (in progress)
• micrometeoroid.py
• parker_spiral.py
• radiation_belts.py
• radiation_flux.py
• solar_cycle.py
• solar_wind.py
• thermal_loading.py

Orbital Mechanics (7 modules)

• two_body.py
• j2.py
• low_thrust.py
• n_body.py
• perturbations.py
• srp.py
• elements.py

Sail Physics (3 modules)

• srp_core.py
• thermal.py
• reflectivity.py

Propulsion (9 modules)

• ion_thrust.py
• thruster.py
• mhd.py
• plasma.py
• electric_field.py
• efficiency.py
• power.py
• power_coupling.py
• paschen.py

Structural Dynamics (6 modules)

• fem.py
• fem_membrane.py
• membrane.py
• modal.py
• buckling.py
• deployment.py

4️⃣ REFACTORING STRATEGY

Phase 1: High-Priority Modules

1. two_body.py - Core orbital mechanics used by kernel
2. srp_core.py - Core sail physics
3. ion_thrust.py - Core propulsion

Phase 2: Environment Modules

All 8 environment modules following Template B

Phase 3: Complex Modules

Modal analysis, FEM, MHD (use Template C)

Phase 4: Utility Functions

Remaining utility modules wrapped using Template A

5️⃣ TESTING STRATEGY

For each refactored module:

def test_module_conforms_to_contract():
    """Verify module implements all required methods."""
    module = MyModule()
    
    # Check all methods exist and are callable
    assert hasattr(module, "initialize")
    assert hasattr(module, "compute_derivatives")
    assert hasattr(module, "compute_outputs")
    assert hasattr(module, "validate")
    assert hasattr(module, "diagnostics")
    assert hasattr(module, "get_required_state_fields")
    assert hasattr(module, "get_updated_state_fields")
    assert hasattr(module, "get_module_type")
    
    # Test initialization
    module.initialize({})
    assert module.validate() in (True, False)
    
    # Test derivatives
    deriv = module.compute_derivatives(state, env, 0.0)
    assert isinstance(deriv, StateDerivative)
    
    # Test outputs
    outputs = module.compute_outputs(state, env, 0.0)
    assert isinstance(outputs, dict)
    
    # Test field declarations
    required_fields = module.get_required_state_fields()
    assert isinstance(required_fields, list)
    
    updated_fields = module.get_updated_state_fields()
    assert isinstance(updated_fields, list)

6️⃣ INTEGRATION CHECKLIST

• All 33 physics modules implement interface
• All modules pass contract tests
• Kernel can discover and validate all modules
• Field dependency tracking works
• All existing unit tests pass
• Integration tests pass (no regressions)
• Documentation updated for each module
• Type hints validated with mypy/pylance

7️⃣ COMPLIANCE VALIDATION

After refactoring all modules, run:

python -m pytest tests/test_physics_module_contract.py -v
python -m mypy kernel/ --strict

Expected output:

• All 33 modules pass contract tests ✅
• All modules have no type errors ✅
• Field dependency graph is complete ✅

---

Next Steps:

1. Refactor two_body.py (orbital mechanics)
2. Refactor srp_core.py (sail physics)
3. Refactor ion_thrust.py (propulsion)
4. Create base classes for each physics domain
5. Document field dependencies in kernel