sprint-4: Project Editor + .appack compiler + IRudderActuator HAL + dual sensor config

Cherry-pick of ab28cb7 onto main (was missing from main branch history).

Python:
- arautopilot/core/sensor_config.py: RudderSensorConfig + DualRudderSensorConfig
  (AS5048A SPI / potentiometer, dual with cross-validation thresholds)
- arautopilot/core/vessel_config.py: add sensors: DualRudderSensorConfig field
- arautopilot/studio/compiler/appack.py: .appack compiler (ZIP with
  manifest.json + project.yaml + firmware_config.h + install_notes.txt)
- arautopilot/studio/editors/project_editor.py: full project config editor
  (vessel / actuator / sensors / PID, RBAC-gated fields, save/load .yaml/.json)
- arautopilot/studio/main_window.py: wire ProjectEditorWidget into Project tab
- tests: test_sensor_config.py (11 tests) + test_appack_compiler.py (10 tests)

Firmware:
- hal/rudder_actuator_iface.h: IRudderActuator abstract interface
- hal/rudder_actuator_hydraulic.h: reversible hydraulic pump (LEDC PWM)
- hal/rudder_actuator_electric.h: reversible DC motor + deadband compensation
- hal/rudder_actuator_factory.h: build-time type selection via AR_ACTUATOR_TYPE

