pcb-afm-simulation-system/web/backend/app/services/rule_engine.py

"""Rule engine for AFM motor simulation plan generation.

Provides:
- Scan parameter registry (Motor-CAD variable names, units, physical ranges)
- Boundary-condition-based range recommendation
- Initial plan generation from project boundary conditions

All source is ASCII. Chinese labels use Unicode escapes.
"""

from __future__ import annotations

import math
from dataclasses import dataclass, field
from typing import Any


# ---------------------------------------------------------------------------
# Scan parameter registry
# ---------------------------------------------------------------------------

@dataclass
class ScanParameter:
    """Definition of a scannable Motor-CAD parameter."""
    name: str                    # Motor-CAD variable name (exact)
    display_name: str            # Human-readable label
    unit: str                    # Unit string
    category: str                # Geometry / Electrical / Thermal / Mesh
    default_start: float         # Recommended range start
    default_stop: float          # Recommended range stop
    default_step: float          # Recommended step
    min_allowed: float           # Hard physical minimum
    max_allowed: float           # Hard physical maximum
    description: str = ""
    topology_supported: list[str] = field(default_factory=lambda: ["SSSR", "DRSS", "SDSR"])


# Registry of scannable parameters for AFM motors.
# Ranges based on MARS-12S10P reference model and AFM design literature.
SCAN_PARAMETERS: dict[str, ScanParameter] = {
    # --- Geometry ---
    "Airgap": ScanParameter(
        name="Airgap",
        display_name="Airgap Length",
        unit="mm",
        category="Geometry",
        default_start=0.6, default_stop=1.5, default_step=0.3,
        min_allowed=0.2, max_allowed=5.0,
        description="Mechanical airgap between stator and rotor.",
    ),
    "Magnet_Length": ScanParameter(
        name="Magnet_Length",
        display_name="Magnet Axial Thickness",
        unit="mm",
        category="Geometry",
        default_start=2.0, default_stop=5.0, default_step=1.0,
        min_allowed=0.5, max_allowed=15.0,
        description="Magnet axial thickness (NOT radial depth).",
    ),
    "Magnet_Thickness": ScanParameter(
        name="Magnet_Thickness",
        display_name="Magnet Radial Depth",
        unit="mm",
        category="Geometry",
        default_start=8.0, default_stop=16.0, default_step=2.0,
        min_allowed=2.0, max_allowed=40.0,
        description="Magnet ring radial depth = (D_out - D_in) / 2.",
    ),
    "Magnet_Arc_[ED]": ScanParameter(
        name="Magnet_Arc_[ED]",
        display_name="Magnet Pole Arc",
        unit="deg",
        category="Geometry",
        default_start=100.0, default_stop=140.0, default_step=10.0,
        min_allowed=30.0, max_allowed=180.0,
        description="Magnet pole arc in electrical degrees.",
    ),
    # --- Electrical ---
    "RMSCurrent": ScanParameter(
        name="RMSCurrent",
        display_name="RMS Phase Current",
        unit="A",
        category="Electrical",
        default_start=10.0, default_stop=30.0, default_step=5.0,
        min_allowed=0.5, max_allowed=200.0,
        description="RMS phase current (CurrentDefinition=1).",
    ),
    "Shaft_Speed": ScanParameter(
        name="Shaft_Speed",
        display_name="Shaft Speed",
        unit="rpm",
        category="Electrical",
        default_start=3000.0, default_stop=8000.0, default_step=1000.0,
        min_allowed=100.0, max_allowed=50000.0,
        description="Rotational shaft speed.",
    ),
    # --- Thermal ---
    "Magnet_Temperature": ScanParameter(
        name="Magnet_Temperature",
        display_name="Magnet Temperature",
        unit="C",
        category="Thermal",
        default_start=60.0, default_stop=120.0, default_step=20.0,
        min_allowed=-40.0, max_allowed=200.0,
        description="Magnet operating temperature (default 100C hot).",
    ),
    # --- Mesh ---
    "TorquePointsPerCycle": ScanParameter(
        name="TorquePointsPerCycle",
        display_name="Torque Points / Cycle",
        unit="pts",
        category="Mesh",
        default_start=60.0, default_stop=180.0, default_step=60.0,
        min_allowed=12.0, max_allowed=720.0,
        description="Torque sampling points per electrical cycle.",
    ),
}


