API Reference

Complete API documentation auto-generated from code.

Routers

Main Router

Main API router aggregating all sub-routers.

This module serves as the entry point for all API endpoints, combining routers for history, test management, sensor operations, and graphique data visualization.

History Router

archive_history(name) async

Archive a test history (move to archived storage).

Source code in src/routers/history.py

@router.put("/{name}/archive", status_code=204, responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    }
})
async def archive_history(name: str) -> None:
    """
    Archive a test history (move to archived storage).
    """
    success = test_manager.archive_test(name)
    if not success:
        raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

delete_history(name) async

Permanently delete a test history.

Source code in src/routers/history.py

@router.delete("/{name}", status_code=204, responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    }
})
async def delete_history(name: str) -> None:
    """
    Permanently delete a test history.
    """
    success = test_manager.delete_test(name)
    if not success:
        raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

download_history(name) async

Download a test history as a ZIP file containing metadata, raw log, and CSV.

Source code in src/routers/history.py

@router.get("/{name}/download", responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    }
})
async def download_history(name: str):
    """
    Download a test history as a ZIP file containing metadata, raw log, and CSV.
    """
    from core.services.test_manager import TEST_DATA_DIR, ARCHIVE_DIR

    # Check if test exists in current or archived
    test_dir = os.path.join(TEST_DATA_DIR, name)
    if not os.path.exists(test_dir):
        test_dir = os.path.join(ARCHIVE_DIR, name)
        if not os.path.exists(test_dir):
            raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

    # Create ZIP in memory
    buf = io.BytesIO()
    with zipfile.ZipFile(buf, 'w', zipfile.ZIP_DEFLATED) as zf:
        for root, dirs, files in os.walk(test_dir):
            for file in files:
                file_path = os.path.join(root, file)
                # Use a path relative to the test directory itself so the ZIP
                # does not contain the top-level test folder when extracted.
                arcname = os.path.relpath(file_path, test_dir)
                zf.write(file_path, arcname)
    buf.seek(0)

    return StreamingResponse(
        buf, 
        media_type="application/zip", 
        headers={"Content-Disposition": f"attachment; filename=\"{name}.zip\""}
    )

get_description(name) async

Get the description.md content for a test history. Returns the markdown content as plain text.

Source code in src/routers/history.py

@router.get("/{name}/description", responses={
    404: {
        "description": "Test history or description not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Description not found for test test_123"}
            }
        }
    }
})
async def get_description(name: str) -> dict:
    """
    Get the description.md content for a test history.
    Returns the markdown content as plain text.
    """
    try:
        content = test_manager.get_description(name)
        return {"content": content}
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail=f"Description not found for test {name}")

get_history_metadata(name) async

Get metadata for a specific test history.

Source code in src/routers/history.py

@router.get("/{name}", response_model=TestMetaData, responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    }
})
async def get_history_metadata(name: str) -> TestMetaData:
    """
    Get metadata for a specific test history.
    """
    histories = test_manager.get_history()
    for h in histories:
        if h.test_id == name:
            return h
    raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

get_treatment(name) async

Get the treatment.json content for a test history. Returns the JSON content as a dictionary.

Source code in src/routers/history.py

@router.get("/{name}/treatment", responses={
    404: {
        "description": "Test history or treatment not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Treatment not found for test test_123"}
            }
        }
    }
})
async def get_treatment(name: str) -> dict:
    """
    Get the treatment.json content for a test history.
    Returns the JSON content as a dictionary.
    """
    content = test_manager.get_treatment_json(name)
    if content is not None:
        return content

    raise HTTPException(status_code=404, detail=f"Treatment not found for test {os.name}")

get_treatment_image(name) async

Get the treatment image for a test history. Returns the image as a PNG stream.

Source code in src/routers/history.py

@router.get("/{name}/treatment-image", responses={
    404: {
        "description": "Test history or treatment image not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Treatment image not found for test test_123"}
            }
        }
    }
})
async def get_treatment_image(name: str) -> dict:
    """
    Get the treatment image for a test history.
    Returns the image as a PNG stream.
    """
    png_bytes = test_manager.get_treatment_png_base64(name)
    if png_bytes is not None:
        return {
            "data": f"data:image/png;base64,{png_bytes}"
        }

    raise HTTPException(status_code=404, detail=f"Treatment image not found for test {name}")

list_histories() async

List all available test histories (sorted by date, newest first).

