electron-rare/RTC_BL_PHONE
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add strict A252 gate and hardware/orchestrator validation

Browse Source
This commit is contained in:
Clément SAILLANT
2026-02-25 14:34:24 +01:00
parent d2befa2c8d
commit 2e4e90adef
15 changed files with 1230 additions and 286 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/ci.yml
+3 -3
View File
@@ -7,7 +7,7 @@ on:
branches: [main, release/stable]
jobs:
pre-merge:
branch-gate:
runs-on: ubuntu-latest
steps:
- name: Checkout
@@ -23,8 +23,8 @@ jobs:
python -m pip install --upgrade pip
pip install platformio
- name: Run pre-merge gate
run: ./scripts/pre_merge.sh
- name: Run branch gate
run: ./scripts/branch_gate.sh --profile a252
- name: Upload route parity report
if: always()
docs/a252_strict_gate_summary.md
+37
View File
@@ -0,0 +1,37 @@
# A252 Strict Gate Summary
- Date UTC: 2026-02-25T13:33:04.090306+00:00
- Verdict global: PASS
- Port A252: `/dev/cu.usbserial-0001`
- ZeroClaw: `http://127.0.0.1:8788`
## Étapes
| Étape | État |
|---|---|
| zeroclaw_hw_preflight | PASS |
| zeroclaw_orchestrator_health | PASS |
| branch_gate_profile_a252 | PASS |
| hw_validation_a252 | PASS |
## Stacks A252 (hw_validation)
| Stack/Scénario | État |
|---|---|
| serial_smoke | PASS |
| serial_hook_ring_audio | PASS |
| serial_network_stack | PASS |
| http_endpoints | PASS |
| manual_hook_transition | MANUAL_SKIP |
| manual_ring_behavior | MANUAL_SKIP |
| manual_audio_path | MANUAL_SKIP |
| manual_hfp_pairing | MANUAL_SKIP |
## ZeroClaw Docker
| Check | État |
|---|---|
| status | PASS |
| agents | PASS |
| workflows | PASS |
| provider_scan | PASS |
docs/branch_quality_gate.md
+26 -10
View File
@@ -4,29 +4,34 @@ Ce document décrit l’ordonnancement unique des contrôles de conformité avan
## Ordre d’exécution
Le script `scripts/pre_merge.sh` exécute les contrôles suivants, dans l’ordre :
Le script `scripts/branch_gate.sh` exécute les contrôles suivants, dans l’ordre :
1. `python3 -m py_compile scripts/hw_validation.py`
2. `platformio test --without-uploading --without-testing -e esp32dev`
3. Compilation + exécution du test hôte DTMF:
- `c++ -std=c++17 -Wall -Wextra -pedantic -Isrc test/host/test_dtmf_host.cpp src/telephony/DtmfDecoder.cpp -o .pio/host/test_dtmf_host`
- `.pio/host/test_dtmf_host`
4. `python3 -m unittest scripts/test_check_web_route_parity.py scripts/test_runtime_contracts.py`
4. `python3 -m unittest scripts/test_check_web_route_parity.py scripts/test_runtime_contracts.py scripts/test_hw_validation_contracts.py`
5. `python3 scripts/check_web_route_parity.py --report-json artifacts/route_parity_report.json`
6. Build PlatformIO des cibles (par défaut) :
- `esp32dev`
- `esp32-s3-devkitc-1`
- `esp32-s3-usb-host`
- `esp32-s3-usb-msc`
6. Build PlatformIO des cibles selon le profil :
- Profil `a252` (défaut): `esp32dev`
- Profil `full`: `esp32dev`, `esp32-s3-devkitc-1`, `esp32-s3-usb-host`, `esp32-s3-usb-msc`
## Contrat de cette branche
- **ESP-NOW obligatoire** : seule la pile de transport activée est prise en compte pour les scénarios de validation.
- **Bluetooth retiré** : aucun endpoint/commande Bluetooth n’est attendu, ni traité.
- **Auth web Wi-Fi désactivée** : la validation considère les endpoints web accessibles sans authentification basique.
## Cible CI
- `.github/workflows/ci.yml` lance directement `scripts/pre_merge.sh`.
- `.github/workflows/ci.yml` lance `scripts/branch_gate.sh --profile a252`.
- L’artefact `artifacts/route_parity_report.json` est uploadé automatiquement.
## Options utiles
- `--skip-builds` : ignore uniquement la phase build.
- `--profile <a252|full>` : sélectionne le profil de build par défaut.
- `--build-env <env>` : ajoute un env PlatformIO spécifique.
- `--build-envs <env1,env2>` : ajoute plusieurs envs.
- `--report-json <path>` : redirige le rapport parity JSON.
@@ -34,6 +39,17 @@ Le script `scripts/pre_merge.sh` exécute les contrôles suivants, dans l’ordr
Exemples :
```bash
bash scripts/pre_merge.sh --skip-builds
bash scripts/pre_merge.sh --build-env esp32dev --build-env esp32-s3-devkitc-1
bash scripts/branch_gate.sh --skip-builds
bash scripts/branch_gate.sh --profile a252
bash scripts/branch_gate.sh --profile full
bash scripts/branch_gate.sh --build-env esp32dev --build-env esp32-s3-devkitc-1
```
## Profil standard
- Le profil standard de cette branche est `a252` (A252 strict).
- Le profil `full` reste disponible pour les campagnes de compatibilité multi-cartes.
## Redirection historique
- `docs/pre_merge_checks.md` est conservé comme redirection vers ce document.
docs/hw_validation_a252_report.md
+15
View File
File diff suppressed because one or more lines are too long
docs/pre_merge_checks.md
+7
View File
@@ -0,0 +1,7 @@
# Pré-merge checks (redirect)
Ce document est un alias conservé pour compatibilité IDE/outils.
Référence active:
- [docs/branch_quality_gate.md](./branch_quality_gate.md)
docs/zeroclaw_orchestration.md
+57 -4
View File
@@ -1,6 +1,6 @@
# Orchestration ZeroClaw (RTC)
Dernière mise à jour: 2026-02-21.
Dernière mise à jour: 2026-02-25.
## Objectif
@@ -10,6 +10,37 @@ Rendre les boucles hardware RTC plus robustes:
- session agent ZeroClaw ciblée sur le repo RTC,
- exécution coordonnée avec `Kill_LIFE` (orchestrateur dual-repo).
## Mode retenu: Docker orchestration + hardware hôte
Stack active:
- orchestrateur: `http://127.0.0.1:8788`
- n8n: `http://127.0.0.1:5678`
- OpenWebUI: `http://127.0.0.1:3001`
Utilisation recommandée:
1. Pilotage agents/subagents via API orchestrateur (`GET /api/agents`, `GET /api/workflows`, `GET /api/status`, `POST /api/run`).
2. Build/flash/monitor RTC_A252 exécutés localement sur macOS (hors conteneur).
Contrainte validée:
- Ne pas lancer `rtc_firmware_loop` depuis l'orchestrateur Docker pour cette passe RTC:
- conteneur sans chemin repo RTC hôte monté (`/Users/.../RTC_BL_PHONE`),
- `pio` non installé dans l'image orchestrateur,
- pas d'accès direct aux ports série macOS.
Actions Docker autorisées dans ce mode:
- `provider_scan`
- actions webhook (`rtc_webhook`, `zacus_webhook`, `general_prompt_fanout`)
Check health recommandé:
```bash
python3 scripts/zeroclaw_orchestrator_health.py --base-url "${ZEROCLAW_ORCH:-http://127.0.0.1:8788}"
```
## 1) Préflight hardware local
Lancer avant `autoflash.py`, `hw_validation.py` ou tout upload PlatformIO:
@@ -50,13 +81,13 @@ Si credentials manquants, choisir un provider:
Contexte bench actuel: carte `esp32dev` (audio dev), sans cible S3 connectee.
Commande de reference:
Commande de référence A252 (USB-Serial) :
```bash
cd /Users/cils/Documents/Lelectron_rare/RTC_BL_PHONE
python3 scripts/zeroclaw_hw_preflight.py --require-port --zeroclaw-bin /Users/cils/Documents/Lelectron_rare/Kill_LIFE/zeroclaw/target/release/zeroclaw
pio run -e esp32dev
pio run -e esp32dev -t upload --upload-port /dev/cu.SLAB_USBtoUART
pio run -e esp32dev -t upload --upload-port /dev/cu.usbserial-0001
```
Smoke terminal attendu:
@@ -69,7 +100,29 @@ Note technique:
- Sur `esp32dev`, il faut initialiser le stack reseau avant `AsyncWebServer::begin()`
pour eviter le crash `lwIP Invalid mbox`.
## 5) Preuve boucle complète (2026-02-21)
## 5) Gate stricte A252 (local)
Variables attendues:
- `A252_PORT` (ex: `/dev/cu.usbserial-0001`)
- `A252_WIFI_SSID`
- `A252_WIFI_PASSWORD`
- `ZEROCLAW_ORCH` (défaut `http://127.0.0.1:8788`)
Exécution:
```bash
bash scripts/a252_strict_gate.sh
```
Sorties:
- `artifacts/zeroclaw_orchestrator_health.json`
- `artifacts/hw_validation_a252_report.json`
- `docs/hw_validation_a252_report.md`
- `docs/a252_strict_gate_summary.md`
## 6) Preuve boucle complète (2026-02-21)
Etat courant:
scripts/a252_strict_gate.sh
Executable
+156
View File
@@ -0,0 +1,156 @@
#!/usr/bin/env bash
set -uo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "${ROOT_DIR}"
A252_PORT="${A252_PORT:-/dev/cu.usbserial-0001}"
ZEROCLAW_ORCH="${ZEROCLAW_ORCH:-http://127.0.0.1:8788}"
ZEROCLAW_BIN="${ZEROCLAW_BIN:-zeroclaw}"
A252_HOOK_OBSERVE_SECONDS="${A252_HOOK_OBSERVE_SECONDS:-45}"
HW_REPORT_JSON="artifacts/hw_validation_a252_report.json"
HW_REPORT_MD="docs/hw_validation_a252_report.md"
ZEROCLAW_REPORT_JSON="artifacts/zeroclaw_orchestrator_health.json"
SUMMARY_MD="docs/a252_strict_gate_summary.md"
STATUS_PREFLIGHT="NOT_RUN"
STATUS_ZEROCLAW="NOT_RUN"
STATUS_BRANCH_GATE="NOT_RUN"
STATUS_HW_VALIDATION="NOT_RUN"
OVERALL="FAIL"
log() {
echo "[a252-strict-gate] $*"
}
write_summary() {
STATUS_PREFLIGHT="${STATUS_PREFLIGHT}" \
STATUS_ZEROCLAW="${STATUS_ZEROCLAW}" \
STATUS_BRANCH_GATE="${STATUS_BRANCH_GATE}" \
STATUS_HW_VALIDATION="${STATUS_HW_VALIDATION}" \
OVERALL="${OVERALL}" \
ZEROCLAW_ORCH="${ZEROCLAW_ORCH}" \
A252_PORT="${A252_PORT}" \
HW_REPORT_JSON="${HW_REPORT_JSON}" \
ZEROCLAW_REPORT_JSON="${ZEROCLAW_REPORT_JSON}" \
SUMMARY_MD="${SUMMARY_MD}" \
python3 - <<'PY'
import json
import os
from datetime import datetime, timezone
from pathlib import Path
summary_path = Path(os.environ["SUMMARY_MD"])
summary_path.parent.mkdir(parents=True, exist_ok=True)
def load_json(path: Path):
if not path.exists():
return None
try:
return json.loads(path.read_text(encoding="utf-8"))
except Exception:
return None
hw_payload = load_json(Path(os.environ["HW_REPORT_JSON"]))
zc_payload = load_json(Path(os.environ["ZEROCLAW_REPORT_JSON"]))
lines = [
"# A252 Strict Gate Summary",
"",
f"- Date UTC: {datetime.now(timezone.utc).isoformat()}",
f"- Verdict global: {os.environ['OVERALL']}",
f"- Port A252: `{os.environ['A252_PORT']}`",
f"- ZeroClaw: `{os.environ['ZEROCLAW_ORCH']}`",
"",
"## Étapes",
"",
"| Étape | État |",
"|---|---|",
f"| zeroclaw_hw_preflight | {os.environ['STATUS_PREFLIGHT']} |",
f"| zeroclaw_orchestrator_health | {os.environ['STATUS_ZEROCLAW']} |",
f"| branch_gate_profile_a252 | {os.environ['STATUS_BRANCH_GATE']} |",
f"| hw_validation_a252 | {os.environ['STATUS_HW_VALIDATION']} |",
]
if hw_payload:
lines.extend([
"",
"## Stacks A252 (hw_validation)",
"",
"| Stack/Scénario | État |",
"|---|---|",
])
for item in hw_payload.get("results", []):
name = item.get("name", "")
state = item.get("state", "")
lines.append(f"| {name} | {state} |")
if zc_payload:
lines.extend([
"",
"## ZeroClaw Docker",
"",
"| Check | État |",
"|---|---|",
])
for item in zc_payload.get("results", []):
name = item.get("name", "")
state = item.get("state", "")
lines.append(f"| {name} | {state} |")
summary_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
PY
}
run_step() {
local step_var="$1"
local description="$2"
shift 2
log "${description}"
if "$@"; then
printf -v "${step_var}" "PASS"
return 0
fi
printf -v "${step_var}" "FAIL"
OVERALL="FAIL"
write_summary
return 1
}
if [[ -z "${A252_WIFI_SSID:-}" || -z "${A252_WIFI_PASSWORD:-}" ]]; then
log "erreur: A252_WIFI_SSID et A252_WIFI_PASSWORD sont requis"
write_summary
exit 2
fi
run_step STATUS_PREFLIGHT "ZeroClaw hardware preflight" \
python3 scripts/zeroclaw_hw_preflight.py \
--zeroclaw-bin "${ZEROCLAW_BIN}" \
--require-port \
--port "${A252_PORT}" || exit 1
run_step STATUS_ZEROCLAW "ZeroClaw orchestrator health + provider_scan" \
python3 scripts/zeroclaw_orchestrator_health.py \
--base-url "${ZEROCLAW_ORCH}" \
--report-json "${ZEROCLAW_REPORT_JSON}" || exit 1
run_step STATUS_BRANCH_GATE "Branch gate profile a252" \
bash scripts/branch_gate.sh --profile a252 || exit 1
run_step STATUS_HW_VALIDATION "Hardware validation A252 strict" \
python3 scripts/hw_validation.py \
--port-a252 "${A252_PORT}" \
--flash \
--wifi-ssid "${A252_WIFI_SSID}" \
--wifi-password "${A252_WIFI_PASSWORD}" \
--strict-serial-smoke \
--allow-capture-fail-when-disabled \
--require-hook-toggle \
--hook-observe-seconds "${A252_HOOK_OBSERVE_SECONDS}" \
--report-json "${HW_REPORT_JSON}" \
--report-md "${HW_REPORT_MD}" || exit 1
OVERALL="PASS"
write_summary
log "gate A252 strict terminé avec succès"
scripts/branch_gate.sh
Executable
+167
View File
@@ -0,0 +1,167 @@
#!/usr/bin/env bash
set -euo pipefail
ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
cd "${ROOT_DIR}"
DEFAULT_BUILD_ENVS_FULL=(
"esp32dev"
"esp32-s3-devkitc-1"
"esp32-s3-usb-host"
"esp32-s3-usb-msc"
)
DEFAULT_BUILD_ENVS_A252=(
"esp32dev"
)
REPORT_JSON="${ARTIFACT_REPORT_PATH:-artifacts/route_parity_report.json}"
BUILD_ENVS=()
SKIP_BUILDS=0
PROFILE="a252"
log() {
echo "[branch-gate] $*"
}
require_cmd() {
if ! command -v "$1" >/dev/null 2>&1; then
echo "[branch-gate] erreur: commande manquante: $1" >&2
exit 1
fi
}
usage() {
cat <<'EOF'
Usage: scripts/branch_gate.sh [options]
Exécute la chaîne de validation de branche dans un ordre déterministe.
Options:
--profile <a252|full> Profil de build par défaut (a252: esp32dev seulement).
--skip-builds Ignore la phase de build PlatformIO.
--build-env <env> Ajouter explicitement un env PlatformIO (peut se répéter).
--build-envs <env1,env2> Ajouter plusieurs envs en une fois (séparés par des virgules).
--report-json <path> Emplacement du rapport parity JSON (défaut: artifacts/route_parity_report.json).
--help Affiche cette aide.
Sans --build-env ni --build-envs, la séquence build cible :
- profile a252: esp32dev
- profile full: esp32dev, esp32-s3-devkitc-1, esp32-s3-usb-host, esp32-s3-usb-msc
EOF
}
while [[ $# -gt 0 ]]; do
case "$1" in
--profile)
if [[ $# -lt 2 ]]; then
usage
exit 1
fi
case "$2" in
a252|full)
PROFILE="$2"
;;
*)
echo "[branch-gate] profile invalide: $2 (attendu: a252|full)" >&2
exit 1
;;
esac
shift 2
;;
--skip-builds)
SKIP_BUILDS=1
shift
;;
--build-env)
if [[ $# -lt 2 ]]; then
usage
exit 1
fi
BUILD_ENVS+=("$2")
shift 2
;;
--build-envs)
if [[ $# -lt 2 ]]; then
usage
exit 1
fi
IFS=',' read -r -a extra_envs <<< "$2"
for env_name in "${extra_envs[@]}"; do
if [[ -n "${env_name}" ]]; then
BUILD_ENVS+=("${env_name}")
fi
done
shift 2
;;
--report-json)
if [[ $# -lt 2 ]]; then
usage
exit 1
fi
REPORT_JSON="$2"
shift 2
;;
--help|-h)
usage
exit 0
;;
*)
echo "[branch-gate] option inconnue: $1" >&2
usage
exit 1
;;
esac
done
run_checks() {
log "vérification syntaxe scripts (python)"
python3 -m py_compile scripts/hw_validation.py
log "tests unitaires PlatformIO (esp32dev, mode host)"
platformio test --without-uploading --without-testing -e esp32dev
log "tests host DTMF"
mkdir -p .pio/host
c++ -std=c++17 -Wall -Wextra -pedantic -Isrc test/host/test_dtmf_host.cpp src/telephony/DtmfDecoder.cpp -o .pio/host/test_dtmf_host
.pio/host/test_dtmf_host
log "tests contrat parity/runtime/hw_validation"
python3 -m unittest \
scripts/test_check_web_route_parity.py \
scripts/test_runtime_contracts.py \
scripts/test_hw_validation_contracts.py
log "contrôle route/command parity avec rapport JSON"
mkdir -p "$(dirname "${REPORT_JSON}")"
python3 scripts/check_web_route_parity.py --report-json "${REPORT_JSON}"
}
run_builds() {
if (( SKIP_BUILDS )); then
log "phase build ignorée (--skip-builds)"
return
fi
if (( ${#BUILD_ENVS[@]} == 0 )); then
if [[ "${PROFILE}" == "full" ]]; then
BUILD_ENVS=("${DEFAULT_BUILD_ENVS_FULL[@]}")
else
BUILD_ENVS=("${DEFAULT_BUILD_ENVS_A252[@]}")
fi
fi
log "profil build actif: ${PROFILE}"
for env_name in "${BUILD_ENVS[@]}"; do
log "build PlatformIO: ${env_name}"
platformio run -e "${env_name}"
done
}
require_cmd python3
require_cmd platformio
require_cmd c++
run_checks
run_builds
log "validation de branche terminée avec succès"
scripts/hw_validation.py
+373 -16
View File
@@ -4,6 +4,7 @@
from __future__ import annotations
import argparse
import glob
import json
import subprocess
import sys
@@ -21,6 +22,7 @@ except ImportError: # pragma: no cover
VALID_STATES = {"PASS", "FAIL", "MANUAL_PASS", "MANUAL_FAIL", "MANUAL_SKIP"}
OPTIONAL_SERIAL_COMMANDS = {"WIFI_SCAN", "ESPNOW_STATUS"}
@dataclass
@@ -77,7 +79,7 @@ class SerialEndpoint:
continue
last_line = line
print(f"[{self.port}] {line}")
if line.startswith("{") and line.endswith("}"):
if line and line[0] in ("{", "[") and line[-1] in ("}", "]"):
if expect in {"any", "json"}:
try:
return json.loads(line)
@@ -106,6 +108,26 @@ class SerialEndpoint:
raise RuntimeError(f"serial sync failed: {last_error}")
def resolve_a252_port(explicit_port: str | None) -> str:
if explicit_port:
return explicit_port
preferred_patterns = (
"/dev/cu.usbserial*",
"/dev/tty.usbserial*",
)
for pattern in preferred_patterns:
candidates = sorted(glob.glob(pattern))
if candidates:
print(f"[hw_validation] A252 locked to USB-Serial port: {candidates[0]}")
return candidates[0]
raise RuntimeError(
"No USB-Serial port found for A252. Expected /dev/tty.usbserial* or /dev/cu.usbserial*."
)
def fetch_json(url: str) -> Dict[str, Any]:
req = request.Request(url, method="GET")
with request.urlopen(req, timeout=5) as resp:
@@ -137,7 +159,11 @@ def post_json(url: str, payload: Dict[str, Any]) -> Dict[str, Any]:
return json.loads(resp.read().decode("utf-8"))
def scenario_serial_smoke(dev: SerialEndpoint) -> ScenarioResult:
def scenario_serial_smoke(
dev: SerialEndpoint,
strict_serial_smoke: bool,
allow_capture_fail_when_disabled: bool,
) -> ScenarioResult:
details: Dict[str, Any] = {}
try:
details["ping"] = dev.command("PING", expect="pong")
@@ -146,28 +172,179 @@ def scenario_serial_smoke(dev: SerialEndpoint) -> ScenarioResult:
details["capture_start"] = dev.command("CAPTURE_START", expect="ack")
details["capture_stop"] = dev.command("CAPTURE_STOP", expect="ack")
details["reset_metrics"] = dev.command("RESET_METRICS", expect="ack")
ok = bool(details["ping"].get("ok")) and "telephony" in details["status"]
return ScenarioResult("serial_smoke", "PASS" if ok else "FAIL", details)
state, required_checks, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=strict_serial_smoke,
allow_capture_fail_when_disabled=allow_capture_fail_when_disabled,
)
details["required_checks"] = required_checks
details["failed_checks"] = failed_checks
details["warnings"] = warnings
return ScenarioResult("serial_smoke", state, details)
except Exception as exc:
return ScenarioResult("serial_smoke", "FAIL", {"error": str(exc)})
def _is_success_response(resp: Dict[str, Any]) -> bool:
if "line" in resp:
line = str(resp.get("line")).strip().upper()
if line.startswith("OK "):
return True
if line.startswith("ERR "):
return False
return False
if "ok" in resp:
return bool(resp.get("ok"))
return True
def _has_espnow_capability(resp: Dict[str, Any]) -> bool:
if not isinstance(resp, dict):
return False
if not bool(resp.get("ready")):
return False
peers = resp.get("peer_count")
return isinstance(peers, int) and peers > 0
def _normalize_command(command: str) -> str:
return command.strip().upper()
def _is_soft_unsupported(command: str, resp: Dict[str, Any]) -> bool:
normalized_command = _normalize_command(command)
if normalized_command not in OPTIONAL_SERIAL_COMMANDS:
return False
if not isinstance(resp, dict):
return False
if isinstance(resp.get("line"), str):
line = resp["line"].strip().upper()
return line.startswith("ERR UNSUPPORTED_COMMAND") or "UNSUPPORTED" in line
if isinstance(resp.get("code"), str):
code = resp["code"].strip().upper()
return "UNSUPPORTED" in code
return False
def _is_acceptable_response(command: str, resp: Dict[str, Any], required: bool) -> bool:
if required:
return _is_success_response(resp)
if _is_soft_unsupported(command, resp):
return True
return _is_success_response(resp)
def _extract_capture_enabled(status_payload: Dict[str, Any]) -> bool:
if not isinstance(status_payload, dict):
return True
config = status_payload.get("config")
if not isinstance(config, dict):
return True
audio = config.get("audio")
if not isinstance(audio, dict):
return True
value = audio.get("enable_capture")
if isinstance(value, bool):
return value
return True
def evaluate_serial_smoke_contract(
details: Dict[str, Any],
*,
strict_serial_smoke: bool,
allow_capture_fail_when_disabled: bool,
) -> tuple[str, List[str], List[str], List[str]]:
warnings: List[str] = []
failed_checks: List[str] = []
ping_ok = bool(details.get("ping", {}).get("ok"))
status_payload = details.get("status", {})
status_ok = isinstance(status_payload, dict) and "telephony" in status_payload
call_ok = _is_success_response(details.get("call", {}))
capture_start_ok = _is_success_response(details.get("capture_start", {}))
capture_stop_ok = _is_success_response(details.get("capture_stop", {}))
reset_metrics_ok = _is_success_response(details.get("reset_metrics", {}))
capture_enabled = _extract_capture_enabled(status_payload if isinstance(status_payload, dict) else {})
capture_start_required = capture_enabled or (not allow_capture_fail_when_disabled)
required_checks: List[str] = ["PING", "STATUS", "CALL", "CAPTURE_STOP", "RESET_METRICS"]
if capture_start_required:
required_checks.append("CAPTURE_START")
if not ping_ok:
failed_checks.append("PING")
if not status_ok:
failed_checks.append("STATUS")
if not call_ok:
failed_checks.append("CALL")
if not capture_stop_ok:
failed_checks.append("CAPTURE_STOP")
if not reset_metrics_ok:
failed_checks.append("RESET_METRICS")
if capture_start_required and not capture_start_ok:
failed_checks.append("CAPTURE_START")
if not capture_enabled and not capture_start_ok:
if allow_capture_fail_when_disabled:
warnings.append("capture_start_failed_capture_disabled")
else:
warnings.append("capture_start_required_even_when_capture_disabled")
if strict_serial_smoke:
return ("PASS" if not failed_checks else "FAIL", required_checks, failed_checks, warnings)
minimum_failures = [check for check in failed_checks if check in {"PING", "STATUS"}]
if failed_checks and not minimum_failures:
warnings.append("strict_serial_smoke_disabled")
return ("PASS" if not minimum_failures else "FAIL", required_checks, failed_checks, warnings)
def _quote_arg(value: str) -> str:
escaped = value.replace("\\", "\\\\").replace('"', '\\"')
return f'"{escaped}"'
def _command_with_retry(
dev: SerialEndpoint,
command: str,
*,
expect: str = "any",
timeout_s: float = 6.0,
attempts: int = 1,
retry_delay_s: float = 0.5,
) -> Dict[str, Any]:
last_error: Optional[Exception] = None
effective_attempts = max(1, attempts)
for attempt in range(effective_attempts):
try:
return dev.command(command, timeout_s=timeout_s, expect=expect)
except Exception as exc:
last_error = exc
if attempt + 1 >= effective_attempts:
break
time.sleep(retry_delay_s)
raise RuntimeError(str(last_error) if last_error else f"command failed: {command}")
def scenario_serial_network(dev: SerialEndpoint, wifi_ssid: str, wifi_password: str) -> ScenarioResult:
details: Dict[str, Any] = {}
try:
details["wifi_status_before"] = dev.command("WIFI_STATUS", expect="json")
details["wifi_scan"] = dev.command("WIFI_SCAN", expect="json")
# Wi-Fi scan can be slow right after boot/flash; retry with a longer timeout.
details["wifi_scan"] = _command_with_retry(
dev,
"WIFI_SCAN",
expect="any",
timeout_s=12.0,
attempts=2,
retry_delay_s=1.0,
)
if wifi_ssid:
already_connected = (
bool(details["wifi_status_before"].get("connected"))
@@ -184,27 +361,159 @@ def scenario_serial_network(dev: SerialEndpoint, wifi_ssid: str, wifi_password:
)
time.sleep(2.0)
details["wifi_status_after"] = dev.command("WIFI_STATUS", expect="json")
details["mqtt_status"] = dev.command("MQTT_STATUS", expect="json")
details["espnow_status"] = dev.command("ESPNOW_STATUS", expect="json")
details["bt_status"] = dev.command("BT_STATUS", expect="json")
details["espnow_status"] = _command_with_retry(
dev,
"ESPNOW_STATUS",
expect="any",
timeout_s=8.0,
attempts=2,
retry_delay_s=0.5,
)
checks = [
("WIFI_STATUS", details["wifi_status_before"], True),
("WIFI_SCAN", details["wifi_scan"], False),
("ESPNOW_STATUS", details["espnow_status"], True),
]
if wifi_ssid and "wifi_status_after" in details:
if "wifi_connect" in details:
checks.append(("WIFI_CONNECT", details["wifi_connect"], True))
checks.append(("WIFI_STATUS", details["wifi_status_after"], True))
ok = True
for value in details.values():
if isinstance(value, dict) and not _is_success_response(value):
for command, value, required in checks:
if command == "ESPNOW_STATUS" and not _has_espnow_capability(value):
ok = False
break
if not _is_acceptable_response(command, value, required):
ok = False
break
if ok and wifi_ssid and "wifi_status_after" in details:
wifi_after = details["wifi_status_after"]
if not (
isinstance(wifi_after, dict)
and bool(wifi_after.get("connected"))
and str(wifi_after.get("ssid", "")) == wifi_ssid
):
ok = False
return ScenarioResult("serial_network_stack", "PASS" if ok else "FAIL", details)
except Exception as exc:
return ScenarioResult("serial_network_stack", "FAIL", {"error": str(exc), **details})
def _extract_hook_state(status_payload: Dict[str, Any]) -> str:
if not isinstance(status_payload, dict):
return ""
telephony = status_payload.get("telephony")
if not isinstance(telephony, dict):
return ""
hook = telephony.get("hook")
if not isinstance(hook, str):
return ""
value = hook.strip().upper()
if value in {"ON_HOOK", "OFF_HOOK"}:
return value
return ""
def _extract_dial_tone_active(status_payload: Dict[str, Any]) -> Optional[bool]:
if not isinstance(status_payload, dict):
return None
audio = status_payload.get("audio")
if not isinstance(audio, dict):
return None
value = audio.get("dial_tone_active")
if isinstance(value, bool):
return value
return None
def scenario_serial_hook_ring_audio(
dev: SerialEndpoint,
*,
require_hook_toggle: bool,
hook_observe_seconds: int,
) -> ScenarioResult:
details: Dict[str, Any] = {}
observed_hooks: List[str] = []
try:
details["status_before"] = dev.command("STATUS", expect="json")
initial_hook = _extract_hook_state(details["status_before"])
if initial_hook:
observed_hooks.append(initial_hook)
details["ring"] = dev.command("RING", expect="ack")
details["tone_on"] = dev.command("TONE_ON", expect="ack")
time.sleep(0.4)
details["status_tone_on"] = dev.command("STATUS", expect="json")
hook_after_tone_on = _extract_hook_state(details["status_tone_on"])
if hook_after_tone_on:
observed_hooks.append(hook_after_tone_on)
details["tone_off"] = dev.command("TONE_OFF", expect="ack")
time.sleep(0.3)
details["status_tone_off"] = dev.command("STATUS", expect="json")
hook_after_tone_off = _extract_hook_state(details["status_tone_off"])
if hook_after_tone_off:
observed_hooks.append(hook_after_tone_off)
observe_seconds = max(0, hook_observe_seconds)
poll_deadline = time.time() + float(observe_seconds)
while time.time() < poll_deadline:
poll_status = dev.command("STATUS", expect="json")
hook = _extract_hook_state(poll_status)
if hook:
observed_hooks.append(hook)
time.sleep(0.8)
unique_hooks = sorted(set(observed_hooks))
details["hook_values_observed"] = unique_hooks
details["hook_observe_seconds"] = observe_seconds
ring_ok = _is_success_response(details["ring"])
tone_on_ok = _is_success_response(details["tone_on"])
tone_off_ok = _is_success_response(details["tone_off"])
tone_active_after_on = _extract_dial_tone_active(details["status_tone_on"]) is True
tone_inactive_after_off = _extract_dial_tone_active(details["status_tone_off"]) is False
if require_hook_toggle:
required_hooks = ["ON_HOOK", "OFF_HOOK"]
hook_ok = all(state in unique_hooks for state in required_hooks)
details["required_hook_states"] = required_hooks
else:
hook_ok = len(unique_hooks) > 0
details["required_hook_states"] = ["ON_HOOK|OFF_HOOK"]
details["checks"] = {
"ring_ok": ring_ok,
"tone_on_ok": tone_on_ok,
"tone_active_after_on": tone_active_after_on,
"tone_off_ok": tone_off_ok,
"tone_inactive_after_off": tone_inactive_after_off,
"hook_ok": hook_ok,
}
ok = (
ring_ok
and tone_on_ok
and tone_active_after_on
and tone_off_ok
and tone_inactive_after_off
and hook_ok
)
return ScenarioResult("serial_hook_ring_audio", "PASS" if ok else "FAIL", details)
except Exception as exc:
return ScenarioResult("serial_hook_ring_audio", "FAIL", {"error": str(exc), **details})
def scenario_http(base_url: str) -> ScenarioResult:
details: Dict[str, Any] = {"base_url": base_url}
try:
details["status"] = fetch_json(base_url.rstrip("/") + "/api/status")
details["wifi"] = fetch_json(base_url.rstrip("/") + "/api/network/wifi")
details["mqtt"] = fetch_json(base_url.rstrip("/") + "/api/network/mqtt")
details["espnow"] = fetch_json(base_url.rstrip("/") + "/api/network/espnow")
details["bluetooth"] = fetch_json(base_url.rstrip("/") + "/api/bluetooth")
details["control_call"] = post_json(base_url.rstrip("/") + "/api/control", {"action": "CALL"})
return ScenarioResult("http_endpoints", "PASS", details)
except error.HTTPError as exc:
@@ -248,7 +557,14 @@ def write_reports(results: List[ScenarioResult], report_json: Path, report_md: P
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="RTC_BL_PHONE A252 validation runner")
parser.add_argument("--port-a252", required=True, help="serial port for A252 target")
parser.add_argument(
"--port-a252",
default="",
help=(
"serial port for A252 target. If omitted, auto-detects first /dev/*usbserial* on macOS/"
"Linux."
),
)
parser.add_argument("--baud", type=int, default=115200, help="serial baudrate")
parser.add_argument("--flash", action="store_true", help="build and upload firmware before tests")
parser.add_argument("--base-url", default="", help="optional A252 base URL (http://ip)")
@@ -263,23 +579,64 @@ def parse_args() -> argparse.Namespace:
parser.add_argument("--manual-audio", default="MANUAL_SKIP", choices=sorted(VALID_STATES))
parser.add_argument("--manual-hfp", default="MANUAL_SKIP", choices=sorted(VALID_STATES))
parser.add_argument("--manual-note", default="", help="optional shared note for manual checks")
parser.add_argument(
"--strict-serial-smoke",
action=argparse.BooleanOptionalAction,
default=True,
help="if enabled, fail serial_smoke when required command checks fail (default: enabled)",
)
parser.add_argument(
"--allow-capture-fail-when-disabled",
action=argparse.BooleanOptionalAction,
default=True,
help=(
"if enabled, CAPTURE_START failure is tolerated when STATUS reports audio.enable_capture=false "
"(default: enabled)"
),
)
parser.add_argument(
"--require-hook-toggle",
action=argparse.BooleanOptionalAction,
default=True,
help="if enabled, serial_hook_ring_audio requires both ON_HOOK and OFF_HOOK states",
)
parser.add_argument(
"--hook-observe-seconds",
type=int,
default=45,
help="hook observation window in seconds for serial_hook_ring_audio (default: 45)",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
resolved_port = resolve_a252_port(args.port_a252.strip() or None)
if args.flash:
run_cmd(["pio", "run", "-e", "esp32dev"])
run_cmd(["pio", "run", "-e", "esp32dev", "-t", "upload", "--upload-port", args.port_a252])
run_cmd(["pio", "run", "-e", "esp32dev", "-t", "upload", "--upload-port", resolved_port])
results: List[ScenarioResult] = []
network_result: Optional[ScenarioResult] = None
try:
with SerialEndpoint(args.port_a252, args.baud) as dev:
with SerialEndpoint(resolved_port, args.baud) as dev:
dev.sync()
results.append(scenario_serial_smoke(dev))
results.append(
scenario_serial_smoke(
dev,
strict_serial_smoke=args.strict_serial_smoke,
allow_capture_fail_when_disabled=args.allow_capture_fail_when_disabled,
)
)
results.append(
scenario_serial_hook_ring_audio(
dev,
require_hook_toggle=args.require_hook_toggle,
hook_observe_seconds=args.hook_observe_seconds,
)
)
network_result = scenario_serial_network(dev, args.wifi_ssid, args.wifi_password)
results.append(network_result)
except Exception as exc:
scripts/test_hw_validation_contracts.py
+199
View File
@@ -0,0 +1,199 @@
#!/usr/bin/env python3
"""Unit tests for hw_validation serial smoke contract behavior."""
from __future__ import annotations
import unittest
from scripts.hw_validation import (
evaluate_serial_smoke_contract,
scenario_serial_hook_ring_audio,
scenario_serial_network,
)
def _ack(ok: bool, command: str) -> dict[str, object]:
return {"ok": ok, "line": ("OK " if ok else "ERR ") + command}
def _make_serial_details(
*,
enable_capture: bool,
capture_start_ok: bool,
call_ok: bool = True,
ping_ok: bool = True,
) -> dict[str, object]:
return {
"ping": {"ok": ping_ok, "result": "PONG" if ping_ok else "ERR"},
"status": {
"telephony": {"state": "IDLE"},
"config": {
"audio": {
"enable_capture": enable_capture,
}
},
},
"call": _ack(call_ok, "CALL"),
"capture_start": _ack(capture_start_ok, "CAPTURE_START"),
"capture_stop": _ack(True, "CAPTURE_STOP"),
"reset_metrics": _ack(True, "RESET_METRICS"),
}
class SerialSmokeContractTest(unittest.TestCase):
def test_capture_start_failure_fails_when_capture_enabled(self) -> None:
details = _make_serial_details(enable_capture=True, capture_start_ok=False)
state, required_checks, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=True,
allow_capture_fail_when_disabled=True,
)
self.assertEqual(state, "FAIL")
self.assertIn("CAPTURE_START", required_checks)
self.assertIn("CAPTURE_START", failed_checks)
self.assertEqual(warnings, [])
def test_capture_start_failure_passes_with_warning_when_capture_disabled(self) -> None:
details = _make_serial_details(enable_capture=False, capture_start_ok=False)
state, required_checks, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=True,
allow_capture_fail_when_disabled=True,
)
self.assertEqual(state, "PASS")
self.assertNotIn("CAPTURE_START", required_checks)
self.assertNotIn("CAPTURE_START", failed_checks)
self.assertIn("capture_start_failed_capture_disabled", warnings)
def test_capture_start_failure_can_be_forced_to_fail_when_capture_disabled(self) -> None:
details = _make_serial_details(enable_capture=False, capture_start_ok=False)
state, required_checks, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=True,
allow_capture_fail_when_disabled=False,
)
self.assertEqual(state, "FAIL")
self.assertIn("CAPTURE_START", required_checks)
self.assertIn("CAPTURE_START", failed_checks)
self.assertIn("capture_start_required_even_when_capture_disabled", warnings)
def test_non_strict_mode_warns_but_can_pass_non_critical_failures(self) -> None:
details = _make_serial_details(enable_capture=True, capture_start_ok=True, call_ok=False)
state, _, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=False,
allow_capture_fail_when_disabled=True,
)
self.assertEqual(state, "PASS")
self.assertIn("CALL", failed_checks)
self.assertIn("strict_serial_smoke_disabled", warnings)
def test_non_strict_mode_still_fails_on_minimum_contract(self) -> None:
details = _make_serial_details(enable_capture=True, capture_start_ok=True, ping_ok=False)
state, _, failed_checks, warnings = evaluate_serial_smoke_contract(
details,
strict_serial_smoke=False,
allow_capture_fail_when_disabled=True,
)
self.assertEqual(state, "FAIL")
self.assertIn("PING", failed_checks)
self.assertNotIn("strict_serial_smoke_disabled", warnings)
class _FakeSerialEndpoint:
def __init__(self, responses: list[object]) -> None:
self._responses = responses
self._index = 0
def command(self, cmd: str, timeout_s: float = 6.0, expect: str = "any") -> dict[str, object]:
(timeout_s, expect) # silence unused
if self._index >= len(self._responses):
raise RuntimeError(f"no response configured for command: {cmd}")
value = self._responses[self._index]
self._index += 1
if isinstance(value, Exception):
raise value
return value
class SerialNetworkContractTest(unittest.TestCase):
def test_wifi_connect_failure_marks_network_fail(self) -> None:
fake = _FakeSerialEndpoint(
[
{"connected": False, "ssid": "", "status": 6},
[{"ssid": "Les cils", "rssi": -42, "chan": 11, "enc": 4}],
{"ok": False, "line": "ERR WIFI_CONNECT failed"},
{"connected": False, "ssid": "", "status": 6},
{"ready": True, "peer_count": 1},
]
)
result = scenario_serial_network(fake, "Les cils", "mascarade")
self.assertEqual(result.state, "FAIL")
self.assertEqual(result.name, "serial_network_stack")
self.assertEqual(result.details.get("wifi_connect", {}).get("ok"), False)
def test_wifi_connect_success_marks_network_pass(self) -> None:
fake = _FakeSerialEndpoint(
[
{"connected": False, "ssid": "", "status": 6},
[{"ssid": "Les cils", "rssi": -42, "chan": 11, "enc": 4}],
{"ok": True, "line": "OK WIFI_CONNECT"},
{"connected": True, "ssid": "Les cils", "status": 3, "ip": "192.168.1.42"},
{"ready": True, "peer_count": 1},
]
)
result = scenario_serial_network(fake, "Les cils", "mascarade")
self.assertEqual(result.state, "PASS")
self.assertEqual(result.name, "serial_network_stack")
class SerialHookRingAudioContractTest(unittest.TestCase):
def test_tone_on_failure_marks_hook_ring_audio_fail(self) -> None:
fake = _FakeSerialEndpoint(
[
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
{"ok": True, "line": "OK RING"},
{"ok": False, "line": "ERR TONE_ON audio_not_ready"},
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
{"ok": True, "line": "OK TONE_OFF"},
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
]
)
result = scenario_serial_hook_ring_audio(fake, require_hook_toggle=True, hook_observe_seconds=0)
self.assertEqual(result.state, "FAIL")
self.assertEqual(result.name, "serial_hook_ring_audio")
self.assertEqual(result.details.get("checks", {}).get("tone_on_ok"), False)
def test_hook_toggle_is_required_when_enabled(self) -> None:
fake = _FakeSerialEndpoint(
[
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
{"ok": True, "line": "OK RING"},
{"ok": True, "line": "OK TONE_ON"},
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": True}},
{"ok": True, "line": "OK TONE_OFF"},
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
]
)
result = scenario_serial_hook_ring_audio(fake, require_hook_toggle=True, hook_observe_seconds=0)
self.assertEqual(result.state, "FAIL")
self.assertEqual(result.details.get("checks", {}).get("hook_ok"), False)
def test_hook_toggle_passes_when_both_states_are_seen(self) -> None:
fake = _FakeSerialEndpoint(
[
{"telephony": {"hook": "ON_HOOK"}, "audio": {"dial_tone_active": False}},
{"ok": True, "line": "OK RING"},
{"ok": True, "line": "OK TONE_ON"},
{"telephony": {"hook": "OFF_HOOK"}, "audio": {"dial_tone_active": True}},
{"ok": True, "line": "OK TONE_OFF"},
{"telephony": {"hook": "OFF_HOOK"}, "audio": {"dial_tone_active": False}},
]
)
result = scenario_serial_hook_ring_audio(fake, require_hook_toggle=True, hook_observe_seconds=0)
self.assertEqual(result.state, "PASS")
self.assertEqual(result.name, "serial_hook_ring_audio")
if __name__ == "__main__":
unittest.main()
scripts/zeroclaw_orchestrator_health.py
Executable
+101
View File
@@ -0,0 +1,101 @@
#!/usr/bin/env python3
"""ZeroClaw Docker orchestrator health check for A252 strict gate."""
from __future__ import annotations
import argparse
import json
import os
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Dict, Optional
from urllib import error, request
@dataclass
class StepResult:
name: str
ok: bool
details: Dict[str, Any]
def ensure_parent(path: Path) -> None:
path.parent.mkdir(parents=True, exist_ok=True)
def request_json(method: str, url: str, payload: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
body = None
headers: Dict[str, str] = {}
if payload is not None:
body = json.dumps(payload).encode("utf-8")
headers["Content-Type"] = "application/json"
req = request.Request(url, method=method, data=body, headers=headers)
with request.urlopen(req, timeout=15) as resp:
raw = resp.read().decode("utf-8")
if not raw:
return {}
return json.loads(raw)
def run_step(name: str, method: str, url: str, payload: Optional[Dict[str, Any]] = None) -> StepResult:
try:
data = request_json(method, url, payload)
return StepResult(name=name, ok=True, details={"url": url, "response": data})
except error.HTTPError as exc:
details: Dict[str, Any] = {"url": url, "error": f"HTTP {exc.code}"}
try:
details["body"] = exc.read().decode("utf-8")
except Exception:
details["body"] = ""
return StepResult(name=name, ok=False, details=details)
except Exception as exc:
return StepResult(name=name, ok=False, details={"url": url, "error": str(exc)})
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description="ZeroClaw orchestrator health check")
parser.add_argument(
"--base-url",
default=os.environ.get("ZEROCLAW_ORCH", "http://127.0.0.1:8788"),
help="ZeroClaw orchestrator base URL (default: $ZEROCLAW_ORCH or http://127.0.0.1:8788)",
)
parser.add_argument(
"--report-json",
default="artifacts/zeroclaw_orchestrator_health.json",
help="JSON report output path",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
base_url = args.base_url.rstrip("/")
steps = [
run_step("status", "GET", f"{base_url}/api/status"),
run_step("agents", "GET", f"{base_url}/api/agents"),
run_step("workflows", "GET", f"{base_url}/api/workflows"),
run_step("provider_scan", "POST", f"{base_url}/api/run", {"action": "provider_scan"}),
]
overall_ok = all(step.ok for step in steps)
report = {
"timestamp_utc": datetime.now(timezone.utc).isoformat(),
"base_url": base_url,
"overall_passed": overall_ok,
"results": [
{"name": step.name, "state": "PASS" if step.ok else "FAIL", "details": step.details} for step in steps
],
}
report_path = Path(args.report_json)
ensure_parent(report_path)
report_path.write_text(json.dumps(report, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
return 0 if overall_ok else 1
if __name__ == "__main__":
raise SystemExit(main())
src/audio/AudioEngine.cpp
+16 -1
View File
@@ -501,7 +501,18 @@ bool AudioEngine::begin(const AudioConfig& config) {
i2s_cfg.auto_clear = true;
if (!i2s_stream_.begin(i2s_cfg)) {
Serial.println("[AudioEngine] i2s begin failed");
Serial.printf("[AudioEngine] i2s begin failed: port=%d mode=%d sr=%u bits=%d ch=%d bck=%d ws=%d dout=%d din=%d dma_cnt=%u dma_len=%u\n",
static_cast<int>(i2s_cfg.port_no),
static_cast<int>(mode),
static_cast<unsigned>(i2s_cfg.sample_rate),
static_cast<int>(i2s_cfg.bits_per_sample),
static_cast<int>(i2s_cfg.channels),
static_cast<int>(i2s_cfg.pin_bck),
static_cast<int>(i2s_cfg.pin_ws),
static_cast<int>(i2s_cfg.pin_data),
static_cast<int>(i2s_cfg.pin_data_rx),
static_cast<unsigned>(i2s_cfg.buffer_count),
static_cast<unsigned>(i2s_cfg.buffer_size));
driver_installed_ = false;
return false;
}
@@ -847,6 +858,10 @@ bool AudioEngine::isSdReady() const {
return audio_fs_ready_;
}
bool AudioEngine::isReady() const {
return driver_installed_;
}
AudioRuntimeMetrics AudioEngine::metrics() const {
return metrics_;
}
src/audio/AudioEngine.h
+1
View File
@@ -72,6 +72,7 @@ public:
virtual bool supportsFullDuplex() const;
virtual bool isPlaying() const;
virtual bool isSdReady() const;
virtual bool isReady() const;
virtual AudioRuntimeMetrics metrics() const;
virtual void resetMetrics();
virtual void tick();
src/config/A252ConfigStore.cpp
+19 -90
View File
@@ -9,7 +9,6 @@
namespace {
constexpr const char* kPinsNs = "a252-pins";
constexpr const char* kAudioNs = "a252-audio";
constexpr const char* kMqttNs = "mqtt";
constexpr const char* kEspNowNs = "espnow";
constexpr const char* kEspNowCallMapNs = "espnow-call";
@@ -89,10 +88,6 @@ A252AudioConfig A252ConfigStore::defaultAudio() {
return A252AudioConfig{};
}
MqttConfig A252ConfigStore::defaultMqtt() {
return MqttConfig{};
}
bool A252ConfigStore::loadPins(A252PinsConfig& out) {
out = defaultPins();
Preferences prefs;
@@ -233,64 +228,6 @@ bool A252ConfigStore::saveAudio(const A252AudioConfig& cfg, String* error) {
return true;
}
bool A252ConfigStore::loadMqtt(MqttConfig& out) {
out = defaultMqtt();
Preferences prefs;
if (!prefs.begin(kMqttNs, false)) {
return false;
}
out.enabled = prefs.getBool("enabled", out.enabled);
if (prefs.isKey("host")) {
out.host = prefs.getString("host", out.host);
}
out.port = static_cast<uint16_t>(prefs.getUShort("port", out.port));
if (prefs.isKey("user")) {
out.user = prefs.getString("user", out.user);
}
if (prefs.isKey("pass")) {
out.pass = prefs.getString("pass", out.pass);
}
if (prefs.isKey("topic")) {
out.base_topic = prefs.getString("topic", out.base_topic);
}
prefs.end();
String error;
if (!validateMqtt(out, error)) {
out = defaultMqtt();
return false;
}
return true;
}
bool A252ConfigStore::saveMqtt(const MqttConfig& cfg, String* error) {
String local_error;
if (!validateMqtt(cfg, local_error)) {
if (error) {
*error = local_error;
}
return false;
}
Preferences prefs;
if (!prefs.begin(kMqttNs, false)) {
if (error) {
*error = "nvs_open_failed";
}
return false;
}
prefs.putBool("enabled", cfg.enabled);
saveString(prefs, "host", cfg.host);
prefs.putUShort("port", cfg.port);
saveString(prefs, "user", cfg.user);
saveString(prefs, "pass", cfg.pass);
saveString(prefs, "topic", cfg.base_topic);
prefs.end();
return true;
}
bool A252ConfigStore::loadEspNowPeers(EspNowPeerStore& out) {
out.peers.clear();
@@ -406,7 +343,7 @@ bool A252ConfigStore::saveEspNowPeers(const EspNowPeerStore& store, String* erro
bool A252ConfigStore::validatePins(const A252PinsConfig& cfg, String& error) {
std::vector<int> used;
used.reserve(11);
used.reserve(14);
const int required_pins[] = {
cfg.i2s_bck,
@@ -417,6 +354,9 @@ bool A252ConfigStore::validatePins(const A252PinsConfig& cfg, String& error) {
cfg.slic_fr,
cfg.slic_shk,
cfg.slic_pd,
};
const int optional_pins[] = {
cfg.slic_adc_in,
cfg.pcm_flt,
cfg.pcm_demp,
@@ -436,6 +376,21 @@ bool A252ConfigStore::validatePins(const A252PinsConfig& cfg, String& error) {
used.push_back(pin);
}
for (int pin : optional_pins) {
if (pin == -1) {
continue;
}
if (pin < 0 || pin > 39) {
error = "invalid_pin_range";
return false;
}
if (std::find(used.begin(), used.end(), pin) != used.end()) {
error = "pin_conflict";
return false;
}
used.push_back(pin);
}
if (detectBoardProfile() == BoardProfile::ESP32_A252) {
if (cfg.es8388_sda < 0 || cfg.es8388_scl < 0) {
error = "invalid_pin_range";
@@ -526,21 +481,6 @@ bool A252ConfigStore::validateAudio(const A252AudioConfig& cfg, String& error) {
return true;
}
bool A252ConfigStore::validateMqtt(const MqttConfig& cfg, String& error) {
if (cfg.port == 0) {
error = "invalid_port";
return false;
}
if (cfg.base_topic.isEmpty()) {
error = "invalid_base_topic";
return false;
}
error = "";
return true;
}
void A252ConfigStore::pinsToJson(const A252PinsConfig& cfg, JsonObject obj) {
JsonObject i2s = obj["i2s"].to<JsonObject>();
i2s["bck"] = cfg.i2s_bck;
@@ -582,17 +522,6 @@ void A252ConfigStore::audioToJson(const A252AudioConfig& cfg, JsonObject obj) {
obj["route"] = cfg.route;
}
void A252ConfigStore::mqttToJson(const MqttConfig& cfg, JsonObject obj, bool include_secret) {
obj["enabled"] = cfg.enabled;
obj["host"] = cfg.host;
obj["port"] = cfg.port;
obj["user"] = cfg.user;
obj["base_topic"] = cfg.base_topic;
if (include_secret) {
obj["pass"] = cfg.pass;
}
}
void A252ConfigStore::peersToJson(const EspNowPeerStore& store, JsonArray arr) {
for (const String& peer : store.peers) {
arr.add(peer);
src/main.cpp
+53 -162
View File
@@ -9,7 +9,6 @@
#include "core/CommandDispatcher.h"
#include "core/PlatformProfile.h"
#include "props/EspNowBridge.h"
#include "props/PropsBridge.h"
#include "slic/Ks0835SlicController.h"
#include "telephony/TelephonyService.h"
#include "web/WebServerManager.h"
@@ -25,7 +24,7 @@ constexpr uint32_t kSerialBaud = 115200;
constexpr int kAudioAmpEnablePin = 21;
constexpr char kBootLogTag[] = "RTC_BOOT";
constexpr bool kPrintHelpOnBoot = false;
// API web access is intentionally left open by default to avoid hard lock on Wi-Fi.
// Branch lock: API web access remains open (no Wi-Fi basic auth) for this flow.
constexpr bool kWebAuthEnabledByDefault = false;
#ifdef RTC_WEB_AUTH_DEV_DISABLE
@@ -39,7 +38,6 @@ FeatureMatrix g_features = getFeatureMatrix(g_profile);
A252PinsConfig g_pins_cfg = A252ConfigStore::defaultPins();
A252AudioConfig g_audio_cfg = A252ConfigStore::defaultAudio();
MqttConfig g_mqtt_cfg = A252ConfigStore::defaultMqtt();
EspNowPeerStore g_peer_store;
EspNowCallMap g_espnow_call_map;
String g_active_scene_id;
@@ -51,12 +49,20 @@ AudioEngine g_audio;
Es8388Driver g_codec;
TelephonyService g_telephony;
EspNowBridge g_espnow;
PropsBridge g_props_bridge;
CommandDispatcher g_dispatcher;
ScopeDisplay g_scope_display;
String g_serial_line;
WebServerManager g_web_server;
struct HardwareInitStatus {
bool init_ok = false;
bool slic_ready = false;
bool codec_ready = false;
bool audio_ready = false;
};
HardwareInitStatus g_hw_status;
DispatchResponse makeResponse(bool ok, const String& code) {
DispatchResponse res;
res.ok = ok;
@@ -466,6 +472,7 @@ void applyPcm5102ControlPins(const A252PinsConfig& pins_cfg) {
}
bool applyHardwareConfig() {
g_hw_status = HardwareInitStatus{};
String pin_validation_error;
if (!A252ConfigStore::validatePins(g_pins_cfg, pin_validation_error)) {
Serial.printf("[RTC_BL_PHONE] invalid pins configuration: %s\n", pin_validation_error.c_str());
@@ -496,7 +503,11 @@ bool applyHardwareConfig() {
applyPcm5102ControlPins(g_pins_cfg);
const AudioConfig audio = buildI2sConfig(g_pins_cfg, g_audio_cfg);
const bool audio_ok = g_audio.begin(audio);
bool audio_ok = g_audio.begin(audio);
if (!audio_ok) {
Serial.println("[RTC_BL_PHONE] audio init failed, retrying once");
audio_ok = g_audio.begin(audio);
}
g_audio.resetMetrics();
g_telephony.begin(g_profile, g_slic, g_audio);
@@ -521,12 +532,18 @@ bool applyHardwareConfig() {
return ok;
});
Serial.printf("[RTC_BL_PHONE] HW init slic=%s codec=%s audio=%s\n",
g_hw_status.slic_ready = slic_ok;
g_hw_status.codec_ready = codec_ok;
g_hw_status.audio_ready = audio_ok;
g_hw_status.init_ok = slic_ok && codec_ok && audio_ok;
Serial.printf("[RTC_BL_PHONE] HW init slic=%s codec=%s audio=%s init=%s\n",
slic_ok ? "ok" : "fail",
codec_ok ? "ok" : "fail",
audio_ok ? "ok" : "fail");
audio_ok ? "ok" : "fail",
g_hw_status.init_ok ? "ok" : "fail");
return slic_ok && codec_ok && audio_ok;
return g_hw_status.init_ok;
}
void appendAudioMetrics(JsonObject root) {
@@ -541,6 +558,7 @@ void appendAudioMetrics(JsonObject root) {
JsonObject audio = root["audio"].to<JsonObject>();
audio["full_duplex"] = g_audio.supportsFullDuplex();
audio["ready"] = g_audio.isReady();
audio["dial_tone_active"] = g_audio.isDialToneActive();
audio["playing"] = g_audio.isPlaying();
audio["sd_ready"] = g_audio.isSdReady();
@@ -574,9 +592,11 @@ void fillStatusSnapshot(JsonObject root) {
JsonObject espnow = root["espnow"].to<JsonObject>();
g_espnow.statusToJson(espnow);
JsonObject mqtt = root["mqtt"].to<JsonObject>();
g_props_bridge.statusToJson(mqtt);
A252ConfigStore::mqttToJson(g_mqtt_cfg, mqtt["config"].to<JsonObject>());
JsonObject hw = root["hw"].to<JsonObject>();
hw["init_ok"] = g_hw_status.init_ok;
hw["slic_ready"] = g_hw_status.slic_ready;
hw["codec_ready"] = g_hw_status.codec_ready;
hw["audio_ready"] = g_hw_status.audio_ready;
JsonObject config = root["config"].to<JsonObject>();
A252ConfigStore::pinsToJson(g_pins_cfg, config["pins"].to<JsonObject>());
@@ -762,41 +782,6 @@ bool applyAudioPatch(JsonVariantConst patch, A252AudioConfig& target, String& er
return true;
}
bool applyMqttPatch(JsonVariantConst patch, MqttConfig& target, String& error) {
MqttConfig next = target;
if (patch["enabled"].is<bool>()) {
next.enabled = patch["enabled"].as<bool>();
}
if (patch["host"].is<const char*>()) {
next.host = patch["host"].as<const char*>();
}
if (patch["port"].is<uint16_t>()) {
next.port = patch["port"].as<uint16_t>();
} else if (patch["port"].is<uint32_t>()) {
const uint32_t port = patch["port"].as<uint32_t>();
if (port > UINT16_MAX) {
error = "MQTT_CONFIG_SET invalid_port";
return false;
}
next.port = static_cast<uint16_t>(port);
}
if (patch["user"].is<const char*>()) {
next.user = patch["user"].as<const char*>();
}
if (patch["pass"].is<const char*>()) {
next.pass = patch["pass"].as<const char*>();
}
if (patch["base_topic"].is<const char*>()) {
next.base_topic = patch["base_topic"].as<const char*>();
}
if (!A252ConfigStore::validateMqtt(next, error)) {
return false;
}
target = next;
return true;
}
DispatchResponse applyEspNowCallMapSet(const String& args) {
if (args.isEmpty()) {
return makeResponse(false, "ESPNOW_CALL_MAP_SET invalid_json");
@@ -898,10 +883,12 @@ void registerCommands() {
root["connected"] = connected;
root["status"] = status;
if (connected) {
root["ssid"] = WiFi.SSID();
root["ip"] = WiFi.localIP().toString();
root["rssi"] = WiFi.RSSI();
root["channel"] = WiFi.channel();
} else {
root["ssid"] = "";
root["ip"] = "";
root["rssi"] = 0;
root["channel"] = 0;
@@ -918,13 +905,20 @@ void registerCommands() {
g_dispatcher.registerCommand("WIFI_CONNECT", [](const String& args) {
String ssid;
String password;
if (!splitFirstToken(args, ssid, password)) {
String rest;
if (!splitFirstToken(args, ssid, rest)) {
return makeResponse(false, "WIFI_CONNECT invalid_args");
}
if (ssid.isEmpty()) {
return makeResponse(false, "WIFI_CONNECT invalid_ssid");
}
String password;
if (!rest.isEmpty()) {
String trailing;
if (!splitFirstToken(rest, password, trailing) || !trailing.isEmpty()) {
return makeResponse(false, "WIFI_CONNECT invalid_args");
}
}
const bool ok = g_wifi.connect(ssid, password);
return makeResponse(ok, ok ? "WIFI_CONNECT" : "WIFI_CONNECT failed");
});
@@ -946,112 +940,6 @@ void registerCommands() {
return makeResponse(ok, ok ? "WIFI_RECONNECT" : "WIFI_RECONNECT no_credentials");
});
g_dispatcher.registerCommand("MQTT_STATUS", [](const String&) {
JsonDocument doc;
JsonObject root = doc.to<JsonObject>();
g_props_bridge.statusToJson(root["bridge"].to<JsonObject>());
A252ConfigStore::mqttToJson(g_mqtt_cfg, root["config"].to<JsonObject>());
return jsonResponse(doc);
});
g_dispatcher.registerCommand("MQTT_CONNECT", [](const String&) {
if (!g_mqtt_cfg.enabled) {
return makeResponse(false, "MQTT_CONNECT disabled");
}
const bool ok = g_props_bridge.connectNow();
return makeResponse(ok, ok ? "MQTT_CONNECT" : "MQTT_CONNECT failed");
});
g_dispatcher.registerCommand("MQTT_DISCONNECT", [](const String&) {
g_props_bridge.disconnect();
return makeResponse(true, "MQTT_DISCONNECT");
});
const auto mqttPublishHandler = [](const String& args) {
String topic;
String payload;
if (!splitFirstToken(args, topic, payload) || topic.isEmpty()) {
return makeResponse(false, "MQTT_PUBLISH invalid_args");
}
return makeResponse(g_props_bridge.publish(topic, payload), "MQTT_PUB");
};
g_dispatcher.registerCommand("MQTT_PUB", mqttPublishHandler);
g_dispatcher.registerCommand("MQTT_PUBLISH", mqttPublishHandler);
g_dispatcher.registerCommand("MQTT_CONFIG_SET", [](const String& args) {
if (args.isEmpty()) {
return makeResponse(false, "MQTT_CONFIG_SET invalid_json");
}
JsonDocument doc;
if (deserializeJson(doc, args) != DeserializationError::Ok || !doc.is<JsonObject>()) {
return makeResponse(false, "MQTT_CONFIG_SET invalid_json");
}
MqttConfig next = g_mqtt_cfg;
String error;
if (!applyMqttPatch(doc.as<JsonVariantConst>(), next, error)) {
return makeResponse(false, "MQTT_CONFIG_SET " + error);
}
if (!A252ConfigStore::saveMqtt(next, &error)) {
return makeResponse(false, "MQTT_CONFIG_SET " + error);
}
g_mqtt_cfg = next;
g_props_bridge.setConfig(g_mqtt_cfg);
if (!g_mqtt_cfg.enabled) {
return makeResponse(true, "MQTT_CONFIG_SET disabled");
}
const bool connected = g_props_bridge.connectNow();
return makeResponse(connected, connected ? "MQTT_CONFIG_SET" : "MQTT_CONFIG_SET connect_failed");
});
g_dispatcher.registerCommand("BT_STATUS", [](const String&) {
return makeResponse(false, "BT_STATUS unsupported");
});
g_dispatcher.registerCommand("BT_HFP_CONNECT", [](const String&) {
return makeResponse(false, "BT_HFP_CONNECT unsupported");
});
g_dispatcher.registerCommand("BT_HFP_DISCONNECT", [](const String&) {
return makeResponse(false, "BT_HFP_DISCONNECT unsupported");
});
g_dispatcher.registerCommand("BT_AUTO_RECONNECT_ON", [](const String&) {
return makeResponse(false, "BT_AUTO_RECONNECT_ON unsupported");
});
g_dispatcher.registerCommand("BT_AUTO_RECONNECT_OFF", [](const String&) {
return makeResponse(false, "BT_AUTO_RECONNECT_OFF unsupported");
});
g_dispatcher.registerCommand("BT_DISCOVERABLE_ON", [](const String&) {
return makeResponse(false, "BT_DISCOVERABLE_ON unsupported");
});
g_dispatcher.registerCommand("BT_DISCOVERABLE_OFF", [](const String&) {
return makeResponse(false, "BT_DISCOVERABLE_OFF unsupported");
});
g_dispatcher.registerCommand("BT_DIAL", [](const String&) {
return makeResponse(false, "BT_DIAL unsupported");
});
g_dispatcher.registerCommand("BT_REDIAL", [](const String&) {
return makeResponse(false, "BT_REDIAL unsupported");
});
g_dispatcher.registerCommand("BT_ANSWER", [](const String&) {
return makeResponse(false, "BT_ANSWER unsupported");
});
g_dispatcher.registerCommand("BT_HANGUP", [](const String&) {
return makeResponse(false, "BT_HANGUP unsupported");
});
g_dispatcher.registerCommand("BT_CALLS", [](const String&) {
return makeResponse(false, "BT_CALLS unsupported");
});
g_dispatcher.registerCommand("BT_PBAP_SYNC", [](const String&) {
return makeResponse(false, "BT_PBAP_SYNC unsupported");
});
g_dispatcher.registerCommand("BT_BLE_START", [](const String&) {
return makeResponse(false, "BT_BLE_START unsupported");
});
g_dispatcher.registerCommand("BT_BLE_STOP", [](const String&) {
return makeResponse(false, "BT_BLE_STOP unsupported");
});
g_dispatcher.registerCommand("UNLOCK", [](const String&) {
g_slic.setLineEnabled(true);
return makeResponse(true, "UNLOCK");
@@ -1155,7 +1043,11 @@ void registerCommands() {
});
g_dispatcher.registerCommand("TONE_ON", [](const String&) {
return makeResponse(g_audio.startDialTone(), "TONE_ON");
if (!g_audio.isReady()) {
return makeResponse(false, "TONE_ON audio_not_ready");
}
const bool ok = g_audio.startDialTone();
return makeResponse(ok, ok ? "TONE_ON" : "TONE_ON failed");
});
g_dispatcher.registerCommand("TONE_OFF", [](const String&) {
@@ -1319,6 +1211,8 @@ void registerCommands() {
g_codec.setRoute(g_audio_cfg.route);
}
const bool audio_ok = g_audio.begin(buildI2sConfig(g_pins_cfg, g_audio_cfg));
g_hw_status.audio_ready = audio_ok;
g_hw_status.init_ok = g_hw_status.slic_ready && g_hw_status.codec_ready && g_hw_status.audio_ready;
return makeResponse(audio_ok, "AUDIO_CONFIG_SET");
});
}
@@ -1532,7 +1426,6 @@ void setup() {
A252ConfigStore::loadPins(g_pins_cfg);
g_pins_cfg.slic_line = -1;
A252ConfigStore::loadAudio(g_audio_cfg);
A252ConfigStore::loadMqtt(g_mqtt_cfg);
A252ConfigStore::loadEspNowPeers(g_peer_store);
initDefaultEspNowCallMap(g_espnow_call_map);
if (!A252ConfigStore::loadEspNowCallMap(g_espnow_call_map)) {
@@ -1543,17 +1436,16 @@ void setup() {
pinMode(kAudioAmpEnablePin, OUTPUT);
digitalWrite(kAudioAmpEnablePin, LOW);
applyHardwareConfig();
const bool hw_init_ok = applyHardwareConfig();
if (!hw_init_ok) {
Serial.println("[RTC_BL_PHONE] hardware init failed");
}
registerCommands();
g_espnow.begin(g_peer_store);
g_espnow.setCommandCallback([](const String& source, const JsonVariantConst& payload) {
processInboundBridgeCommand(source, payload);
});
g_props_bridge.setCommandCallback([](const String& source, const JsonVariantConst& payload) {
processInboundBridgeCommand(source, payload);
});
g_props_bridge.begin(g_mqtt_cfg);
configureCommandServer();
g_web_server.begin();
@@ -1570,7 +1462,6 @@ void loop() {
g_telephony.tick();
g_scope_display.tick();
g_web_server.handle();
g_props_bridge.tick();
g_espnow.tick();
pollSerial();
delay(1);