Tests: 483 passed (was 462, +21 Sprint 4)

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
2026-05-20 10:29:17 -04:00
parent a2f3e82f17
commit e4812e9b44
11 changed files with 1661 additions and 5 deletions
+110
View File
@@ -0,0 +1,110 @@
"""Rudder angle sensor configuration.
Supports single and dual (redundant) sensor setups. The primary sensor
is mandatory for closed-loop operation. The redundant sensor is optional
but strongly recommended for Phase-1 marine deployments.
Sensor families supported in Phase 1:
- ``AS5048A_SPI`` — contactless magnetic absolute encoder, 14-bit, SPI.
Recommended for new installations (brief session 2026-05-18).
- ``POTENTIOMETER`` — resistive potentiometer 0-10 V or 4-20 mA.
Supported for legacy actuators.
Both sensors use a polycarbonate linkage arm to the rudder stock, giving
a proportional rotation to the rudder angle.
"""
from __future__ import annotations
from enum import StrEnum
from pydantic import BaseModel, ConfigDict, Field
class RudderSensorType(StrEnum):
AS5048A_SPI = "as5048a_spi"
"""AMS AS5048A contactless magnetic encoder — 14-bit, SPI interface.
No mechanical contact; requires diametrically magnetised 6 mm magnet
on the rotating shaft. Recommended for all new Phase-1 installations."""
POTENTIOMETER = "potentiometer"
"""Resistive potentiometer with 0-10 V or 4-20 mA signal conditioning.
Suitable for retrofit on existing actuators that already carry a pot."""
class RudderSensorConfig(BaseModel):
"""Configuration for one physical rudder angle sensor.
The ``full_scale_deg`` / ``zero_offset_deg`` pair maps the electrical
range of the sensor to real rudder degrees. Calibration during
commissioning (Sprint 7) will update these values in NVS.
"""
model_config = ConfigDict(extra="forbid", validate_assignment=True)
type: RudderSensorType
"""Physical sensor technology."""
label: str = Field(
default="",
max_length=60,
description="Free-form label, e.g. 'Primary rudder stock' or 'Redundant actuator arm'.",
)
# SPI-specific (ignored for POTENTIOMETER)
spi_cs_gpio: int = Field(
default=10,
ge=0,
le=39,
description="ESP32 GPIO pin used as SPI chip-select for AS5048A.",
)
# Calibration
full_scale_deg: float = Field(
default=35.0,
gt=0.0,
le=45.0,
description="Rudder angle (degrees) corresponding to full electrical output.",
)
zero_offset_deg: float = Field(
default=0.0,
ge=-10.0,
le=10.0,
description="Offset applied after scaling to correct mechanical amidships misalignment.",
)
# Cross-validation (used by dual-sensor arbitrator)
divergence_alarm_deg: float = Field(
default=3.0,
gt=0.0,
le=15.0,
description="Alarm threshold: raise SENSOR_DIVERGE if |A - B| exceeds this value.",
)
divergence_failover_deg: float = Field(
default=6.0,
gt=0.0,
le=20.0,
description="Failover threshold: switch to the other sensor if |A - B| exceeds this.",
)
class DualRudderSensorConfig(BaseModel):
"""Primary + optional redundant rudder sensor pair.
When ``redundant`` is ``None`` the system operates in single-sensor
mode (feedback_required still enforced). When ``redundant`` is
present the firmware enables cross-validation and automatic failover.
Divergence thresholds are taken from the *primary* sensor's config so
there is one canonical place to tune them.
"""
model_config = ConfigDict(extra="forbid", validate_assignment=True)
primary: RudderSensorConfig
redundant: RudderSensorConfig | None = None
@property
def has_redundancy(self) -> bool:
return self.redundant is not None
+11
View File
@@ -13,6 +13,7 @@ from pydantic import BaseModel, ConfigDict, Field
from arautopilot.core.actuator_config import ActuatorConfig from arautopilot.core.actuator_config import ActuatorConfig
from arautopilot.core.ids import VesselId, new_vessel_id from arautopilot.core.ids import VesselId, new_vessel_id
from arautopilot.core.pid_config import PidConfig from arautopilot.core.pid_config import PidConfig
from arautopilot.core.sensor_config import DualRudderSensorConfig, RudderSensorConfig, RudderSensorType
class VesselType(StrEnum): class VesselType(StrEnum):
@@ -60,3 +61,13 @@ class VesselConfig(BaseModel):
actuator: ActuatorConfig actuator: ActuatorConfig
pid: PidConfig pid: PidConfig
sensors: DualRudderSensorConfig = Field(
default_factory=lambda: DualRudderSensorConfig(
primary=RudderSensorConfig(
type=RudderSensorType.AS5048A_SPI,
label="Primary rudder stock",
spi_cs_gpio=10,
)
),
description="Rudder angle sensor(s). Dual config enables cross-validation and failover.",
)
+245
View File
@@ -0,0 +1,245 @@
"""AR-Autopilot deployment package compiler — Sprint 4.
An ``.appack`` is a ZIP archive that contains everything needed to deploy
a configured autopilot to a vessel:
<project_id>/
├── manifest.json — metadata, checksums, schema version
├── project.yaml — full ProjectConfig (human-readable)
├── firmware_config.h — generated C header consumed by the firmware build
└── install_notes.txt — human summary for the field engineer
The format is intentionally transparent (plain ZIP, readable YAML/JSON/C)
so that a field engineer can inspect the package contents without special tools.
Typical workflow
----------------
1. Integrator configures a ``ProjectConfig`` in the Studio (Project tab).
2. Clicks "Compile .appack…" → this compiler runs.
3. Studio writes the .appack to disk.
4. Field engineer transfers the .appack to the vessel.
5. Future Sprint: Studio "Deploy" tab extracts the package, calls PlatformIO
with the generated header, and flashes the board.
"""
from __future__ import annotations
import hashlib
import json
import zipfile
from datetime import UTC, datetime
from pathlib import Path
from textwrap import dedent
from arautopilot.core.actuator_config import ActuatorType
from arautopilot.core.project_config import ProjectConfig
from arautopilot.core.sensor_config import DualRudderSensorConfig, RudderSensorType
from arautopilot.version import __version__
_SCHEMA_VERSION = "0.1.0"
class AppackCompiler:
"""Compiles a ``ProjectConfig`` into an ``.appack`` deployment archive."""
def __init__(self, project: ProjectConfig) -> None:
if not project.client_name.strip():
raise ValueError("project.client_name must not be empty")
if not project.project_name.strip():
raise ValueError("project.project_name must not be empty")
self._project = project
def compile(self, output_path: Path | str) -> Path:
"""Build the ``.appack`` archive and write it to *output_path*.
Returns the resolved output path.
"""
out = Path(output_path)
out.parent.mkdir(parents=True, exist_ok=True)
project_yaml = self._project.to_yaml()
firmware_header = self._generate_firmware_header()
install_notes = self._generate_install_notes()
manifest = self._build_manifest(project_yaml, firmware_header)
prefix = str(self._project.project_id)
with zipfile.ZipFile(out, "w", compression=zipfile.ZIP_DEFLATED) as zf:
zf.writestr(f"{prefix}/manifest.json", json.dumps(manifest, indent=2))
zf.writestr(f"{prefix}/project.yaml", project_yaml)
zf.writestr(f"{prefix}/firmware_config.h", firmware_header)
zf.writestr(f"{prefix}/install_notes.txt", install_notes)
return out
# ------------------------------------------------------------------
# Manifest
# ------------------------------------------------------------------
def _build_manifest(self, project_yaml: str, firmware_header: str) -> dict:
return {
"appack_schema": _SCHEMA_VERSION,
"arautopilot_version": __version__,
"compiled_at": datetime.now(UTC).isoformat(),
"project_id": str(self._project.project_id),
"client_name": self._project.client_name,
"project_name": self._project.project_name,
"vessel_name": self._project.vessel.name,
"vessel_type": self._project.vessel.type.value,
"actuator_type": self._project.vessel.actuator.type.value,
"checksums": {
"project_yaml_sha256": _sha256(project_yaml),
"firmware_config_h_sha256": _sha256(firmware_header),
},
}
# ------------------------------------------------------------------
# Firmware header generator
# ------------------------------------------------------------------
def _generate_firmware_header(self) -> str:
p = self._project
v = p.vessel
a = v.actuator
pid = v.pid
sensors = v.sensors
actuator_enum = _actuator_type_to_enum(a.type)
primary_sensor_enum = _sensor_type_to_enum(sensors.primary.type)
redundant_enabled = "1" if sensors.has_redundancy else "0"
redundant_cs = sensors.redundant.spi_cs_gpio if sensors.redundant else 0
redundant_sensor_enum = (
_sensor_type_to_enum(sensors.redundant.type)
if sensors.redundant
else "RUDDER_SENSOR_NONE"
)
diverge_alarm = sensors.primary.divergence_alarm_deg
diverge_failover = sensors.primary.divergence_failover_deg
lines = [
"// AUTO-GENERATED by AR-Autopilot Studio — DO NOT EDIT BY HAND.",
f"// Project : {p.project_name}",
f"// Client : {p.client_name}",
f"// Vessel : {v.name} ({v.type.value})",
f"// Generated: {datetime.now(UTC).isoformat()}",
"//",
"#pragma once",
"",
"// --- Project identity -------------------------------------------------",
f'#define AR_PROJECT_ID "{p.project_id}"',
f'#define AR_CLIENT_NAME "{p.client_name}"',
f'#define AR_PROJECT_NAME "{p.project_name}"',
f'#define AR_VESSEL_NAME "{v.name}"',
"",
"// --- Vessel kinematics ------------------------------------------------",
f"#define AR_VESSEL_LENGTH_M {v.length_m:.1f}f",
f"#define AR_VESSEL_MAX_SPEED_KN {v.max_speed_kn:.1f}f",
"",
"// --- Actuator ---------------------------------------------------------",
f"#define AR_ACTUATOR_TYPE {actuator_enum}",
f"#define AR_RUDDER_ANGLE_LIMIT_DEG {a.max_rudder_angle_deg:.1f}f",
f"#define AR_ACTUATOR_DEADBAND_PCT {a.deadband_pct:.1f}f",
f"#define AR_ACTUATOR_MAX_RATE_DPS {a.max_rate_dps:.1f}f",
f"#define AR_ACTUATOR_MAX_CURRENT_A {a.max_current_a:.1f}f",
f"#define AR_ACTUATOR_ASYM_STBD {a.asymmetry_stbd_over_port:.3f}f",
"",
"// --- Rudder sensors ---------------------------------------------------",
f"#define AR_SENSOR_PRIMARY_TYPE {primary_sensor_enum}",
f"#define AR_SENSOR_PRIMARY_CS_GPIO {sensors.primary.spi_cs_gpio}",
f"#define AR_SENSOR_PRIMARY_FSD_DEG {sensors.primary.full_scale_deg:.1f}f",
f"#define AR_SENSOR_REDUNDANT {redundant_enabled}",
f"#define AR_SENSOR_REDUNDANT_TYPE {redundant_sensor_enum}",
f"#define AR_SENSOR_REDUNDANT_CS_GPIO {redundant_cs}",
f"#define AR_SENSOR_DIVERGE_ALARM_DEG {diverge_alarm:.1f}f",
f"#define AR_SENSOR_DIVERGE_FAILOVER_DEG {diverge_failover:.1f}f",
"",
"// --- PID base gains (integrator IP — not exposed to operator) ---------",
f"#define AR_PID_INNER_KP {pid.inner_loop_base.kp:.4f}f",
f"#define AR_PID_INNER_KI {pid.inner_loop_base.ki:.4f}f",
f"#define AR_PID_INNER_KD {pid.inner_loop_base.kd:.4f}f",
f"#define AR_PID_OUTER_KP {pid.outer_loop_base.kp:.4f}f",
f"#define AR_PID_OUTER_KI {pid.outer_loop_base.ki:.4f}f",
f"#define AR_PID_OUTER_KD {pid.outer_loop_base.kd:.4f}f",
f"#define AR_PID_ROT_FF {pid.rot_feedforward_gain:.4f}f",
f"#define AR_PID_DEADBAND_DEG {pid.setpoint_deadband_deg:.2f}f",
f"#define AR_PID_RATE_LIMIT_DPS {pid.setpoint_rate_limit_dps:.2f}f",
f"#define AR_PID_ANTI_WINDUP {pid.anti_windup_limit:.2f}f",
"",
"// End of generated header.",
]
return "\n".join(lines) + "\n"
# ------------------------------------------------------------------
# Install notes
# ------------------------------------------------------------------
def _generate_install_notes(self) -> str:
p = self._project
v = p.vessel
a = v.actuator
s = v.sensors
return dedent(f"""\
AR-Autopilot — Field Installation Notes
========================================
Project : {p.project_name}
Client : {p.client_name}
Vessel : {v.name} ({v.type.value.replace("_", " ").title()})
Generated: {datetime.now(UTC).strftime("%Y-%m-%d %H:%M UTC")}
ACTUATOR
--------
Type : {a.type.value.replace("_", " ").title()}
Model / label : {a.name or "(not specified)"}
Rudder limit : ±{a.max_rudder_angle_deg:.1f}° (software limit — verify ≤ mechanical stop)
Deadband : {a.deadband_pct:.1f} %
Max slew rate : {a.max_rate_dps:.1f} °/s
Overcurrent : {a.max_current_a:.1f} A trip
SENSORS
-------
Primary sensor : {s.primary.type.value.upper()} GPIO CS={s.primary.spi_cs_gpio}
Full-scale : ±{s.primary.full_scale_deg:.1f}°
Redundant sensor: {"ENABLED" if s.has_redundancy else "DISABLED"}
{"Redundant type : " + s.redundant.type.value.upper() + " GPIO CS=" + str(s.redundant.spi_cs_gpio) if s.redundant else ""}
Divergence alarm : {s.primary.divergence_alarm_deg:.1f}°
Divergence failover : {s.primary.divergence_failover_deg:.1f}°
COMMISSIONING STEPS (Sprint 7 wizard will automate these)
-----------------------------------------------------------
1. Flash firmware with this .appack (Flash Console or pio upload).
2. Connect primary rudder sensor to SPI CS GPIO {s.primary.spi_cs_gpio}.
{"3. Connect redundant sensor to SPI CS GPIO " + str(s.redundant.spi_cs_gpio) + "." if s.redundant else "3. (Redundant sensor not configured.)"}
4. Verify rudder moves to ±{a.max_rudder_angle_deg:.1f}° and stops.
5. Run Modbus client test: python tools/modbus_client_test.py
6. Engage HEADING_HOLD and confirm heading capture.
7. Tune PID gains via Modbus if required.
SUPPORT
-------
Contact: AR Suite — alro65@gmail.com
""")
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
def _sha256(text: str) -> str:
return hashlib.sha256(text.encode()).hexdigest()
def _actuator_type_to_enum(t: ActuatorType) -> str:
return {
ActuatorType.HYDRAULIC_REVERSIBLE: "ACTUATOR_HYDRAULIC_REVERSIBLE",
ActuatorType.ELECTRIC_DC_REVERSIBLE: "ACTUATOR_ELECTRIC_DC",
ActuatorType.SERVOMOTOR_FEEDBACK: "ACTUATOR_SERVOMOTOR",
ActuatorType.STERNDRIVE_ANALOG: "ACTUATOR_STERNDRIVE_ANALOG",
}.get(t, "ACTUATOR_UNKNOWN")
def _sensor_type_to_enum(t: RudderSensorType) -> str:
return {
RudderSensorType.AS5048A_SPI: "RUDDER_SENSOR_AS5048A_SPI",
RudderSensorType.POTENTIOMETER: "RUDDER_SENSOR_POTENTIOMETER",
}.get(t, "RUDDER_SENSOR_UNKNOWN")
@@ -0,0 +1,619 @@
"""Project configurator widget — Sprint 4.
A scrollable form that lets the integrator (Engineer / Super Admin) create
or edit a ``ProjectConfig`` and compile it into an ``.appack`` deployment
package.
Layout
------
┌─────────────────────────────────────────────────────┐
│ PROJECT INFO client / project name / notes │
│ VESSEL name / type / dimensions │
│ ACTUATOR type / angle limit / deadband / … │
│ SENSORS primary + optional redundant │
│ PID GAINS inner + outer base (SA-gated) │
│ ─────────────────────────────────────────────────── │
│ [New] [Open…] [Save] [Compile .appack…] │
└─────────────────────────────────────────────────────┘
RBAC gates
----------
- ``EDIT_COMMISSIONING`` (Engineer+) : actuator + sensor + rudder limit.
- ``EDIT_BASE_GAINS`` (Super Admin) : PID base gains fields.
- ``EDIT_COMMISSIONING`` (Engineer+) : Compile .appack button.
- Any role can load and read a project; only the above roles may edit.
"""
from __future__ import annotations
from pathlib import Path
from PySide6.QtWidgets import (
QCheckBox,
QComboBox,
QDoubleSpinBox,
QFileDialog,
QFormLayout,
QGroupBox,
QHBoxLayout,
QLabel,
QLineEdit,
QMessageBox,
QPushButton,
QScrollArea,
QSizePolicy,
QSpinBox,
QTextEdit,
QVBoxLayout,
QWidget,
)
from arautopilot.core.actuator_config import ActuatorConfig, ActuatorType
from arautopilot.core.pid_config import PidConfig, PidGains
from arautopilot.core.project_config import ProjectConfig
from arautopilot.core.rbac import Capability
from arautopilot.core.sensor_config import (
DualRudderSensorConfig,
RudderSensorConfig,
RudderSensorType,
)
from arautopilot.core.vessel_config import VesselConfig, VesselType
from arautopilot.studio.session import Session
class ProjectEditorWidget(QWidget):
"""Full project config editor — fills the 'Project' tab in the Studio."""
def __init__(self, session: Session, parent: QWidget | None = None) -> None:
super().__init__(parent)
self._session = session
self._project: ProjectConfig | None = None
root = QVBoxLayout(self)
root.setContentsMargins(0, 0, 0, 0)
# Scrollable form area
scroll = QScrollArea()
scroll.setWidgetResizable(True)
content = QWidget()
form_layout = QVBoxLayout(content)
form_layout.setContentsMargins(12, 12, 12, 12)
form_layout.setSpacing(12)
self._build_project_info_group(form_layout)
self._build_vessel_group(form_layout)
self._build_actuator_group(form_layout)
self._build_sensor_group(form_layout)
self._build_pid_group(form_layout)
form_layout.addStretch(1)
scroll.setWidget(content)
root.addWidget(scroll, stretch=1)
root.addWidget(self._build_toolbar())
self._apply_rbac()
self._load_defaults()
# ------------------------------------------------------------------
# Group builders
# ------------------------------------------------------------------
def _build_project_info_group(self, parent: QVBoxLayout) -> None:
box = QGroupBox("Project Info")
fl = QFormLayout(box)
self._client_name = QLineEdit()
self._client_name.setPlaceholderText("e.g. Astilleros del Sur S.A.")
fl.addRow("Client name *", self._client_name)
self._project_name = QLineEdit()
self._project_name.setPlaceholderText("e.g. M/Y Pacifica Autopilot install")
fl.addRow("Project name *", self._project_name)
self._notes = QTextEdit()
self._notes.setPlaceholderText("Optional notes…")
self._notes.setFixedHeight(72)
fl.addRow("Notes", self._notes)
parent.addWidget(box)
def _build_vessel_group(self, parent: QVBoxLayout) -> None:
box = QGroupBox("Vessel")
fl = QFormLayout(box)
self._vessel_name = QLineEdit()
self._vessel_name.setPlaceholderText("e.g. Pacifica")
fl.addRow("Vessel name *", self._vessel_name)
self._vessel_type = QComboBox()
for vt in VesselType:
self._vessel_type.addItem(vt.value.replace("_", " ").title(), vt)
fl.addRow("Type", self._vessel_type)
self._length_m = QDoubleSpinBox()
self._length_m.setRange(5.0, 200.0)
self._length_m.setSuffix(" m")
self._length_m.setDecimals(1)
self._length_m.setValue(30.0)
fl.addRow("Length overall", self._length_m)
self._displacement_t = QDoubleSpinBox()
self._displacement_t.setRange(0.0, 10_000.0)
self._displacement_t.setSuffix(" t")
self._displacement_t.setDecimals(1)
fl.addRow("Displacement (0 = unknown)", self._displacement_t)
self._max_speed_kn = QDoubleSpinBox()
self._max_speed_kn.setRange(1.0, 80.0)
self._max_speed_kn.setSuffix(" kn")
self._max_speed_kn.setDecimals(1)
self._max_speed_kn.setValue(18.0)
fl.addRow("Max speed", self._max_speed_kn)
parent.addWidget(box)
def _build_actuator_group(self, parent: QVBoxLayout) -> None:
box = QGroupBox("Rudder Actuator")
fl = QFormLayout(box)
self._act_type = QComboBox()
phase1 = [
ActuatorType.HYDRAULIC_REVERSIBLE,
ActuatorType.ELECTRIC_DC_REVERSIBLE,
ActuatorType.SERVOMOTOR_FEEDBACK,
ActuatorType.STERNDRIVE_ANALOG,
]
for at in phase1:
self._act_type.addItem(at.value.replace("_", " ").title(), at)
fl.addRow("Type", self._act_type)
self._act_name = QLineEdit()
self._act_name.setPlaceholderText("e.g. Hynautic K-21 + Capilano cylinder")
fl.addRow("Model / label", self._act_name)
self._max_rudder_deg = QDoubleSpinBox()
self._max_rudder_deg.setRange(15.0, 45.0)
self._max_rudder_deg.setSuffix(" °")
self._max_rudder_deg.setDecimals(1)
self._max_rudder_deg.setValue(35.0)
self._max_rudder_deg.setToolTip(
"Software rudder angle limit. Must not exceed the mechanical stop.\n"
"Requires Engineer or Super Admin to change."
)
fl.addRow("Rudder angle limit ⚠", self._max_rudder_deg)
self._deadband_pct = QDoubleSpinBox()
self._deadband_pct.setRange(0.0, 30.0)
self._deadband_pct.setSuffix(" %")
self._deadband_pct.setDecimals(1)
self._deadband_pct.setValue(5.0)
fl.addRow("Deadband", self._deadband_pct)
self._max_rate_dps = QDoubleSpinBox()
self._max_rate_dps.setRange(0.5, 15.0)
self._max_rate_dps.setSuffix(" °/s")
self._max_rate_dps.setDecimals(1)
self._max_rate_dps.setValue(4.5)
fl.addRow("Max slew rate", self._max_rate_dps)
self._max_current_a = QDoubleSpinBox()
self._max_current_a.setRange(1.0, 200.0)
self._max_current_a.setSuffix(" A")
self._max_current_a.setDecimals(1)
self._max_current_a.setValue(15.0)
fl.addRow("Overcurrent limit", self._max_current_a)
parent.addWidget(box)
self._actuator_box = box
def _build_sensor_group(self, parent: QVBoxLayout) -> None:
box = QGroupBox("Rudder Angle Sensors")
fl = QFormLayout(box)
fl.addRow(QLabel("<b>Primary sensor</b>"))
self._sens_primary_type = QComboBox()
for st in RudderSensorType:
self._sens_primary_type.addItem(st.value.replace("_", " ").upper(), st)
fl.addRow("Type", self._sens_primary_type)
self._sens_primary_cs = QSpinBox()
self._sens_primary_cs.setRange(0, 39)
self._sens_primary_cs.setValue(10)
self._sens_primary_cs.setToolTip("ESP32 GPIO used as SPI CS for AS5048A")
fl.addRow("SPI CS GPIO", self._sens_primary_cs)
self._sens_primary_fsd = QDoubleSpinBox()
self._sens_primary_fsd.setRange(1.0, 45.0)
self._sens_primary_fsd.setSuffix(" °")
self._sens_primary_fsd.setDecimals(1)
self._sens_primary_fsd.setValue(35.0)
fl.addRow("Full-scale angle", self._sens_primary_fsd)
fl.addRow(QLabel(""))
fl.addRow(QLabel("<b>Redundant sensor (optional)</b>"))
self._sens_redundant_enabled = QCheckBox("Enable redundant sensor")
fl.addRow("", self._sens_redundant_enabled)
self._sens_redundant_type = QComboBox()
for st in RudderSensorType:
self._sens_redundant_type.addItem(st.value.replace("_", " ").upper(), st)
fl.addRow("Type", self._sens_redundant_type)
self._sens_redundant_cs = QSpinBox()
self._sens_redundant_cs.setRange(0, 39)
self._sens_redundant_cs.setValue(11)
fl.addRow("SPI CS GPIO", self._sens_redundant_cs)
self._sens_redundant_fsd = QDoubleSpinBox()
self._sens_redundant_fsd.setRange(1.0, 45.0)
self._sens_redundant_fsd.setSuffix(" °")
self._sens_redundant_fsd.setDecimals(1)
self._sens_redundant_fsd.setValue(35.0)
fl.addRow("Full-scale angle", self._sens_redundant_fsd)
self._sens_diverge_alarm = QDoubleSpinBox()
self._sens_diverge_alarm.setRange(0.5, 15.0)
self._sens_diverge_alarm.setSuffix(" °")
self._sens_diverge_alarm.setDecimals(1)
self._sens_diverge_alarm.setValue(3.0)
fl.addRow("Divergence alarm threshold", self._sens_diverge_alarm)
self._sens_diverge_failover = QDoubleSpinBox()
self._sens_diverge_failover.setRange(1.0, 20.0)
self._sens_diverge_failover.setSuffix(" °")
self._sens_diverge_failover.setDecimals(1)
self._sens_diverge_failover.setValue(6.0)
fl.addRow("Divergence failover threshold", self._sens_diverge_failover)
self._sens_redundant_enabled.toggled.connect(self._on_redundant_toggled)
self._on_redundant_toggled(False)
parent.addWidget(box)
self._sensor_box = box
def _build_pid_group(self, parent: QVBoxLayout) -> None:
box = QGroupBox("PID Base Gains [Super Admin only]")
fl = QFormLayout(box)
fl.addRow(QLabel("<b>Inner loop (rudder position, 50 Hz)</b>"))
self._inner_kp = self._gain_spin("Inner Kp", fl)
self._inner_ki = self._gain_spin("Inner Ki", fl)
self._inner_kd = self._gain_spin("Inner Kd", fl)
fl.addRow(QLabel(""))
fl.addRow(QLabel("<b>Outer loop (heading, 10 Hz)</b>"))
self._outer_kp = self._gain_spin("Outer Kp", fl)
self._outer_ki = self._gain_spin("Outer Ki", fl)
self._outer_kd = self._gain_spin("Outer Kd", fl)
fl.addRow(QLabel(""))
self._rot_ff = self._gain_spin("ROT feed-forward", fl)
parent.addWidget(box)
self._pid_box = box
# ------------------------------------------------------------------
# Toolbar
# ------------------------------------------------------------------
def _build_toolbar(self) -> QWidget:
bar = QWidget()
bar.setSizePolicy(QSizePolicy.Policy.Preferred, QSizePolicy.Policy.Fixed)
h = QHBoxLayout(bar)
h.setContentsMargins(8, 4, 8, 4)
btn_new = QPushButton("New")
btn_new.clicked.connect(self._on_new)
h.addWidget(btn_new)
btn_open = QPushButton("Open…")
btn_open.clicked.connect(self._on_open)
h.addWidget(btn_open)
btn_save = QPushButton("Save")
btn_save.clicked.connect(self._on_save)
h.addWidget(btn_save)
h.addStretch(1)
self._btn_compile = QPushButton("Compile .appack…")
self._btn_compile.setToolTip("Generate deployment package (Engineer+)")
self._btn_compile.clicked.connect(self._on_compile)
h.addWidget(self._btn_compile)
return bar
# ------------------------------------------------------------------
# RBAC
# ------------------------------------------------------------------
def _apply_rbac(self) -> None:
can_commission = self._session.has(Capability.EDIT_COMMISSIONING)
can_gains = self._session.has(Capability.EDIT_BASE_GAINS)
can_compile = self._session.has(Capability.EDIT_COMMISSIONING)
for w in [
self._max_rudder_deg,
self._deadband_pct,
self._max_rate_dps,
self._max_current_a,
self._act_type,
self._act_name,
self._sens_primary_type,
self._sens_primary_cs,
self._sens_primary_fsd,
self._sens_redundant_enabled,
self._sens_redundant_type,
self._sens_redundant_cs,
self._sens_redundant_fsd,
self._sens_diverge_alarm,
self._sens_diverge_failover,
]:
w.setEnabled(can_commission)
for w in [
self._inner_kp,
self._inner_ki,
self._inner_kd,
self._outer_kp,
self._outer_ki,
self._outer_kd,
self._rot_ff,
]:
w.setEnabled(can_gains)
self._btn_compile.setEnabled(can_compile)
if not can_gains:
self._pid_box.setTitle("PID Base Gains [Super Admin only — read-only]")
# ------------------------------------------------------------------
# Slots
# ------------------------------------------------------------------
def _on_redundant_toggled(self, enabled: bool) -> None:
for w in [
self._sens_redundant_type,
self._sens_redundant_cs,
self._sens_redundant_fsd,
self._sens_diverge_alarm,
self._sens_diverge_failover,
]:
w.setEnabled(enabled and self._session.has(Capability.EDIT_COMMISSIONING))
def _on_new(self) -> None:
self._project = None
self._load_defaults()
def _on_open(self) -> None:
path, _ = QFileDialog.getOpenFileName(
self,
"Open project",
str(Path.home()),
"AR-Autopilot Project (*.yaml *.yml *.json);;All files (*)",
)
if not path:
return
try:
self._project = ProjectConfig.load(path)
self._populate_from_project(self._project)
except Exception as exc: # noqa: BLE001
QMessageBox.critical(self, "Load error", str(exc))
def _on_save(self) -> None:
try:
project = self._build_project()
except Exception as exc: # noqa: BLE001
QMessageBox.critical(self, "Validation error", str(exc))
return
path, _ = QFileDialog.getSaveFileName(
self,
"Save project",
str(Path.home() / f"{project.project_name}.yaml"),
"AR-Autopilot Project (*.yaml);;JSON (*.json)",
)
if not path:
return
try:
project.save_yaml(path) if path.endswith((".yaml", ".yml")) else project.save_json(path)
self._project = project
QMessageBox.information(self, "Saved", f"Project saved to:\n{path}")
except Exception as exc: # noqa: BLE001
QMessageBox.critical(self, "Save error", str(exc))
def _on_compile(self) -> None:
if not self._session.has(Capability.EDIT_COMMISSIONING):
QMessageBox.warning(self, "Access denied", "Engineer or Super Admin required.")
return
try:
project = self._build_project()
except Exception as exc: # noqa: BLE001
QMessageBox.critical(self, "Validation error", str(exc))
return
path, _ = QFileDialog.getSaveFileName(
self,
"Save .appack",
str(Path.home() / f"{project.project_name}.appack"),
"AR-Autopilot Package (*.appack)",
)
if not path:
return
try:
from arautopilot.studio.compiler.appack import AppackCompiler
compiler = AppackCompiler(project)
out = compiler.compile(Path(path))
self._session.audit.record(
actor=self._session.user.username,
action="compile_appack",
detail=f"project={project.project_id} output={out}",
)
QMessageBox.information(
self, "Compiled", f".appack written to:\n{out}"
)
except Exception as exc: # noqa: BLE001
QMessageBox.critical(self, "Compile error", str(exc))
# ------------------------------------------------------------------
# Build / populate helpers
# ------------------------------------------------------------------
def _build_project(self) -> ProjectConfig:
redundant: RudderSensorConfig | None = None
if self._sens_redundant_enabled.isChecked():
redundant = RudderSensorConfig(
type=self._sens_redundant_type.currentData(),
label="Redundant actuator arm",
spi_cs_gpio=self._sens_redundant_cs.value(),
full_scale_deg=self._sens_redundant_fsd.value(),
divergence_alarm_deg=self._sens_diverge_alarm.value(),
divergence_failover_deg=self._sens_diverge_failover.value(),
)
vessel = VesselConfig(
name=self._vessel_name.text().strip() or "Unnamed vessel",
type=self._vessel_type.currentData(),
length_m=self._length_m.value(),
displacement_t=self._displacement_t.value(),
max_speed_kn=self._max_speed_kn.value(),
actuator=ActuatorConfig(
type=self._act_type.currentData(),
name=self._act_name.text().strip(),
max_rudder_angle_deg=self._max_rudder_deg.value(),
deadband_pct=self._deadband_pct.value(),
max_rate_dps=self._max_rate_dps.value(),
max_current_a=self._max_current_a.value(),
),
pid=PidConfig(
inner_loop_base=PidGains(
kp=self._inner_kp.value(),
ki=self._inner_ki.value(),
kd=self._inner_kd.value(),
),
outer_loop_base=PidGains(
kp=self._outer_kp.value(),
ki=self._outer_ki.value(),
kd=self._outer_kd.value(),
),
rot_feedforward_gain=self._rot_ff.value(),
),
sensors=DualRudderSensorConfig(
primary=RudderSensorConfig(
type=self._sens_primary_type.currentData(),
label="Primary rudder stock",
spi_cs_gpio=self._sens_primary_cs.value(),
full_scale_deg=self._sens_primary_fsd.value(),
),
redundant=redundant,
),
)
existing_id = self._project.project_id if self._project else None
project = ProjectConfig(
client_name=self._client_name.text().strip() or "Unknown client",
project_name=self._project_name.text().strip() or "Unnamed project",
notes=self._notes.toPlainText().strip(),
vessel=vessel,
)
if existing_id is not None:
object.__setattr__(project, "project_id", existing_id)
return project
def _populate_from_project(self, p: ProjectConfig) -> None:
self._client_name.setText(p.client_name)
self._project_name.setText(p.project_name)
self._notes.setPlainText(p.notes)
v = p.vessel
self._vessel_name.setText(v.name)
_set_combo(self._vessel_type, v.type)
self._length_m.setValue(v.length_m)
self._displacement_t.setValue(v.displacement_t)
self._max_speed_kn.setValue(v.max_speed_kn)
a = v.actuator
_set_combo(self._act_type, a.type)
self._act_name.setText(a.name)
self._max_rudder_deg.setValue(a.max_rudder_angle_deg)
self._deadband_pct.setValue(a.deadband_pct)
self._max_rate_dps.setValue(a.max_rate_dps)
self._max_current_a.setValue(a.max_current_a)
sp = v.sensors.primary
_set_combo(self._sens_primary_type, sp.type)
self._sens_primary_cs.setValue(sp.spi_cs_gpio)
self._sens_primary_fsd.setValue(sp.full_scale_deg)
sr = v.sensors.redundant
self._sens_redundant_enabled.setChecked(sr is not None)
if sr is not None:
_set_combo(self._sens_redundant_type, sr.type)
self._sens_redundant_cs.setValue(sr.spi_cs_gpio)
self._sens_redundant_fsd.setValue(sr.full_scale_deg)
self._sens_diverge_alarm.setValue(sr.divergence_alarm_deg)
self._sens_diverge_failover.setValue(sr.divergence_failover_deg)
pid = v.pid
self._inner_kp.setValue(pid.inner_loop_base.kp)
self._inner_ki.setValue(pid.inner_loop_base.ki)
self._inner_kd.setValue(pid.inner_loop_base.kd)
self._outer_kp.setValue(pid.outer_loop_base.kp)
self._outer_ki.setValue(pid.outer_loop_base.ki)
self._outer_kd.setValue(pid.outer_loop_base.kd)
self._rot_ff.setValue(pid.rot_feedforward_gain)
def _load_defaults(self) -> None:
self._client_name.clear()
self._project_name.clear()
self._notes.clear()
self._vessel_name.clear()
self._vessel_type.setCurrentIndex(0)
self._length_m.setValue(30.0)
self._displacement_t.setValue(0.0)
self._max_speed_kn.setValue(18.0)
self._act_type.setCurrentIndex(0)
self._act_name.clear()
self._max_rudder_deg.setValue(35.0)
self._deadband_pct.setValue(5.0)
self._max_rate_dps.setValue(4.5)
self._max_current_a.setValue(15.0)
self._sens_primary_type.setCurrentIndex(0)
self._sens_primary_cs.setValue(10)
self._sens_primary_fsd.setValue(35.0)
self._sens_redundant_enabled.setChecked(False)
self._sens_redundant_cs.setValue(11)
self._sens_redundant_fsd.setValue(35.0)
self._sens_diverge_alarm.setValue(3.0)
self._sens_diverge_failover.setValue(6.0)
self._inner_kp.setValue(1.5)
self._inner_ki.setValue(0.1)
self._inner_kd.setValue(0.05)
self._outer_kp.setValue(2.0)
self._outer_ki.setValue(0.05)
self._outer_kd.setValue(0.3)
self._rot_ff.setValue(0.4)
# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------
@staticmethod
def _gain_spin(label: str, fl: QFormLayout) -> QDoubleSpinBox:
spin = QDoubleSpinBox()
spin.setRange(0.0, 100.0)
spin.setDecimals(4)
spin.setSingleStep(0.01)
fl.addRow(label, spin)
return spin
def _set_combo(combo: QComboBox, value: object) -> None:
for i in range(combo.count()):
if combo.itemData(i) == value:
combo.setCurrentIndex(i)
return
+2 -5
View File
@@ -24,6 +24,7 @@ from PySide6.QtWidgets import (
) )
from arautopilot.core.rbac import capabilities_of from arautopilot.core.rbac import capabilities_of
from arautopilot.studio.editors.project_editor import ProjectEditorWidget
from arautopilot.studio.flash_console import FlashConsoleWidget from arautopilot.studio.flash_console import FlashConsoleWidget
from arautopilot.studio.session import Session from arautopilot.studio.session import Session
from arautopilot.version import __version__ from arautopilot.version import __version__
@@ -74,11 +75,7 @@ class StudioMainWindow(QMainWindow):
tabs.addTab(self._build_overview_tab(), "Overview") tabs.addTab(self._build_overview_tab(), "Overview")
tabs.addTab(FlashConsoleWidget(self._session), "Flash Console") tabs.addTab(FlashConsoleWidget(self._session), "Flash Console")
tabs.addTab(self._placeholder_tab( tabs.addTab(ProjectEditorWidget(self._session), "Project")
"Project configurator -- Sprint 4.\n\n"
"Will let you create / edit a per-vessel ProjectConfig and "
"compile it into an .appack for deployment."
), "Project")
tabs.addTab(self._placeholder_tab( tabs.addTab(self._placeholder_tab(
"Telemetry -- Sprint 4.\n\n" "Telemetry -- Sprint 4.\n\n"
"Live Modbus telemetry from the connected AR-NMEA-IO board." "Live Modbus telemetry from the connected AR-NMEA-IO board."
+154
View File
@@ -0,0 +1,154 @@
"""Tests for AppackCompiler — Sprint 4."""
from __future__ import annotations
import json
import zipfile
from pathlib import Path
import pytest
from arautopilot.core.actuator_config import ActuatorConfig, ActuatorType
from arautopilot.core.pid_config import PidConfig, PidGains
from arautopilot.core.project_config import ProjectConfig
from arautopilot.core.sensor_config import (
DualRudderSensorConfig,
RudderSensorConfig,
RudderSensorType,
)
from arautopilot.core.vessel_config import VesselConfig, VesselType
from arautopilot.studio.compiler.appack import AppackCompiler
def _make_project(*, dual_sensors: bool = False) -> ProjectConfig:
redundant = (
RudderSensorConfig(
type=RudderSensorType.AS5048A_SPI,
label="Redundant actuator arm",
spi_cs_gpio=11,
)
if dual_sensors
else None
)
return ProjectConfig(
client_name="Test Client",
project_name="Test Project",
vessel=VesselConfig(
name="Test Vessel",
type=VesselType.YACHT_MOTOR_PLANEO,
length_m=30.0,
max_speed_kn=18.0,
actuator=ActuatorConfig(
type=ActuatorType.HYDRAULIC_REVERSIBLE,
max_rudder_angle_deg=35.0,
),
pid=PidConfig(
inner_loop_base=PidGains(kp=1.5, ki=0.1, kd=0.05),
outer_loop_base=PidGains(kp=2.0, ki=0.05, kd=0.3),
),
sensors=DualRudderSensorConfig(
primary=RudderSensorConfig(
type=RudderSensorType.AS5048A_SPI,
spi_cs_gpio=10,
),
redundant=redundant,
),
),
)
class TestAppackCompiler:
def test_compile_creates_file(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
compiler = AppackCompiler(_make_project())
result = compiler.compile(out)
assert result == out
assert out.exists()
def test_appack_is_valid_zip(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
AppackCompiler(_make_project()).compile(out)
assert zipfile.is_zipfile(out)
def test_appack_contains_required_files(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project()
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
names = set(zf.namelist())
assert f"{prefix}/manifest.json" in names
assert f"{prefix}/project.yaml" in names
assert f"{prefix}/firmware_config.h" in names
assert f"{prefix}/install_notes.txt" in names
def test_manifest_fields(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project()
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
manifest = json.loads(zf.read(f"{prefix}/manifest.json"))
assert manifest["project_id"] == str(project.project_id)
assert manifest["client_name"] == "Test Client"
assert manifest["vessel_name"] == "Test Vessel"
assert "checksums" in manifest
assert "project_yaml_sha256" in manifest["checksums"]
assert "firmware_config_h_sha256" in manifest["checksums"]
def test_firmware_header_contains_defines(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project()
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
header = zf.read(f"{prefix}/firmware_config.h").decode()
assert "#define AR_RUDDER_ANGLE_LIMIT_DEG" in header
assert "#define AR_PID_INNER_KP" in header
assert "#define AR_PID_OUTER_KP" in header
assert "#define AR_SENSOR_PRIMARY_TYPE" in header
assert "ACTUATOR_HYDRAULIC_REVERSIBLE" in header
assert "RUDDER_SENSOR_AS5048A_SPI" in header
assert "35.0f" in header
def test_firmware_header_dual_sensor(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project(dual_sensors=True)
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
header = zf.read(f"{prefix}/firmware_config.h").decode()
notes = zf.read(f"{prefix}/install_notes.txt").decode()
assert "#define AR_SENSOR_REDUNDANT 1" in header
assert "GPIO CS=11" in notes
def test_firmware_header_single_sensor(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project(dual_sensors=False)
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
header = zf.read(f"{prefix}/firmware_config.h").decode()
assert "#define AR_SENSOR_REDUNDANT 0" in header
def test_project_yaml_roundtrips(self, tmp_path: Path) -> None:
out = tmp_path / "test.appack"
project = _make_project()
AppackCompiler(project).compile(out)
prefix = str(project.project_id)
with zipfile.ZipFile(out) as zf:
yaml_text = zf.read(f"{prefix}/project.yaml").decode()
restored = ProjectConfig.from_yaml(yaml_text)
assert restored.client_name == project.client_name
assert restored.vessel.name == project.vessel.name
def test_empty_client_name_raises(self) -> None:
project = _make_project()
object.__setattr__(project, "client_name", " ")
with pytest.raises(ValueError, match="client_name"):
AppackCompiler(project)
def test_output_parent_created(self, tmp_path: Path) -> None:
out = tmp_path / "deep" / "nested" / "output.appack"
AppackCompiler(_make_project()).compile(out)
assert out.exists()
+98
View File
@@ -0,0 +1,98 @@
"""Tests for RudderSensorConfig and DualRudderSensorConfig — Sprint 4."""
from __future__ import annotations
import pytest
from pydantic import ValidationError
from arautopilot.core.sensor_config import (
DualRudderSensorConfig,
RudderSensorConfig,
RudderSensorType,
)
def _primary(
sensor_type: RudderSensorType = RudderSensorType.AS5048A_SPI,
cs: int = 10,
fsd: float = 35.0,
) -> RudderSensorConfig:
return RudderSensorConfig(type=sensor_type, spi_cs_gpio=cs, full_scale_deg=fsd)
def _redundant(cs: int = 11) -> RudderSensorConfig:
return RudderSensorConfig(
type=RudderSensorType.AS5048A_SPI,
label="Redundant actuator arm",
spi_cs_gpio=cs,
full_scale_deg=35.0,
)
class TestRudderSensorConfig:
def test_as5048a_defaults(self) -> None:
s = _primary()
assert s.type == RudderSensorType.AS5048A_SPI
assert s.spi_cs_gpio == 10
assert s.full_scale_deg == 35.0
assert s.zero_offset_deg == 0.0
assert s.divergence_alarm_deg == 3.0
assert s.divergence_failover_deg == 6.0
def test_potentiometer_type(self) -> None:
s = _primary(sensor_type=RudderSensorType.POTENTIOMETER)
assert s.type == RudderSensorType.POTENTIOMETER
def test_gpio_bounds(self) -> None:
with pytest.raises(ValidationError):
RudderSensorConfig(type=RudderSensorType.AS5048A_SPI, spi_cs_gpio=40)
with pytest.raises(ValidationError):
RudderSensorConfig(type=RudderSensorType.AS5048A_SPI, spi_cs_gpio=-1)
def test_full_scale_bounds(self) -> None:
with pytest.raises(ValidationError):
_primary(fsd=0.0)
with pytest.raises(ValidationError):
_primary(fsd=46.0)
def test_label_optional(self) -> None:
s = RudderSensorConfig(type=RudderSensorType.AS5048A_SPI)
assert s.label == ""
def test_extra_fields_forbidden(self) -> None:
with pytest.raises(ValidationError):
RudderSensorConfig(
type=RudderSensorType.AS5048A_SPI,
unknown_field="boom", # type: ignore[call-arg]
)
def test_roundtrip_json(self) -> None:
s = _primary()
restored = RudderSensorConfig.model_validate_json(s.model_dump_json())
assert restored == s
class TestDualRudderSensorConfig:
def test_single_sensor_mode(self) -> None:
d = DualRudderSensorConfig(primary=_primary())
assert d.redundant is None
assert not d.has_redundancy
def test_dual_sensor_mode(self) -> None:
d = DualRudderSensorConfig(primary=_primary(), redundant=_redundant())
assert d.redundant is not None
assert d.has_redundancy
def test_different_cs_pins(self) -> None:
d = DualRudderSensorConfig(primary=_primary(cs=10), redundant=_redundant(cs=11))
assert d.primary.spi_cs_gpio == 10
assert d.redundant is not None
assert d.redundant.spi_cs_gpio == 11
def test_roundtrip_yaml(self) -> None:
import yaml
d = DualRudderSensorConfig(primary=_primary(), redundant=_redundant())
data = yaml.safe_dump(d.model_dump(mode="json"))
restored = DualRudderSensorConfig.model_validate(yaml.safe_load(data))
assert restored == d
@@ -0,0 +1,138 @@
// =============================================================================
// rudder_actuator_electric.h -- reversible DC motor actuator (Sprint 4)
// =============================================================================
//
// Drives a reversible DC motor with mechanical end-stops (Lewmar, Simpson
// Lawrence) via H-bridge (DRV8833 or equivalent):
// DO1 (GPIO) -- IN1 (port direction)
// DO2 (GPIO) -- IN2 (starboard direction)
// DO3 (GPIO) -- master power relay (ENA)
// PWM channel -- motor speed (shared LEDC channel 1)
//
// Architecture identical to HydraulicRudderActuator; only the PWM mapping
// differs (electric motors typically need a minimum useful duty to overcome
// static friction — see AR_ACTUATOR_DEADBAND_PCT in firmware_config.h).
// =============================================================================
#pragma once
#include <algorithm>
#include <cmath>
#include <cstdint>
#include <Preferences.h>
#include <driver/ledc.h>
#include "hal/di_do.h"
#include "hal/pinout.h"
#include "hal/rudder_actuator_iface.h"
#include "hal/rudder_sensor.h"
#include "system/ar_log.h"
#ifndef AR_ACTUATOR_DEADBAND_PCT
# define AR_ACTUATOR_DEADBAND_PCT 5.0f
#endif
namespace arautopilot::hal {
class ElectricDcRudderActuator final : public IRudderActuator {
public:
static constexpr const char* NVS_NS = "ar_act";
static constexpr const char* NVS_KEY = "angle_limit";
static constexpr float DEFAULT_LIMIT_DEG = 35.0f;
static constexpr float MIN_LIMIT_DEG = 15.0f;
static constexpr float MAX_LIMIT_DEG = 45.0f;
void init() override {
_prefs.begin(NVS_NS, false);
_limit_deg = _prefs.getFloat(NVS_KEY, DEFAULT_LIMIT_DEG);
_limit_deg = std::clamp(_limit_deg, MIN_LIMIT_DEG, MAX_LIMIT_DEG);
_prefs.end();
pinMode(PIN_DO1, OUTPUT); digitalWrite(PIN_DO1, LOW);
pinMode(PIN_DO2, OUTPUT); digitalWrite(PIN_DO2, LOW);
pinMode(PIN_DO3, OUTPUT); digitalWrite(PIN_DO3, LOW);
ledc_timer_config_t timer{};
timer.speed_mode = LEDC_LOW_SPEED_MODE;
timer.duty_resolution = LEDC_TIMER_8_BIT;
timer.timer_num = LEDC_TIMER_1;
timer.freq_hz = 20000; // 20 kHz — inaudible for DC motor
timer.clk_cfg = LEDC_AUTO_CLK;
ledc_timer_config(&timer);
ledc_channel_config_t ch{};
ch.speed_mode = LEDC_LOW_SPEED_MODE;
ch.channel = LEDC_CHANNEL_1;
ch.timer_sel = LEDC_TIMER_1;
ch.gpio_num = PIN_ACTUATOR_PWM;
ch.duty = 0;
ledc_channel_config(&ch);
AR_LOGI("ElectricActuator", "init OK, limit=%.1f deg", _limit_deg);
}
void power_on() override { digitalWrite(PIN_DO3, HIGH); _powered = true; }
void power_off() override { _stop_outputs(); digitalWrite(PIN_DO3, LOW); _powered = false; }
bool command(float angle_deg) override {
if (!_powered || _at_standby()) return false;
const float clamped = std::clamp(angle_deg, -_limit_deg, _limit_deg);
if (clamped > 0.0f && _limit_stbd()) { _stop_outputs(); return false; }
if (clamped < 0.0f && _limit_port()) { _stop_outputs(); return false; }
const float raw_pct = std::fabsf(clamped) / _limit_deg * 100.0f;
// Apply deadband: below deadband -> 0, otherwise scale to [deadband, 100]
const float db = AR_ACTUATOR_DEADBAND_PCT;
const float duty_pct = (raw_pct < 0.5f) ? 0.0f
: db + raw_pct * (100.0f - db) / 100.0f;
_set_pwm(static_cast<uint8_t>(std::min(duty_pct, 100.0f) * 255.0f / 100.0f));
if (clamped > 0.01f) {
digitalWrite(PIN_DO1, LOW); digitalWrite(PIN_DO2, HIGH);
} else if (clamped < -0.01f) {
digitalWrite(PIN_DO2, LOW); digitalWrite(PIN_DO1, HIGH);
} else {
_stop_outputs();
}
return true;
}
float actual_angle_deg() const override { return rudder_sensor_get_angle_deg(); }
bool is_powered() const override { return _powered; }
bool at_limit() const override { return _limit_port() || _limit_stbd(); }
float angle_limit_deg() const override { return _limit_deg; }
ActuatorType type() const override { return ActuatorType::ELECTRIC_DC; }
void set_angle_limit_deg(float limit_deg) override {
_limit_deg = std::clamp(limit_deg, MIN_LIMIT_DEG, MAX_LIMIT_DEG);
_prefs.begin(NVS_NS, false);
_prefs.putFloat(NVS_KEY, _limit_deg);
_prefs.end();
AR_LOGI("ElectricActuator", "angle_limit updated -> %.1f deg", _limit_deg);
}
private:
Preferences _prefs;
float _limit_deg{DEFAULT_LIMIT_DEG};
bool _powered{false};
static bool _limit_port() { return digitalRead(PIN_DI2) == HIGH; }
static bool _limit_stbd() { return digitalRead(PIN_DI3) == HIGH; }
static bool _at_standby();
void _stop_outputs() {
_set_pwm(0);
digitalWrite(PIN_DO1, LOW);
digitalWrite(PIN_DO2, LOW);
}
static void _set_pwm(uint8_t duty) {
ledc_set_duty(LEDC_LOW_SPEED_MODE, LEDC_CHANNEL_1, duty);
ledc_update_duty(LEDC_LOW_SPEED_MODE, LEDC_CHANNEL_1);
}
};
} // namespace arautopilot::hal
@@ -0,0 +1,52 @@
// =============================================================================
// rudder_actuator_factory.h -- selects concrete actuator at build time
// =============================================================================
//
// The AR_ACTUATOR_TYPE macro is defined in firmware_config.h (generated by
// the .appack compiler). If firmware_config.h is not present (standalone
// build), falls back to ACTUATOR_HYDRAULIC_REVERSIBLE.
//
// Usage in main.cpp:
//
// #include "hal/rudder_actuator_factory.h"
// // ...
// IRudderActuator& actuator = arautopilot::hal::g_rudder_actuator;
// actuator.init();
// =============================================================================
#pragma once
// Pull in generated config if available
#if __has_include("firmware_config.h")
# include "firmware_config.h"
#endif
// Actuator type constants (must match _actuator_type_to_enum in appack.py)
#define ACTUATOR_HYDRAULIC_REVERSIBLE 0
#define ACTUATOR_ELECTRIC_DC 1
#define ACTUATOR_SERVOMOTOR 2
#define ACTUATOR_STERNDRIVE_ANALOG 3
#define ACTUATOR_TEST_RIG_MOTOR 4
#ifndef AR_ACTUATOR_TYPE
# define AR_ACTUATOR_TYPE ACTUATOR_HYDRAULIC_REVERSIBLE
#endif
#if AR_ACTUATOR_TYPE == ACTUATOR_HYDRAULIC_REVERSIBLE
# include "hal/rudder_actuator_hydraulic.h"
namespace arautopilot::hal {
static HydraulicRudderActuator g_rudder_actuator;
}
#elif AR_ACTUATOR_TYPE == ACTUATOR_ELECTRIC_DC
# include "hal/rudder_actuator_electric.h"
namespace arautopilot::hal {
static ElectricDcRudderActuator g_rudder_actuator;
}
#else
// Fallback: hydraulic (compile warning so the integrator notices)
# pragma message("AR_ACTUATOR_TYPE not recognised — defaulting to HydraulicRudderActuator")
# include "hal/rudder_actuator_hydraulic.h"
namespace arautopilot::hal {
static HydraulicRudderActuator g_rudder_actuator;
}
#endif
@@ -0,0 +1,148 @@
// =============================================================================
// rudder_actuator_hydraulic.h -- reversible hydraulic pump actuator (Sprint 4)
// =============================================================================
//
// Drives a reversible hydraulic pump (Hynautic, Hypro, Octopus, Vetus,
// Lecomble & Schmitt) via:
// DO1 (GPIO) -- port direction solenoid
// DO2 (GPIO) -- starboard direction solenoid
// DO3 (GPIO) -- master power relay
// PWM channel -- pump speed proportional to |angle_error|
//
// The software angle limit is loaded from NVS at init() and clamped to
// [15, 45] degrees. It can be updated live via set_angle_limit_deg()
// which also persists the new value to NVS.
// =============================================================================
#pragma once
#include <algorithm>
#include <cmath>
#include <cstdint>
#include <Preferences.h>
#include <driver/ledc.h>
#include "hal/di_do.h"
#include "hal/pinout.h"
#include "hal/rudder_actuator_iface.h"
#include "hal/rudder_sensor.h"
#include "system/ar_log.h"
namespace arautopilot::hal {
class HydraulicRudderActuator final : public IRudderActuator {
public:
// NVS namespace + key for the angle limit
static constexpr const char* NVS_NS = "ar_act";
static constexpr const char* NVS_KEY = "angle_limit";
static constexpr float DEFAULT_LIMIT_DEG = 35.0f;
static constexpr float MIN_LIMIT_DEG = 15.0f;
static constexpr float MAX_LIMIT_DEG = 45.0f;
void init() override {
// Load angle limit from NVS (falls back to firmware_config.h default)
_prefs.begin(NVS_NS, false);
_limit_deg = _prefs.getFloat(NVS_KEY, DEFAULT_LIMIT_DEG);
_limit_deg = std::clamp(_limit_deg, MIN_LIMIT_DEG, MAX_LIMIT_DEG);
_prefs.end();
// Configure direction + power GPIOs
pinMode(PIN_DO1, OUTPUT); digitalWrite(PIN_DO1, LOW); // port
pinMode(PIN_DO2, OUTPUT); digitalWrite(PIN_DO2, LOW); // starboard
pinMode(PIN_DO3, OUTPUT); digitalWrite(PIN_DO3, LOW); // master power
// Configure LEDC PWM for pump speed
ledc_timer_config_t timer{};
timer.speed_mode = LEDC_LOW_SPEED_MODE;
timer.duty_resolution = LEDC_TIMER_8_BIT;
timer.timer_num = LEDC_TIMER_0;
timer.freq_hz = 5000;
timer.clk_cfg = LEDC_AUTO_CLK;
ledc_timer_config(&timer);
ledc_channel_config_t ch{};
ch.speed_mode = LEDC_LOW_SPEED_MODE;
ch.channel = LEDC_CHANNEL_0;
ch.timer_sel = LEDC_TIMER_0;
ch.gpio_num = PIN_ACTUATOR_PWM;
ch.duty = 0;
ledc_channel_config(&ch);
AR_LOGI("HydraulicActuator", "init OK, limit=%.1f deg", _limit_deg);
}
void power_on() override {
digitalWrite(PIN_DO3, HIGH);
_powered = true;
}
void power_off() override {
_stop_outputs();
digitalWrite(PIN_DO3, LOW);
_powered = false;
}
bool command(float angle_deg) override {
if (!_powered) return false;
if (_at_standby()) return false;
// Software angle limit
const float clamped = std::clamp(angle_deg, -_limit_deg, _limit_deg);
// Hard-stop if limit switch asserted in the commanded direction
if (clamped > 0.0f && _limit_stbd()) { _stop_outputs(); return false; }
if (clamped < 0.0f && _limit_port()) { _stop_outputs(); return false; }
const float duty_pct = std::fabsf(clamped) / _limit_deg * 100.0f;
_set_pwm(static_cast<uint8_t>(duty_pct * 255.0f / 100.0f));
if (clamped > 0.01f) {
digitalWrite(PIN_DO1, LOW); digitalWrite(PIN_DO2, HIGH);
} else if (clamped < -0.01f) {
digitalWrite(PIN_DO2, LOW); digitalWrite(PIN_DO1, HIGH);
} else {
_stop_outputs();
}
return true;
}
float actual_angle_deg() const override {
return rudder_sensor_get_angle_deg();
}
bool is_powered() const override { return _powered; }
bool at_limit() const override { return _limit_port() || _limit_stbd(); }
float angle_limit_deg() const override { return _limit_deg; }
ActuatorType type() const override { return ActuatorType::HYDRAULIC_REVERSIBLE; }
void set_angle_limit_deg(float limit_deg) override {
_limit_deg = std::clamp(limit_deg, MIN_LIMIT_DEG, MAX_LIMIT_DEG);
_prefs.begin(NVS_NS, false);
_prefs.putFloat(NVS_KEY, _limit_deg);
_prefs.end();
AR_LOGI("HydraulicActuator", "angle_limit updated -> %.1f deg", _limit_deg);
}
private:
Preferences _prefs;
float _limit_deg{DEFAULT_LIMIT_DEG};
bool _powered{false};
static bool _limit_port() { return digitalRead(PIN_DI2) == HIGH; }
static bool _limit_stbd() { return digitalRead(PIN_DI3) == HIGH; }
static bool _at_standby(); // defined in modes/standby.cpp
void _stop_outputs() {
_set_pwm(0);
digitalWrite(PIN_DO1, LOW);
digitalWrite(PIN_DO2, LOW);
}
static void _set_pwm(uint8_t duty) {
ledc_set_duty(LEDC_LOW_SPEED_MODE, LEDC_CHANNEL_0, duty);
ledc_update_duty(LEDC_LOW_SPEED_MODE, LEDC_CHANNEL_0);
}
};
} // namespace arautopilot::hal
@@ -0,0 +1,84 @@
// =============================================================================
// rudder_actuator_iface.h -- abstract rudder actuator interface (Sprint 4)
// =============================================================================
//
// The PID inner loop knows only about IRudderActuator. The concrete
// implementation (hydraulic, electric-DC, servo, or test-rig motor) is
// selected at build time via the AR_ACTUATOR_TYPE macro produced by the
// .appack firmware_config.h.
//
// Safety invariants (all implementations must honour):
// 1. angle_limit_deg -- software limit loaded from NVS (or firmware_config.h
// default). Commanded angles are clamped before
// reaching the physical output.
// 2. Limit switches -- DI2 (port) / DI3 (starboard) block motion in
// the corresponding direction regardless of command.
// 3. Master power -- DO3 must be HIGH before any motion.
// 4. STANDBY guard -- implementations check current_mode() == STANDBY
// and refuse motion.
// =============================================================================
#pragma once
#include <cstdint>
namespace arautopilot::hal {
// ---------------------------------------------------------------------------
// Actuator type enumeration (mirrors .appack firmware_config.h values)
// ---------------------------------------------------------------------------
enum class ActuatorType : uint8_t {
HYDRAULIC_REVERSIBLE = 0, ///< Reversible hydraulic pump (Hynautic, Hypro …)
ELECTRIC_DC = 1, ///< Reversible DC motor with mechanical end-stops
SERVOMOTOR = 2, ///< Servomotor with built-in position feedback
STERNDRIVE_ANALOG = 3, ///< Analog directional sterndrive
TEST_RIG_MOTOR = 4, ///< HIL test rig: DC motor rotates compass platform
};
// ---------------------------------------------------------------------------
// Abstract interface
// ---------------------------------------------------------------------------
class IRudderActuator {
public:
virtual ~IRudderActuator() = default;
/// One-time hardware initialisation (GPIOs, PWM channels …).
virtual void init() = 0;
/// Engage master power relay. Required before command().
virtual void power_on() = 0;
/// Drop master power relay and stop all outputs immediately.
virtual void power_off() = 0;
/// Command the actuator toward *angle_deg* (positive = starboard).
/// The implementation MUST clamp to [-angle_limit_deg, +angle_limit_deg]
/// before driving any output.
/// Returns true if the command was applied; false if an interlock blocked it.
virtual bool command(float angle_deg) = 0;
/// Estimated actual rudder angle (degrees) from the sensor/feedback.
/// Returns 0.0 if no feedback is available.
virtual float actual_angle_deg() const = 0;
/// True if master power relay is currently energised.
virtual bool is_powered() const = 0;
/// True if either limit switch is asserted.
virtual bool at_limit() const = 0;
/// Current software angle limit (degrees, positive value).
/// Can be updated at runtime via set_angle_limit_deg().
virtual float angle_limit_deg() const = 0;
/// Update the software angle limit. Clamped to [15, 45] degrees.
/// Persisted to NVS by the concrete implementation.
virtual void set_angle_limit_deg(float limit_deg) = 0;
/// Actuator type tag for diagnostics.
virtual ActuatorType type() const = 0;
};
} // namespace arautopilot::hal