Source code in src/routers/history.py

@router.get("", response_model=HistoryList)
async def list_histories() -> HistoryList:
    """
    List all available test histories (sorted by date, newest first).
    """
    histories = test_manager.get_history()
    test_ids = [h.test_id for h in histories]
    return HistoryList(list=test_ids)

update_description(name, payload) async

Update the description.md content for a test history.

Request body:
```json
{
    "content": "# My Test Description

This is a markdown file." } ```

Source code in src/routers/history.py

@router.put("/{name}/description", status_code=204, responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    },
    500: {
        "description": "Failed to update description.",
        "content": {
            "application/json": {
                "example": {"detail": "Failed to update description: Permission denied"}
            }
        }
    }
})
async def update_description(name: str, payload: dict) -> None:
    """
    Update the description.md content for a test history.

    Request body:
    ```json
    {
        "content": "# My Test Description\n\nThis is a markdown file."
    }
    ```
    """
    content = payload.get("content", "")
    if not test_manager.set_description(name, content):
        raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

update_history_metadata(name, metadata) async

Update metadata for a test history.

Source code in src/routers/history.py

@router.put("/{name}", status_code=204, responses={
    404: {
        "description": "Test history not found.",
        "content": {
            "application/json": {
                "example": {"detail": "Test history 'test_123' not found"}
            }
        }
    },
    500: {
        "description": "Failed to update metadata.",
        "content": {
            "application/json": {
                "example": {"detail": "Failed to update metadata: Permission denied"}
            }
        }
    }
})
async def update_history_metadata(name: str, metadata: TestMetaData) -> None:
    """
    Update metadata for a test history.
    """
    from core.services.test_manager import TEST_DATA_DIR
    from dataclasses import asdict
    import json

    test_dir = os.path.join(TEST_DATA_DIR, name)
    if not os.path.exists(test_dir):
        raise HTTPException(status_code=404, detail=f"Test history '{name}' not found")

    metadata_file = os.path.join(test_dir, "metadata.json")
    try:
        with open(metadata_file, 'w') as f:
            json.dump(asdict(metadata), f, indent=2)
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Failed to update metadata: {str(e)}")

Test Router

add_file_to_current_test(file, filename) async

Add a file to the current test.

The file is saved in the current test directory and will be part of the test record. Available after POST /info has been called.

Request body:

{
    "file": "<base64-encoded file content>",
    "filename": "example.txt"
}

Source code in src/routers/test.py

@router.post("/files", status_code=204, responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }    },
    400: {
        "description": "Invalid base64 encoded file content.",
        "content": {
            "application/json": {
                "example": {"detail": "Invalid base64 encoded file content: Incorrect padding"}
            }
        }
    },
    500: {
        "description": "Failed to add file to test.",
        "content": {
            "application/json": {
                "example": {"detail": "Failed to add file to test: No test directory available."}
            }
        }
    }
})
async def add_file_to_current_test(file: bytes, filename: str) -> None:
    """
    Add a file to the current test.

    The file is saved in the current test directory and will be part of the test record.
    Available after POST /info has been called.

    Request body:
    ```json
    {
        "file": "<base64-encoded file content>",
        "filename": "example.txt"
    }
    ```
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    try:
        decoded_file = base64.b64decode(file)
    except Exception as e:
        raise HTTPException(status_code=400, detail=f"Invalid base64 encoded file content: {e}")

    success = test_manager.add_file(decoded_file, filename)
    if not success:
        raise HTTPException(status_code=500, detail="Failed to add file to test: No test directory available.")

delete_file_from_current_test(filename) async

Delete a file from the current test.

Removes the specified file from the test directory and the tracked file list. Available after POST /info has been called.

Source code in src/routers/test.py

@router.delete("/files/{filename}", status_code=204, responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    },
    404: {
        "description": "File not found in current test.",
        "content": {
            "application/json": {
                "example": {"detail": "File example.txt not found in current test."}
            }
        }
    }
})
async def delete_file_from_current_test(filename: str) -> None:
    """
    Delete a file from the current test.

    Removes the specified file from the test directory and the tracked file list.
    Available after POST /info has been called.
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    success = test_manager.delete_file(filename)
    if not success:
        raise HTTPException(status_code=404, detail=f"File {filename} not found in current test.")

finalize_test() async

Finalize a stopped test and move it to history.

Changes state from STOPPED to NOTHING. Test data is now part of the historical record and cannot be modified.