def get_parameter_registry() -> list[dict[str, Any]]:
    """Return the scan parameter registry as a list of dicts (for API)."""
    return [
        {
            "name": p.name,
            "display_name": p.display_name,
            "unit": p.unit,
            "category": p.category,
            "default_start": p.default_start,
            "default_stop": p.default_stop,
            "default_step": p.default_step,
            "min_allowed": p.min_allowed,
            "max_allowed": p.max_allowed,
            "description": p.description,
            "topology_supported": p.topology_supported,
        }
        for p in SCAN_PARAMETERS.values()
    ]


def get_parameter(name: str) -> ScanParameter | None:
    """Get a parameter definition by name."""
    return SCAN_PARAMETERS.get(name)


# ---------------------------------------------------------------------------
# Boundary condition based range recommendation
# ---------------------------------------------------------------------------

@dataclass
class BoundaryConditions:
    """Parsed project boundary conditions."""
    topology: str = "SSSR"
    outer_diameter_mm: float | None = None
    inner_diameter_mm: float | None = None
    speed_rpm: float | None = None
    current_a: float | None = None
    magnet_temp_c: float | None = None
    target_torque_nm: float | None = None
    target_efficiency_pct: float | None = None
    max_losses_w: float | None = None
    raw: dict[str, Any] = field(default_factory=dict)

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> "BoundaryConditions":
        """Parse boundary conditions from a project's JSON dict."""
        if not data:
            return cls()
        return cls(
            topology=data.get("topology", "SSSR"),
            outer_diameter_mm=_to_float(data.get("outer_diameter_mm")),
            inner_diameter_mm=_to_float(data.get("inner_diameter_mm")),
            speed_rpm=_to_float(data.get("speed_rpm")),
            current_a=_to_float(data.get("current_a")),
            magnet_temp_c=_to_float(data.get("magnet_temp_c")),
            target_torque_nm=_to_float(data.get("target_torque_nm")),
            target_efficiency_pct=_to_float(data.get("target_efficiency_pct")),
            max_losses_w=_to_float(data.get("max_losses_w")),
            raw=data,
        )


def _to_float(v: Any) -> float | None:
    """Safely convert a value to float."""
    if v is None:
        return None
    try:
        return float(v)
    except (ValueError, TypeError):
        return None


