set up of API #2

.gitignore

@@ -1,3 +1,4 @@
+.idea
 # Byte-compiled / optimized / DLL files
 __pycache__/
 *.py[cod]

nodeorc_api/__init__.py

@@ -0,0 +1,3 @@
+"""Backend API for operations and configuration for NodeORC front end."""
+
+__version__ = "0.1.0"

nodeorc_api/crud/__init__.py

@@ -0,0 +1,2 @@
+from . import device
+from . import disk_management

nodeorc_api/crud/device.py

@@ -0,0 +1,11 @@
+from sqlalchemy.orm import Session
+from nodeorc import db as models
+
+def get(db: Session):
+    # there should always only be one device. Hence retrieve the first.
+    return db.query(models.Device).first()
+
+def update(db: Session, device: models.Device):
+    db.query(models.Device).update(device.dict())
+    db.commit()
+    return db.query(models.Device).first()

nodeorc_api/crud/disk_management.py

@@ -0,0 +1,8 @@
+from sqlalchemy.orm import Session
+from nodeorc import db as models
+
+def get(db: Session):
+    # there should always only be one disk management config. Hence retrieve the first.
+    dms = db.query(models.DiskManagement)
+    if dms.count() > 0:
+        return db.query(models.DiskManagement).first()

nodeorc_api/database.py

@@ -0,0 +1,8 @@
+from nodeorc.db import Session
+# Dependency to get the DB session
+def get_db():
+    db = Session()
+    try:
+        yield db
+    finally:
+        db.close()

nodeorc_api/main.py

@@ -0,0 +1,28 @@
+from fastapi import FastAPI, Depends, Request
+from fastapi.middleware.cors import CORSMiddleware
+
+from nodeorc_api.routers import device, video, disk_management
+app = FastAPI()
+
+# origins = ["http://localhost:5173"]
+origins = ["*"]
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=origins,
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+
+
+app.include_router(device.router)
+app.include_router(video.router)
+app.include_router(disk_management.router)
+
+@app.get("/")
+async def root():
+    return {"message": "Hello World"}
+
+

nodeorc_api/routers/device.py

@@ -0,0 +1,44 @@
+from fastapi import APIRouter, Depends
+from nodeorc.db import Session, Device, DeviceStatus, DeviceFormStatus
+from typing import List, Dict
+
+from nodeorc_api.schemas.device import DeviceCreate, DeviceResponse
+from nodeorc_api.database import get_db
+from nodeorc_api import crud
+
+router: APIRouter = APIRouter(prefix="/device", tags=["device"])
+
+@router.get("/", response_model=DeviceResponse, description="Get device information")
+async def get_device(db: Session = Depends(get_db)):
+    device: List[Device] = crud.device.get(db)
+    return device
+
+
+@router.get("/statuses/", response_model=List[Dict], description="Get all available status options for devices")
+async def get_device_statuses():
+    return [{"key": status.name, "value": status.value} for status in DeviceStatus]
+
+
+@router.get("/form_statuses/", response_model=List[Dict], description="Get all available form status options")
+async def get_device_form_statuses():
+    return [{"key": status.name, "value": status.value} for status in DeviceFormStatus]
+
+
+@router.post("/", response_model=DeviceResponse, status_code=201, description="Update device information")
+async def update_device(device: DeviceCreate, db: Session = Depends(get_db)):
+    # Check if there is already a device
+    existing_device = crud.device.get(db)
+    if existing_device:
+        # Update the existing record's fields
+        for key, value in device.model_dump(exclude_none=True).items():
+            setattr(existing_device, key, value)
+        db.commit()
+        db.refresh(existing_device)  # Refresh to get the updated fields
+        return existing_device
+    else:
+        # Create a new device record if none exists
+        new_device = Device(**device.model_dump(exclude_none=True, exclude={"id"}))
+        db.add(new_device)
+        db.commit()
+        db.refresh(new_device)
+        return new_device

nodeorc_api/routers/disk_management.py