Requires that the test has been stopped via PUT /stop first.

Source code in src/routers/test.py

@router.put("/finalize", status_code=204, responses={
    400: {
        "description": "No test in STOPPED state to finalize.",
        "content": {
            "application/json": {
                "example": {"detail": "No test to finalize."}
            }
        }
    },
    409: {
        "description": "Test is still running. Call PUT /stop first.",
        "content": {
            "application/json": {
                "example": {"detail": "Test is not stopped. Call PUT /stop first."}
            }
        }
    }
})
async def finalize_test() -> None:
    """
    Finalize a stopped test and move it to history.

    Changes state from STOPPED to NOTHING.
    Test data is now part of the historical record and cannot be modified.

    Requires that the test has been stopped via PUT /stop first.
    """
    try:
        test_manager.finalize_test()
    except ValueError as e:
        # No test to finalize
        raise HTTPException(status_code=400, detail=str(e))
    except RuntimeError as e:
        # Test is not stopped
        raise HTTPException(status_code=409, detail=str(e))

get_current_test_description() async

Get the description of the current test (PREPARED or RUNNING state).

Available after POST /info has been called. Returns the markdown content of description.md

Source code in src/routers/test.py

@router.get("/description", responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    }
})
async def get_current_test_description() -> dict:
    """
    Get the description of the current test (PREPARED or RUNNING state).

    Available after POST /info has been called.
    Returns the markdown content of description.md
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    try:
        content = test_manager.get_description(test_manager.current_test.test_id)
        return {"content": content}
    except FileNotFoundError:
        raise HTTPException(status_code=500, detail="Description file not found")

get_current_test_info() async

Return the metadata of the current prepared/test in memory.

Returns 409 if no test metadata was prepared via POST /info.

Source code in src/routers/test.py

@router.get("/info", responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    }
})
async def get_current_test_info() -> Any:
    """
    Return the metadata of the current prepared/test in memory.

    Returns 409 if no test metadata was prepared via POST /info.
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    # Return a JSON-serializable dict of the TestMetaData
    try:
        return asdict(test_manager.current_test)
    except Exception:
        # Fallback: attempt to convert via dataclass fields
        return dict(test_manager.current_test.__dict__)

get_file_from_current_test(filename) async

Retrieve a file from the current test.

Returns the content of the specified file that was added to the current test. Available after POST /info has been called.

Response body:

{
    "file": "<base64-encoded file content>"
}

Source code in src/routers/test.py

@router.get("/files/{filename}", responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    },
    404: {
        "description": "File not found in current test.",
        "content": {
            "application/json": {
                "example": {"detail": "File example.txt not found in current test."}
            }
        }
    }
})
async def get_file_from_current_test(filename: str) -> bytes:
    """
    Retrieve a file from the current test.

    Returns the content of the specified file that was added to the current test.
    Available after POST /info has been called.

    Response body:
    ```json
    {
        "file": "<base64-encoded file content>"
    }
    ```
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    try:
        file_content = test_manager.get_file(filename)
        if file_content is None:
            raise HTTPException(status_code=404, detail=f"File {filename} not found in current test.")

        return base64.b64encode(file_content)
    except FileNotFoundError:
        raise HTTPException(status_code=404, detail=f"File {filename} not found in current test.")

get_test_state() async

Get the current state of the test system.

Returns one of four possible states: - NOTHING: No test is running, no test is stopped, AND no test metadata has been prepared. Ready to start fresh. - PREPARED: No test is currently running AND no test stopped, BUT test metadata has been set. Ready to call PUT /start to begin the test. - RUNNING: A test is currently executing and recording data. Data is being collected; call PUT /stop to end recording. - STOPPED: A test has stopped recording BUT not yet finalized. Review the data, then call PUT /finalize to move to history.

Source code in src/routers/test.py

@router.get("/running", response_model=TestStatusResponse)
async def get_test_state() -> TestStatusResponse:
    """
    Get the current state of the test system.

    Returns one of four possible states:
    - **NOTHING**: No test is running, no test is stopped, AND no test metadata has been prepared.
      Ready to start fresh.
    - **PREPARED**: No test is currently running AND no test stopped, BUT test metadata has been set.
      Ready to call PUT /start to begin the test.
    - **RUNNING**: A test is currently executing and recording data.
      Data is being collected; call PUT /stop to end recording.
    - **STOPPED**: A test has stopped recording BUT not yet finalized.
      Review the data, then call PUT /finalize to move to history.
    """
    state = test_manager.get_test_state()
    return TestStatusResponse(status=state)

list_current_test_files() async

List the files that have been added to the current test.

Returns a list of filenames that have been uploaded via POST /files for the current test. Available after POST /info has been called.

Source code in src/routers/test.py

@router.get("/files", responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    }
})
async def list_current_test_files() -> dict:
    """
    List the files that have been added to the current test.

    Returns a list of filenames that have been uploaded via POST /files for the current test.
    Available after POST /info has been called.
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    files = test_manager.list_files()
    return {"files": files}