def recommend_range(
    param_name: str,
    bc: BoundaryConditions,
) -> dict[str, float]:
    """Recommend a scan range for a parameter based on boundary conditions.

    Returns dict with start, stop, step, and a confidence note.
    Adjustments are heuristic and based on AFM design scaling rules.
    """
    p = get_parameter(param_name)
    if p is None:
        return {"start": 0.0, "stop": 1.0, "step": 0.5}

    start = p.default_start
    stop = p.default_stop
    step = p.default_step
    notes: list[str] = []

    # --- Airgap: scale with outer diameter ---
    if param_name == "Airgap":
        if bc.outer_diameter_mm:
            # Smaller machines tend to use smaller airgaps.
            # Airgap ~ 0.5-2% of outer diameter for AFM.
            d = bc.outer_diameter_mm
            start = max(p.min_allowed, round(d * 0.008, 2))   # ~0.8% D
            stop = min(p.max_allowed, round(d * 0.02, 2))     # ~2% D
            step = round((stop - start) / 3, 2)
            notes.append(f"scaled from D={d}mm")
        if bc.target_torque_nm and bc.target_torque_nm > 5:
            # Higher torque targets benefit from smaller airgap exploration.
            start = max(p.min_allowed, start * 0.8)
            notes.append("torque target > 5Nm: shifted lower")

    # --- Magnet axial thickness: scale with diameter and speed ---
    elif param_name == "Magnet_Length":
        if bc.outer_diameter_mm:
            d = bc.outer_diameter_mm
            start = max(p.min_allowed, round(d * 0.03, 1))   # ~3% D
            stop = min(p.max_allowed, round(d * 0.07, 1))    # ~7% D
            step = round((stop - start) / 3, 1)
            notes.append(f"scaled from D={d}mm")
        if bc.speed_rpm and bc.speed_rpm > 10000:
            # High speed: thinner magnets reduce eddy current loss.
            stop = min(stop, p.default_stop)
            notes.append("high speed: capped upper range")

    # --- Magnet radial depth: scale with diameter ratio ---
    elif param_name == "Magnet_Thickness":
        if bc.outer_diameter_mm and bc.inner_diameter_mm:
            radial_depth = (bc.outer_diameter_mm - bc.inner_diameter_mm) / 2
            start = max(p.min_allowed, round(radial_depth * 0.6, 1))
            stop = min(p.max_allowed, round(radial_depth * 1.0, 1))
            step = round((stop - start) / 3, 1)
            notes.append(f"scaled from radial depth={radial_depth}mm")
        elif bc.outer_diameter_mm:
            d = bc.outer_diameter_mm
            start = max(p.min_allowed, round(d * 0.1, 1))
            stop = min(p.max_allowed, round(d * 0.25, 1))
            step = round((stop - start) / 3, 1)
            notes.append(f"scaled from D={d}mm")

    # --- Pole arc: standard range, adjust for torque/ripple targets ---
    elif param_name == "Magnet_Arc_[ED]":
        if bc.target_torque_nm:
            # Wider arc -> more torque but more ripple.
            start = 110.0
            stop = 150.0
            notes.append("torque target: widened upper range")
        else:
            start = 100.0
            stop = 140.0

    # --- RMS current: center around specified current ---
    elif param_name == "RMSCurrent":
        if bc.current_a:
            i = bc.current_a
            start = max(p.min_allowed, round(i * 0.5, 1))
            stop = min(p.max_allowed, round(i * 1.5, 1))
            step = round((stop - start) / 4, 1)
            notes.append(f"centered on I={i}A")
        if bc.max_losses_w:
            # Loss-limited: don't go too high on current.
            stop = min(stop, p.default_stop * 1.2)
            notes.append("loss constraint: capped upper")

    # --- Shaft speed: center around specified speed ---
    elif param_name == "Shaft_Speed":
        if bc.speed_rpm:
            s = bc.speed_rpm
            start = max(p.min_allowed, round(s * 0.6, -2))
            stop = min(p.max_allowed, round(s * 1.4, -2))
            step = round((stop - start) / 4, -2)
            notes.append(f"centered on N={s}rpm")

    # --- Magnet temperature: hot/cold sweep ---
    elif param_name == "Magnet_Temperature":
        if bc.magnet_temp_c:
            t = bc.magnet_temp_c
            start = max(p.min_allowed, t - 40)
            stop = min(p.max_allowed, t + 40)
            step = 20.0
            notes.append(f"centered on T={t}C")
        else:
            start = 20.0   # cold
            stop = 120.0   # hot
            step = 20.0
            notes.append("default cold-to-hot sweep")

    # --- Torque points: based on speed/accuracy needs ---
    elif param_name == "TorquePointsPerCycle":
        start = 60.0
        stop = 180.0
        step = 60.0
        if bc.target_efficiency_pct and bc.target_efficiency_pct > 90:
            stop = 240.0
            notes.append("high efficiency target: extended upper range")

    # Clamp to physical limits
    start = max(p.min_allowed, min(p.max_allowed, start))
    stop = max(p.min_allowed, min(p.max_allowed, stop))
    if start >= stop:
        start = p.default_start
        stop = p.default_stop
    if step <= 0:
        step = p.default_step

    return {
        "start": round(start, 4),
        "stop": round(stop, 4),
        "step": round(step, 4),
        "notes": "; ".join(notes) if notes else "default range",
    }