@@ -0,0 +1,37 @@
+from fastapi import APIRouter, Depends, Response
+from nodeorc.db import Session, DiskManagement
+from typing import List, Union
+
+from nodeorc_api.schemas.disk_management import DiskManagementResponse, DiskManagementCreate
+from nodeorc_api.database import get_db
+from nodeorc_api import crud
+
+router: APIRouter = APIRouter(prefix="/disk_management", tags=["disk_management"])
+
+@router.get("/", response_model=Union[DiskManagementResponse, None], description="Get disk management configuration.")
+async def get_disk_management_settings(db: Session = Depends(get_db)):
+    disk_management: List[DiskManagement] = crud.disk_management.get(db)
+    return disk_management
+
+
+@router.post("/", response_model=None, status_code=201, description="Update disk management configuration.")
+async def update_device(dm: DiskManagementCreate, db: Session = Depends(get_db)):
+    # Check if there is already a device
+    existing_dm = crud.disk_management.get(db)
+    try:
+        if existing_dm:
+            # Update the existing record's fields
+            for key, value in dm.model_dump(exclude_none=True).items():
+                setattr(existing_dm, key, value)
+            db.commit()
+            db.refresh(existing_dm)  # Refresh to get the updated fields
+            return existing_dm
+        else:
+            # Create a new device record if none exists
+            new_dm = DiskManagement(**dm.model_dump(exclude_none=True, exclude={"id"}))
+            db.add(new_dm)
+            db.commit()
+            db.refresh(new_dm)
+            return new_dm
+    except Exception as e:
+        return Response(f"Error: {e}", status_code=500)

nodeorc_api/routers/video.py

@@ -0,0 +1,67 @@
+from fastapi import Response, Request, APIRouter
+from fastapi.responses import StreamingResponse
+
+import cv2
+
+router: APIRouter = APIRouter(prefix="/video", tags=["video"])
+
+
+@router.get("/feed/", response_class=StreamingResponse, description="Get video stream from user-defined URL")
+async def video_feed(request: Request, video_url: str):
+    async def generate_frames():
+        if not video_url:
+            raise ValueError("No video URL provided")
+        cap = cv2.VideoCapture(video_url)
+        print(f"VIDEO URL IS: {video_url}")
+        if not cap.isOpened():
+            # return Response("Unable to open RTSP stream", status_code=500)
+            raise RuntimeError("Unable to open RTSP stream")
+
+        try:
+            from collections import deque
+            frame_buffer = deque(maxlen=10)  # Adjust maxlen based on desired buffer size
+            while True:
+                if await request.is_disconnected():
+                    break
+                success, frame = cap.read()
+                if not success:
+                    break
+                frame_buffer.append(frame)
+                # Encode the frame as JPEG
+                _, buffer = cv2.imencode(".jpg", frame_buffer.pop())
+
+                # Yield the frame as part of an MJPEG stream
+                yield (
+                    b"--frame\r\n"
+                    b"Content-Type: image/jpeg\r\n\r\n" + buffer.tobytes() + b"\r\n"
+                )
+                # Add a small delay to avoid overloading the server, probably not required.
+                # await asyncio.sleep(0.03)
+        finally:
+            print("Releasing the video capture object")
+            cap.release()
+            del cap
+
+    return StreamingResponse(generate_frames(), media_type="multipart/x-mixed-replace; boundary=frame")
+
+@router.head("/feed/", response_class=Response, description="Check if the video feed in the user defined URL is available")
+async def check_video_feed(response: Response, video_url: str):
+    """
+    HEAD endpoint to check if the video feed is available.
+    """
+    print(f"VIDEO URL IS: {video_url}")
+    if not video_url:
+        raise ValueError("No video URL provided")
+
+    try:
+        # RTSP_URL = "rtsp://nodeorcpi:8554/cam"
+        cap = cv2.VideoCapture(video_url)
+
+        if not cap.isOpened():
+            return Response("Unable to open RTSP stream", status_code=500)
+        else:
+            return Response("Video feed is available", status_code=200)
+    finally:
+        cap.release()
+        del cap
+