set_test_info(metadata) async

Set test metadata to prepare for a test run.

This endpoint stores the test information without starting data collection. After calling this, the test will be in PREPARED state. Then call PUT /start to begin the actual test.

Source code in src/routers/test.py

@router.post("/info", status_code=204, responses={
    409: {
        "description": "A test is already running. Cannot prepare new metadata while a test is in progress.",
        "content": {
            "application/json": {
                "example": {"detail": "A test is already running."}
            }
        }
    }
})
async def set_test_info(metadata: TestMetaData) -> None:
    """
    Set test metadata to prepare for a test run.

    This endpoint stores the test information without starting data collection.
    After calling this, the test will be in PREPARED state.
    Then call PUT /start to begin the actual test.
    """
    # Prepare the test with metadata
    try:
        test_manager.prepare_test(metadata)
    except RuntimeError as e:
        # A test is already running
        raise HTTPException(status_code=409, detail=str(e))

start_test() async

Start the prepared test and begin recording data.

Requires that metadata has been set via POST /info first. Changes state from PREPARED to RUNNING.

Source code in src/routers/test.py

@router.put("/start", status_code=204, responses={
    400: {
        "description": "No test metadata prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test metadata prepared. Call POST /info first."}
            }
        }
    },
    409: {
        "description": "A test is already running.",
        "content": {
            "application/json": {
                "example": {"detail": "A test is already running."}
            }
        }
    }
})
async def start_test() -> None:
    """
    Start the prepared test and begin recording data.

    Requires that metadata has been set via POST /info first.
    Changes state from PREPARED to RUNNING.
    """
    # Start the prepared test via TestManager
    try:
        test_manager.start_test()
    except ValueError as e:
        # Missing prepared test
        raise HTTPException(status_code=400, detail=str(e))
    except RuntimeError as e:
        # A test is already running
        raise HTTPException(status_code=409, detail=str(e))

stop_test() async

Stop the currently running test and end data recording.

Changes state from RUNNING to STOPPED. Test data is preserved in memory and on disk but not yet moved to history. Call PUT /finalize to move the test to history.

Source code in src/routers/test.py

@router.put("/stop", status_code=204)
async def stop_test() -> None:
    """
    Stop the currently running test and end data recording.

    Changes state from RUNNING to STOPPED.
    Test data is preserved in memory and on disk but not yet moved to history.
    Call PUT /finalize to move the test to history.
    """
    # Stop the current test via TestManager
    test_manager.stop_test()

update_current_test_description(payload) async

Update the description of the current test (PREPARED or RUNNING state).

Available after POST /info has been called.
Modifies the description.md file in real-time.

Request body:
```json
{
    "content": "# Updated Description

New markdown content..." } ```

Source code in src/routers/test.py

@router.put("/description", status_code=204, responses={
    409: {
        "description": "No test prepared. Call POST /info first.",
        "content": {
            "application/json": {
                "example": {"detail": "No test prepared. Call POST /info first."}
            }
        }
    },
    500: {
        "description": "Failed to update description.",
        "content": {
            "application/json": {
                "example": {"detail": "Failed to update description: Permission denied"}
            }
        }
    }
})
async def update_current_test_description(payload: dict) -> None:
    """
    Update the description of the current test (PREPARED or RUNNING state).

    Available after POST /info has been called.
    Modifies the description.md file in real-time.

    Request body:
    ```json
    {
        "content": "# Updated Description\n\nNew markdown content..."
    }
    ```
    """
    if test_manager.current_test is None:
        raise HTTPException(status_code=409, detail="No test prepared. Call POST /info first.")

    content = payload.get("content", "")
    test_id = test_manager.current_test.test_id

    if not test_manager.set_description(test_id, content):
        raise HTTPException(status_code=500, detail="Failed to update description")

