Files
IPCam_OrangePi_Dashboard/.trae/IPCam_Backend_FastAPI.md
T
2026-04-30 21:41:16 +07:00

2.9 KiB

IPCam Orange Pi Dashboard - FastAPI Backend Skeleton

1. Project Structure

ipcam-dashboard/
├── main.py
├── mediamtx.py
├── scheduler.py
├── models.py
├── config.py
└── requirements.txt

2. requirements.txt

fastapi
uvicorn
httpx
pydantic

3. config.py

MEDIAMTX_API = "http://127.0.0.1:9997"

CAMERAS = ["cam1", "cam2", "cam3", "cam4"]

4. mediamtx.py (MediaMTX API wrapper)

import httpx
from config import MEDIAMTX_API, CAMERAS


async def set_recording(enabled: bool):
    payload = {
        "paths": {
            cam: {"record": enabled}
            for cam in CAMERAS
        }
    }

    async with httpx.AsyncClient() as client:
        r = await client.post(
            f"{MEDIAMTX_API}/v3/config/paths/patch",
            json=payload,
            timeout=5
        )
        return r.json()


async def get_paths():
    async with httpx.AsyncClient() as client:
        r = await client.get(f"{MEDIAMTX_API}/v3/paths/list")
        return r.json()

5. scheduler.py

from datetime import datetime
from mediamtx import set_recording

_last_state = None


def is_record_time(now: datetime) -> bool:
    day = now.weekday()  # 0=Mon
    hour = now.hour

    # Weekend: always record
    if day in (5, 6):
        return True

    # Weekdays: 18:00 → 08:00
    return hour >= 18 or hour < 8


async def scheduler_loop():
    global _last_state

    while True:
        now = datetime.now()
        should_record = is_record_time(now)

        if _last_state != should_record:
            print(f"[Scheduler] Set recording = {should_record}")
            await set_recording(should_record)
            _last_state = should_record

        import asyncio
        await asyncio.sleep(60)

6. models.py

from pydantic import BaseModel


class RecordingToggle(BaseModel):
    enabled: bool

7. main.py

from fastapi import FastAPI
import asyncio

from mediamtx import set_recording, get_paths
from scheduler import scheduler_loop
from models import RecordingToggle

app = FastAPI()


@app.on_event("startup")
async def startup():
    asyncio.create_task(scheduler_loop())


@app.get("/")
async def root():
    return {"status": "ok"}


@app.get("/paths")
async def list_paths():
    return await get_paths()


@app.post("/recording")
async def toggle_recording(data: RecordingToggle):
    return await set_recording(data.enabled)

8. Run Server

uvicorn main:app --host 0.0.0.0 --port 8008

9. API Endpoints

Get stream status

GET /paths

Enable recording

POST /recording
{
  "enabled": true
}

Disable recording

POST /recording
{
  "enabled": false
}

10. Next Steps

  • Add recordings API (list files)
  • Serve video files (StaticFiles)
  • Add camera config management (DB or JSON)
  • Build frontend dashboard (WebRTC grid)