nodeorc_api/schemas/device.py

@@ -0,0 +1,37 @@
+import uuid
+
+from pydantic import BaseModel, Field
+from typing import Optional
+from nodeorc.db import DeviceStatus, DeviceFormStatus
+
+# Pydantic model for responses
+class DeviceBase(BaseModel):
+    name: Optional[str] = Field(default=None, description="Name of the device.")
+    operating_system: Optional[str] = Field(default=None, description="Operating system of the device.")
+    processor: Optional[str] = Field(default=None, description="Processor of the device.")
+    memory: Optional[float] = Field(default=None, description="Memory in GB available on the device.")
+    status: Optional[DeviceStatus] = Field(default=DeviceStatus.HEALTHY, description="Status of the device.")
+    form_status: Optional[DeviceFormStatus] = Field(default=DeviceFormStatus.NOFORM, description="Form status of the device.")
+    nodeorc_version: Optional[str] = Field(default=None, description="Version of nodeorc.")
+    message: Optional[str] = Field(default=None, description="Error or status message if any.")
+
+class DeviceResponse(DeviceBase):
+    id: uuid.UUID# = Field(description="Device ID")
+
+    class Config:
+        orm_mode = True
+
+class DeviceCreate(DeviceBase):
+    pass
+
+
+class DeviceUpdate(BaseModel):
+    name: Optional[str]
+    operating_system: Optional[str]
+    processor: Optional[str]
+    memory: Optional[float]
+    status: Optional[DeviceStatus]
+    form_status: Optional[DeviceFormStatus]
+    nodeorc_version: Optional[str]
+    message: Optional[str]
+

nodeorc_api/schemas/disk_management.py

@@ -0,0 +1,18 @@
+from datetime import datetime
+from pydantic import BaseModel, Field
+from typing import Optional
+
+
+# Pydantic model for responses
+class DiskManagementBase(BaseModel):
+    home_folder: Optional[str] = Field(default=None, description="Home folder of the device.")
+    min_free_space: Optional[float] = Field(default=None, description="GB of minimum free space required.")
+    critical_space: Optional[float] = Field(default=None, description="GB of free space critical for the device.")
+    frequency: Optional[int] = Field(default=None, description="Frequency [s] for checking disk status and cleanup.")
+
+class DiskManagementResponse(DiskManagementBase):
+    id: int = Field(description="Disk management ID")
+    created_at: datetime = Field(description="Creation date")
+
+class DiskManagementCreate(DiskManagementBase):
+    pass

pyproject.toml

@@ -0,0 +1,46 @@
+[build-system]
+requires = ["flit_core >=3.4.0,<4"]
+build-backend = "flit_core.buildapi"
+
+[project]
+name = "nodeorc_api"
+authors = [
+    {name = "Hessel Winsemius", email = "winsemius@rainbowsensing.com"},
+]
+dependencies = [
+#    "nodeorc",
+    "fastapi[standard]"
+]
+requires-python = ">=3.9" # fix tests to support older versions
+readme = "README.md"
+classifiers = [
+    "Development Status :: 3 - Alpha",
+    "Intended Audience :: Developers",
+    "Intended Audience :: Science/Research",
+    "Topic :: Scientific/Engineering :: Hydrology",
+    "Topic :: Scientific/Engineering :: Image Processing",
+    "License :: OSI Approved :: GNU Affero General Public License v3",
+    "License :: OSI Approved :: Image Processing",
+    "Programming Language :: Python :: 3",
+    "Programming Language :: Python :: 3.9",
+    "Programming Language :: Python :: 3.10",
+    "Programming Language :: Python :: 3.11"
+]
+dynamic = ['version', 'description']
+
+[project.optional-dependencies]
+test = [
+	"pytest",
+	"pytest-cov",
+]
+
+[project.urls]
+Source = "https://github.com/localdevices//nodeorc-interface"
+
+
+[tool.flit.sdist]
+include = ["nodeorc_api"]
+exclude = ["gcloud_dist", "hack", "helm", "htmlcov", "junit", "tests", ".github"]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]