Sensor Router

get_sensor_data(sensor_id) async

Get the latest data point from a sensor (calibrated/processed value). Time is relative to test start if a test is running, otherwise 0.

Source code in src/routers/sensor.py

@router.get("/{sensor_id}/data", response_model=Point, responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": f"Invalid sensor_id: INVALID. Valid values are: {VALID_SENSOR_VALUES}"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    }
})
async def get_sensor_data(sensor_id: str) -> Point:
    """
    Get the latest data point from a sensor (calibrated/processed value).
    Time is relative to test start if a test is running, otherwise 0.
    """
    # Validate sensor_id
    try:
        sid = SensorId[sensor_id.upper()]
    except KeyError:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid sensor_id: {sensor_id}. Valid values are: {VALID_SENSOR_VALUES}"
        )

    # Check sensor connection status
    if not sensor_manager.is_sensor_connected(sid):
        raise HTTPException(
            status_code=503,
            detail=f"Sensor {sensor_id.upper()} is not connected"
        )

    idx = sid.value
    corrected = sensor_manager.sensors[idx]
    return Point(time=corrected.timestamp, value=corrected.value)

get_sensor_data_history(sensor_id, window=30) async

Get historical data points from a sensor (calibrated/processed values). Window is expressed in seconds and must be one of: 30, 60, 120, 300, 600. Returns a fixed number of points (based on 30s sampling) evenly spaced across the window.

Source code in src/routers/sensor.py

@router.get("/{sensor_id}/data/history", response_model=PointsList, responses={
    400: {
        "description": "Invalid sensor_id or window parameter.",
        "content": {
            "application/json": {
                "examples": {
                    "invalid_sensor": {"value": {"detail": f"Invalid sensor_id: INVALID. Valid values are: {VALID_SENSOR_VALUES}"}},
                    "invalid_window": {"value": {"detail": "Invalid window: 45. Allowed values are: [30, 60, 120, 300, 600]"}}
                }
            }
        }
    },
    409: {
        "description": "No test is currently running.",
        "content": {
            "application/json": {
                "example": {"detail": "No test is currently running"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    }
})
async def get_sensor_data_history(sensor_id: str, window: int = 30) -> PointsList:
    """
    Get historical data points from a sensor (calibrated/processed values).
    Window is expressed in seconds and must be one of: 30, 60, 120, 300, 600.
    Returns a fixed number of points (based on 30s sampling) evenly spaced across the window.
    """
    # Validate sensor_id
    try:
        sid = SensorId[sensor_id.upper()]
    except KeyError:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid sensor_id: {sensor_id}. Valid values are: {VALID_SENSOR_VALUES}"
        )

    # Check sensor connection status
    if not sensor_manager.is_sensor_connected(SensorId[sensor_id.upper()]):
        raise HTTPException(
            status_code=503,
            detail=f"Sensor {sensor_id.upper()} is not connected"
        )

    allowed_windows = {d.value_seconds() for d in DisplayDuration}
    if window not in allowed_windows:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid window: {window}. Allowed values are: {sorted(allowed_windows)}"
        )

    try:
        data_points = test_manager.get_sensor_history(sid, window)
    except ValueError as exc:
        raise HTTPException(status_code=400, detail=str(exc))
    except RuntimeError as exc:
        raise HTTPException(status_code=409, detail=str(exc))

    points = [Point(time=t, value=v) for t, v in data_points]
    return PointsList(list=points)

get_sensor_raw_data(sensor_id) async

Get the latest raw (uncalibrated) data point from a sensor. Time is relative to test start if a test is running, otherwise 0.

Source code in src/routers/sensor.py

@router.get("/{sensor_id}/raw", response_model=Point, responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": f"Invalid sensor_id: INVALID. Valid values are: {VALID_SENSOR_VALUES}"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    }
})
async def get_sensor_raw_data(sensor_id: str) -> Point:
    """
    Get the latest raw (uncalibrated) data point from a sensor.
    Time is relative to test start if a test is running, otherwise 0.
    """
    # Validate sensor_id
    try:
        sid = SensorId[sensor_id.upper()]
    except KeyError:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid sensor_id: {sensor_id}. Valid values are: {VALID_SENSOR_VALUES}"
        )

    # Check sensor connection status
    if not sensor_manager.is_sensor_connected(sid):
        raise HTTPException(
            status_code=503,
            detail=f"Sensor {sid.name} is not connected"
        )

    idx = sid.value
    corrected = sensor_manager.sensors[idx]
    return Point(time=corrected.timestamp, value=corrected.raw_value)

