feat: implement set_project_info functionality and add POC test script

This commit is contained in:
2026-04-07 02:49:17 -04:00
parent bcc044655a
commit 7005ae766d
4 changed files with 283 additions and 1 deletions
+77 -1
View File
@@ -40,9 +40,10 @@ from typing import Optional
# FastAPI / Pydantic
try:
from fastapi import FastAPI, HTTPException, Query
from fastapi import Body, FastAPI, HTTPException, Query
from fastapi.middleware.cors import CORSMiddleware
from fastapi.responses import JSONResponse
from pydantic import BaseModel
import uvicorn
except ImportError:
print(
@@ -475,6 +476,81 @@ def device_event_waveform(
}
# ── Write endpoints ───────────────────────────────────────────────────────────
class ProjectInfoBody(BaseModel):
"""Request body for POST /device/config/project."""
project: Optional[str] = None
client_name: Optional[str] = None
operator: Optional[str] = None
seis_loc: Optional[str] = None
notes: Optional[str] = None
@app.post("/device/config/project")
def device_config_project(
body: ProjectInfoBody,
port: Optional[str] = Query(None, description="Serial port (e.g. COM5)"),
baud: int = Query(38400, description="Serial baud rate"),
host: Optional[str] = Query(None, description="TCP host — modem IP or ACH relay"),
tcp_port: int = Query(DEFAULT_TCP_PORT, description=f"TCP port (default {DEFAULT_TCP_PORT})"),
) -> dict:
"""
POC: Read the current device config, patch ASCII project/client/operator/
sensor-location/notes fields, and write the modified config back.
Only fields included in the JSON body (non-null) are modified. All other
bytes are round-tripped verbatim from the device.
Supply either *port* (serial) or *host* (TCP/modem).
Example body:
{
"project": "Bridge Inspection 2026",
"client_name": "City of Portland",
"operator": "Brian Harrison",
"seis_loc": "South Abutment",
"notes": "Pre-blast baseline"
}
Returns:
{"status": "ok", "message": "..."} on success.
Raises:
502 on protocol errors (timeout, bad ack, etc.).
422 if neither port nor host is provided.
"""
log.info(
"POST /device/config/project port=%s host=%s body=%s",
port, host, body.model_dump(exclude_none=True),
)
try:
def _do():
with _build_client(port, baud, host, tcp_port) as client:
client.connect()
client.set_project_info(
project=body.project,
client_name=body.client_name,
operator=body.operator,
seis_loc=body.seis_loc,
notes=body.notes,
)
_run_with_retry(_do, is_tcp=_is_tcp(host))
except HTTPException:
raise
except ProtocolError as exc:
raise HTTPException(status_code=502, detail=f"Protocol error: {exc}") from exc
except OSError as exc:
raise HTTPException(status_code=502, detail=f"Connection error: {exc}") from exc
except Exception as exc:
raise HTTPException(status_code=500, detail=f"Device error: {exc}") from exc
changed = body.model_dump(exclude_none=True)
msg = "Config written. Fields updated: " + (", ".join(changed.keys()) or "none")
return {"status": "ok", "message": msg, "updated_fields": changed}
# ── Entry point ────────────────────────────────────────────────────────────────
if __name__ == "__main__":