# ---------------------------------------------------------------------------
# Plan generation
# ---------------------------------------------------------------------------

def generate_values(start: float, stop: float, step: float) -> list[float]:
    """Generate evenly spaced values from start to stop inclusive.

    Uses round to avoid floating point artifacts.
    """
    if step <= 0:
        return [start]
    count = int(math.floor((stop - start) / step + 1e-9)) + 1
    values = [round(start + i * step, 6) for i in range(count)]
    # Ensure stop is included if rounding pushed it slightly off
    if values and abs(values[-1] - stop) > 1e-9:
        values.append(round(stop, 6))
    return values


def estimate_point_count(variables: list[dict[str, Any]]) -> int:
    """Estimate total scan points (Cartesian product)."""
    total = 1
    for v in variables:
        if v.get("values"):
            total *= len(v["values"])
        elif v.get("start") is not None and v.get("stop") is not None and v.get("step"):
            total *= len(generate_values(v["start"], v["stop"], v["step"]))
        else:
            total *= 1
    return total


def generate_plan(
    bc: BoundaryConditions,
    param_names: list[str] | None = None,
    model_path: str = "",
) -> dict[str, Any]:
    """Generate a recommended simulation plan from boundary conditions.

    Args:
        bc: Parsed boundary conditions.
        param_names: List of parameter names to include. If None, uses a
            default set (Airgap, Magnet_Length, RMSCurrent).
        model_path: Path to the baseline .mot model.

    Returns:
        Plan dict compatible with src/plan_schema.py and the plans API.
    """
    if param_names is None:
        # Default scan set: geometry + electrical
        param_names = ["Airgap", "Magnet_Length", "RMSCurrent"]

    variables = []
    for name in param_names:
        p = get_parameter(name)
        if p is None:
            continue
        if bc.topology not in p.topology_supported:
            continue
        rng = recommend_range(name, bc)
        values = generate_values(rng["start"], rng["stop"], rng["step"])
        variables.append({
            "name": name,
            "display_name": p.display_name,
            "unit": p.unit,
            "start": rng["start"],
            "stop": rng["stop"],
            "step": rng["step"],
            "values": values,
            "recommendation_notes": rng["notes"],
        })

    total_points = estimate_point_count(variables)
    plan_name = _generate_plan_name(bc, param_names)

    return {
        "name": plan_name,
        "model_path": model_path,
        "topology": bc.topology,
        "variables": variables,
        "cases": [],
        "estimated_points": total_points,
        "estimated_time_min": total_points * 3,  # ~3 min per point
        "generation_summary": {
            "boundary_conditions_used": {
                k: v for k, v in bc.raw.items() if v is not None
            },
            "parameters_scanned": [v["name"] for v in variables],
            "total_points": total_points,
            "rule_engine_version": "1.0",
        },
    }


def _generate_plan_name(bc: BoundaryConditions, param_names: list[str]) -> str:
    """Generate a human-readable plan name."""
    short = {
        "Airgap": "Ag",
        "Magnet_Length": "Lm",
        "Magnet_Thickness": "Tm",
        "Magnet_Arc_[ED]": "Arc",
        "RMSCurrent": "I",
        "Shaft_Speed": "N",
        "Magnet_Temperature": "Tmag",
        "TorquePointsPerCycle": "Pts",
    }
    tags = [short.get(n, n) for n in param_names]
    parts = ["Auto"]
    if bc.topology:
        parts.append(bc.topology)
    parts.append("-".join(tags))
    if bc.outer_diameter_mm:
        parts.append(f"D{bc.outer_diameter_mm:g}")
    return "_".join(parts)