get_sensor_zero_offset(sensor_id) async

Get the current zero offset applied to raw readings for a sensor. The offset represents how much raw values are shifted to compute calibrated values.

Source code in src/routers/sensor.py

@router.get("/{sensor_id}/zero", response_model=OffsetResponse, responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": f"Invalid sensor_id: INVALID. Valid values are: {VALID_SENSOR_VALUES}"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    }
})
async def get_sensor_zero_offset(sensor_id: str) -> OffsetResponse:
    """
    Get the current zero offset applied to raw readings for a sensor.
    The offset represents how much raw values are shifted to compute calibrated values.
    """
    try:
        sensor = SensorId[sensor_id.upper()]
    except KeyError:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid sensor_id: {sensor_id}. Valid values are: {VALID_SENSOR_VALUES}"
        )

    # Check sensor connection status
    if not sensor_manager.is_sensor_connected(sensor):
        raise HTTPException(
            status_code=503,
            detail=f"Sensor {sensor_id.upper()} is not connected"
        )

    offset = sensor_manager.offsets[sensor.value]
    return OffsetResponse(offset=offset)

get_sensors_all() async

Get the latest data points from all sensors, including raw values, calibrated values, and zero offsets.

Source code in src/routers/sensor.py

@router.get("", response_model=DictPoint, responses={})
async def get_sensors_all() -> DictPoint:
    """
    Get the latest data points from all sensors, including raw values, calibrated values, and zero offsets.
    """
    points: DictPoint = DictPoint(raw={}, data={}, zeros={})
    for sensor in SensorId:
        points.raw[sensor.name] = Point(time=test_manager.get_relative_time(), value=0)
        points.data[sensor.name] = Point(time=test_manager.get_relative_time(), value=0)
        points.zeros[sensor.name] = OffsetResponse(offset=0)
        if sensor_manager.is_sensor_connected(sensor):
            value = sensor_manager.sensors[sensor.value]
            points.raw[sensor.name].value = value.raw_value
            points.data[sensor.name].value = value.value
            points.zeros[sensor.name].offset = value.offset
        else:
            points.raw[sensor.name].value = math.nan
            points.data[sensor.name].value = math.nan
            points.zeros[sensor.name].offset = math.nan

    return points

zero_sensor(sensor_id) async

Zero a sensor by recording its current value. Future readings from this sensor will be adjusted by subtracting this value.

Source code in src/routers/sensor.py

@router.put("/{sensor_id}/zero", status_code=204, responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": f"Invalid sensor_id: INVALID. Valid values are: {VALID_SENSOR_VALUES}"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    }
})
async def zero_sensor(sensor_id: str) -> None:
    """
    Zero a sensor by recording its current value.
    Future readings from this sensor will be adjusted by subtracting this value.
    """
    try:
        # Convert string to SensorId enum
        sensor = SensorId[sensor_id.upper()]
    except KeyError:
        raise HTTPException(
            status_code=400,
            detail=f"Invalid sensor_id: {sensor_id}. Valid values are: {', '.join([s.name for s in SensorId])}"
        )

    # Check sensor connection status
    if not sensor_manager.is_sensor_connected(sensor):
        raise HTTPException(
            status_code=503,
            detail=f"Sensor {sensor.name} is not connected"
        )

    # Launch the zeroing operation of sensor
    sensor_manager.set_zero(sensor)

Graphique Router

get_graphique(sensor_id=Path(..., description='Sensor name: DISP_1 or ARC')) async

Get the current test graphique as PNG image.

Parameters:

Name	Type	Description	Default
sensor_id	str	Either 'DISP_1' or 'ARC' for X-axis sensor	Path(..., description='Sensor name: DISP_1 or ARC')

Y-axis: FORCE Points are added in real-time as data is processed.

Source code in src/routers/graphique.py

@router.get("/{sensor_id}", response_class=StreamingResponse, responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": "sensor_id must be 'DISP_1' or 'ARC'"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    },
    409: {
        "description": "No test is currently running.",
        "content": {
            "application/json": {
                "example": {"detail": "No test is currently running."}
            }
        }
    }
})
async def get_graphique(
    sensor_id: str = Path(..., description="Sensor name: DISP_1 or ARC")
):
    """
    Get the current test graphique as PNG image.

    Args:
        sensor_id: Either 'DISP_1' or 'ARC' for X-axis sensor

    Y-axis: FORCE
    Points are added in real-time as data is processed.
    """
    if sensor_id not in ['DISP_1', 'ARC']:
        raise HTTPException(status_code=400, detail="sensor_id must be 'DISP_1' or 'ARC'")

    # For DISP_1 X-axis, check DISP_1 connection
    if sensor_id == 'DISP_1':
        if not sensor_manager.is_sensor_connected(SensorId.DISP_1):
            raise HTTPException(status_code=503, detail="Sensor DISP_1 is not connected")
    # For ARC X-axis, check DISP_2 and DISP_3 connection (ARC is calculated from DISP_2, DISP_3)
    else:  # sensor_id == 'ARC'
        if not sensor_manager.is_sensor_connected(SensorId.ARC):
            raise HTTPException(status_code=503, detail="Sensors DISP_2 and DISP_3 are not connected (required for ARC calculation)")

    # Also check FORCE connection (Y-axis)
    if not sensor_manager.is_sensor_connected(SensorId.FORCE):
        raise HTTPException(status_code=503, detail="Sensor FORCE is not connected")

    if not test_manager.get_test_state() == TestState.RUNNING:
        raise HTTPException(status_code=409, detail=f"No test is currently running.")


    png_data = test_manager.get_graphique_png(sensor_id)

    return StreamingResponse(
        io.BytesIO(png_data),
        media_type="image/png",
        headers={"Content-Disposition": f"inline; filename=graph_{sensor_id}.png"}
    )

get_graphique_base64(sensor_id=Path(..., description='Sensor name: DISP_1 or ARC')) async

Get the current test graphique as base64-encoded PNG.

Parameters:

Name	Type	Description	Default
sensor_id	str	Either 'DISP_1' or 'ARC' for X-axis sensor	Path(..., description='Sensor name: DISP_1 or ARC')

Useful for embedding in frontend applications. Returns: {"data": "data:image/png;base64,..."}

Source code in src/routers/graphique.py

@router.get("/{sensor_id}/base64", responses={
    400: {
        "description": "Invalid sensor_id provided.",
        "content": {
            "application/json": {
                "example": {"detail": "sensor_id must be 'DISP_1' or 'ARC'"}
            }
        }
    },
    503: {
        "description": "Sensor is not currently connected.",
        "content": {
            "application/json": {
                "example": {"detail": "Sensor FORCE is not connected"}
            }
        }
    },
    409: {
        "description": "No test is currently running.",
        "content": {
            "application/json": {
                "example": {"detail": "No test is currently running."}
            }
        }
    }
})
async def get_graphique_base64(
    sensor_id: str = Path(..., description="Sensor name: DISP_1 or ARC")
):
    """
    Get the current test graphique as base64-encoded PNG.

    Args:
        sensor_id: Either 'DISP_1' or 'ARC' for X-axis sensor

    Useful for embedding in frontend applications.
    Returns: {"data": "data:image/png;base64,..."}
    """
    if sensor_id not in ['DISP_1', 'ARC']:
        raise HTTPException(status_code=400, detail="sensor_id must be 'DISP_1' or 'ARC'")

    # For DISP_1 X-axis, check DISP_1 connection
    if sensor_id == 'DISP_1':
        if not sensor_manager.is_sensor_connected(SensorId.DISP_1):
            raise HTTPException(status_code=503, detail="Sensor DISP_1 is not connected")
    # For ARC X-axis, check DISP_2 and DISP_3 connection (ARC is calculated from DISP_2, DISP_3)
    else:  # sensor_id == 'ARC'
        if not sensor_manager.is_sensor_connected(SensorId.ARC):
            raise HTTPException(status_code=503, detail="Sensors DISP_2 and DISP_3 are not connected (required for ARC calculation)")

    # Also check FORCE connection (Y-axis)
    if not sensor_manager.is_sensor_connected(SensorId.FORCE):
        raise HTTPException(status_code=503, detail="Sensor FORCE is not connected")

    if not test_manager.get_test_state() == TestState.RUNNING:
        raise HTTPException(status_code=409, detail=f"No test is currently running.")

    png_data = test_manager.get_graphique_png(sensor_id)
    base64_data = base64.b64encode(png_data).decode('utf-8')

    return {
        "data": f"data:image/png;base64,{base64_data}"
    }

Response Schemas

API Response Schemas.

This module defines Pydantic models for all API responses, including health checks, data points, test status, and historical data structures.

HealthOK

Bases: BaseModel

Basic health check response.

Attributes:

Name	Type	Description
status	str	Health status as string (e.g., 'ok', 'healthy').

Source code in src/schemas.py

class HealthOK(BaseModel):
    """Basic health check response.

    Attributes:
        status: Health status as string (e.g., 'ok', 'healthy').
    """
    status: str

AppHealthOK

Bases: BaseModel

Application-level health check response.

Attributes:

Name	Type	Description
status	str	Health status as string.
app	str	Application name or version.

Source code in src/schemas.py

class AppHealthOK(BaseModel):
    """Application-level health check response.

    Attributes:
        status: Health status as string.
        app: Application name or version.
    """
    status: str
    app: str

Point

Bases: BaseModel

Single data point with timestamp.

Attributes:

Name	Type	Description
time	float	Timestamp in seconds.
value	float	Numeric value of the measurement.

Source code in src/schemas.py

class Point(BaseModel):
    """Single data point with timestamp.

    Attributes:
        time: Timestamp in seconds.
        value: Numeric value of the measurement.
    """
    time: float
    value: float

OffsetResponse

Bases: BaseModel

Sensor offset calibration response.

Attributes:

Name	Type	Description
offset	float	Offset value for sensor calibration.

Source code in src/schemas.py

class OffsetResponse(BaseModel):
    """Sensor offset calibration response.

    Attributes:
        offset: Offset value for sensor calibration.
    """
    offset: float

DictPoint

Bases: BaseModel

Complete sensor data with raw, processed, and offset values.

Attributes:

Name	Type	Description
raw	dict[str, Point]	Dictionary of raw sensor measurements by sensor ID.
data	dict[str, Point]	Dictionary of processed sensor measurements by sensor ID.
zeros	dict[str, OffsetResponse]	Dictionary of sensor offset calibrations by sensor ID.

Source code in src/schemas.py

class DictPoint(BaseModel):
    """Complete sensor data with raw, processed, and offset values.

    Attributes:
        raw: Dictionary of raw sensor measurements by sensor ID.
        data: Dictionary of processed sensor measurements by sensor ID.
        zeros: Dictionary of sensor offset calibrations by sensor ID.
    """
    raw: dict[str, Point]
    data: dict[str, Point]
    zeros: dict[str, OffsetResponse]

PointsList

Bases: BaseModel

List of data points.

Attributes:

Name	Type	Description
list	List[Point]	Collection of Point objects.

Source code in src/schemas.py

class PointsList(BaseModel):
    """List of data points.

    Attributes:
        list: Collection of Point objects.
    """
    list: List[Point]

HistoryList

Bases: BaseModel

List of historical data identifiers.

Attributes:

Name	Type	Description
list	List[str]	Collection of history file or record identifiers.

Source code in src/schemas.py

class HistoryList(BaseModel):
    """List of historical data identifiers.

    Attributes:
        list: Collection of history file or record identifiers.
    """
    list: List[str]

EmptyResponse

Bases: BaseModel

Empty response model for endpoints with no return data.

Source code in src/schemas.py

class EmptyResponse(BaseModel):
    """Empty response model for endpoints with no return data."""
    pass

FieldsResponse

Bases: BaseModel

Response containing field metadata.

Attributes:

Name	Type	Description
fields	Union[List[Any], TestMetaData]	Either a list of field definitions or TestMetaData object.

Source code in src/schemas.py

class FieldsResponse(BaseModel):
    """Response containing field metadata.

    Attributes:
        fields: Either a list of field definitions or TestMetaData object.
    """
    fields: Union[List[Any], TestMetaData]

MessageResponse

Bases: BaseModel

Generic message response.

Attributes:

Name	Type	Description
message	str	Response message text.

Source code in src/schemas.py

class MessageResponse(BaseModel):
    """Generic message response.

    Attributes:
        message: Response message text.
    """
    message: str

TestStatusResponse

Bases: BaseModel

Test execution status response.

Attributes:

Name	Type	Description
status	TestState	Current state of the test (e.g., running, completed, failed).

Source code in src/schemas.py

class TestStatusResponse(BaseModel):
    """Test execution status response.

    Attributes:
        status: Current state of the test (e.g., running, completed, failed).
    """
    status: TestState