SYNCRO - Real-Time Assembly Sequence Monitoring (IoT + AI)

Domains/IoT/MiniProjects/syncro/.gitignore

@@ -0,0 +1,43 @@
+# Python
+__pycache__/
+*.py[cod]
+*$py.class
+*.so
+.Python
+**/__pycache__/
+
+# Virtual environments
+venv/
+env/
+ENV/
+.venv
+
+# IDEs
+.vscode/
+.idea/
+*.swp
+*.swo
+*~
+
+# OS files
+.DS_Store
+Thumbs.db
+
+# Environment variables
+.env
+*.env
+
+# Logs
+*.log
+logs/
+
+# YOLOv8 models (these are downloaded automatically)
+*.pt
+*.onnx
+*.engine
+
+# Test/temp files
+temp/
+tmp/
+test_images/
+test_videos/

Domains/IoT/MiniProjects/syncro/backend/.env.example

@@ -0,0 +1,11 @@
+# Environment variables for SYNCRO backend
+
+# YOLOv8 Model (for high accuracy use yolov8l.pt or yolov8x.pt)
+# Options: yolov8n.pt, yolov8s.pt, yolov8m.pt, yolov8l.pt, yolov8x.pt
+YOLO_MODEL=yolov8m.pt
+
+# Camera source (0 for default webcam, or path to video file like "assembly.mp4")
+CAMERA_SOURCE=0
+
+# Confidence threshold for detections (0.0 to 1.0)
+CONFIDENCE_THRESHOLD=0.5

Domains/IoT/MiniProjects/syncro/backend/README.md

@@ -0,0 +1,98 @@
+**Contributor:** k4niz
+
+# SYNCRO - Real-Time Assembly Sequence Monitoring (IoT + AI)
+
+Backend service built with FastAPI that captures camera frames, detects parts using YOLOv8, and validates assembly sequence order in real-time.
+
+## Features
+-  Camera capture (OpenCV)
+-  YOLOv8 object detection with configurable accuracy models
+-  Assembly sequence validation and logging
+-  REST API: start/stop/status/logs
+-  postman collection included
+
+## YOLOv8 Models (Accuracy vs Speed)
+- `yolov8n.pt` - Nano (fastest, lowest accuracy)
+- `yolov8s.pt` - Small
+- `yolov8m.pt` - Medium (**default**, balanced)
+- `yolov8l.pt` - Large (high accuracy)
+- `yolov8x.pt` - Extra Large (highest accuracy, slowest)
+
+## API Endpoints
+- `POST /api/start-monitoring` - Start monitoring loop
+- `POST /api/stop-monitoring` - Stop monitoring
+- `GET /api/status` - Current status and sequence step
+- `GET /api/logs` - Last 50 detection logs
+
+## Setup
+
+1. **Navigate to syncro folder:**
+   ```powershell
+   cd Domains\IoT\MiniProjects\syncro
+   ```
+
+2. **Install dependencies:**
+   ```powershell
+   cd backend
+   pip install -r requirements.txt
+   cd ..
+   ```
+
+3. **(Optional) Configure model:**
+   Copy `.env.example` to `.env` and edit:
+   ```
+   YOLO_MODEL=yolov8l.pt  # For high accuracy
+   ```
+
+4. **Run the server:**
+   ```powershell
+   uvicorn backend.main:app --reload
+   ```
+
+5. **Test with Swagger UI:**
+   Open `http://127.0.0.1:8000/docs`
+
+6. **Or import Postman collection:**
+   `backend/SYNCRO.postman_collection.json`
+
+## How It Works
+
+1. **Camera** captures frames continuously
+2. **YOLOv8** detects objects in each frame
+3. **Label Mapper** converts COCO classes to assembly steps
+4. **Sequence Validator** checks if steps are in correct order
+5. **Logger** records all detections with timestamps
+6. **API** exposes status and logs in real-time
+
+## Customization
+
+### Change YOLO Model
+Edit `backend/utils/config.py` or set env var:
+```powershell
+$env:YOLO_MODEL="yolov8x.pt"  # Highest accuracy
+```
+
+### Customize Assembly Sequence
+Edit `backend/utils/config.py`:
+```python
+SEQUENCE_STEPS = [
+    "pick_part_a",
+    "pick_part_b",
+    "assemble"
+]
+```
+
+### Map Custom Objects
+Edit `vision_service.py` → `_create_label_map()`:
+```python
+return {
+    "screw": "pick_screw",
+    "wrench": "pick_tool",
+    "bolt": "tighten_screw"
+}
+```
+
+## Notes
+- First run downloads the YOLO model (~50MB for yolov8m)
+- If no camera is available, detection returns empty results
+- For custom part detection, train a custom YOLOv8 model on your assembly dataset

Domains/IoT/MiniProjects/syncro/backend/SYNCRO.postman_collection.json

@@ -0,0 +1,76 @@
+{
+  "info": {
+    "name": "SYNCRO - Assembly Sequence Monitoring",
+    "description": "API endpoints for real-time assembly sequence monitoring using Computer Vision",
+    "schema": "https://schema.getpostman.com/json/collection/v2.1.0/collection.json"
+  },
+  "item": [
+    {
+      "name": "Health Check",
+      "request": {
+        "method": "GET",
+        "header": [],
+        "url": {
+          "raw": "{{base_url}}/",
+          "host": ["{{base_url}}"],
+          "path": [""]
+        }
+      }
+    },
+    {
+      "name": "Start Monitoring",
+      "request": {
+        "method": "POST",
+        "header": [],
+        "url": {
+          "raw": "{{base_url}}/api/start-monitoring",
+          "host": ["{{base_url}}"],
+          "path": ["api", "start-monitoring"]
+        }
+      }
+    },
+    {
+      "name": "Stop Monitoring",
+      "request": {
+        "method": "POST",
+        "header": [],
+        "url": {
+          "raw": "{{base_url}}/api/stop-monitoring",
+          "host": ["{{base_url}}"],
+          "path": ["api", "stop-monitoring"]
+        }
+      }
+    },
+    {
+      "name": "Get Status",
+      "request": {
+        "method": "GET",
+        "header": [],
+        "url": {
+          "raw": "{{base_url}}/api/status",
+          "host": ["{{base_url}}"],
+          "path": ["api", "status"]
+        }
+      }
+    },
+    {
+      "name": "Get Logs",
+      "request": {
+        "method": "GET",
+        "header": [],
+        "url": {
+          "raw": "{{base_url}}/api/logs",
+          "host": ["{{base_url}}"],
+          "path": ["api", "logs"]
+        }
+      }
+    }
+  ],
+  "variable": [
+    {
+      "key": "base_url",
+      "value": "http://127.0.0.1:8000",
+      "type": "string"
+    }
+  ]
+}

Domains/IoT/MiniProjects/syncro/backend/__init__.py

@@ -0,0 +1 @@
+# Package init for SYNCRO backend

Domains/IoT/MiniProjects/syncro/backend/controllers/monitor_controller.py

@@ -0,0 +1,17 @@
+from backend.services.monitor_service import MonitorService
+
+monitor = MonitorService()
+
+def start_monitoring():
+    monitor.start()
+    return {"started": True}
+
+def stop_monitoring():
+    monitor.stop()
+    return {"stopped": True}
+
+def get_status():
+    return monitor.status()
+
+def get_logs():
+    return {"logs": monitor.get_logs()}

Domains/IoT/MiniProjects/syncro/backend/main.py

@@ -0,0 +1,10 @@
+from fastapi import FastAPI
+from backend.routes.monitor_routes import router as monitor_router
+
+app = FastAPI(title="SYNCRO - Assembly Sequence Monitoring")
+
+app.include_router(monitor_router, prefix="/api")
+
+@app.get("/")
+def root():
+    return {"service": "SYNCRO", "status": "ok"}

Domains/IoT/MiniProjects/syncro/backend/models/schemas.py

@@ -0,0 +1,16 @@
+from pydantic import BaseModel
+from typing import List, Optional, Any, Dict
+
+class Detection(BaseModel):
+    label: str
+    conf: float
+
+class LogEntry(BaseModel):
+    ts: float
+    detections: List[Dict[str, Any]]
+    result: Dict[str, Any]
+
+class Status(BaseModel):
+    running: bool
+    current_step: Optional[str]
+    completed: bool

Domains/IoT/MiniProjects/syncro/backend/requirements.txt

@@ -0,0 +1,6 @@
+fastapi
+uvicorn
+opencv-python
+pydantic
+python-multipart
+ultralytics

Domains/IoT/MiniProjects/syncro/backend/routes/monitor_routes.py

@@ -0,0 +1,20 @@
+from fastapi import APIRouter
+from backend.controllers.monitor_controller import start_monitoring, stop_monitoring, get_status, get_logs
+
+router = APIRouter(tags=["monitor"])
+
+@router.post("/start-monitoring")
+def start():
+    return start_monitoring()
+
+@router.post("/stop-monitoring")
+def stop():
+    return stop_monitoring()
+
+@router.get("/status")
+def status():
+    return get_status()
+
+@router.get("/logs")
+def logs():
+    return get_logs()

Domains/IoT/MiniProjects/syncro/backend/services/monitor_service.py

@@ -0,0 +1,52 @@
+import threading
+import time
+from backend.services.vision_service import VisionService
+from backend.services.sequence_service import SequenceService
+from backend.utils.logger import logger
+from backend.utils.config import SEQUENCE_STEPS, YOLO_MODEL, CAMERA_SOURCE
+
+class MonitorService:
+    def __init__(self):
+        self._running = False
+        self._thread = None
+        self.vision = VisionService(source=CAMERA_SOURCE, model_name=YOLO_MODEL)
+        self.sequence = SequenceService(steps=SEQUENCE_STEPS)
+        self._logs = []
+
+    def start(self):
+        if self._running:
+            return
+        self._running = True
+        self._thread = threading.Thread(target=self._loop, daemon=True)
+        self._thread.start()
+        logger.info("Monitoring started")
+
+    def stop(self):
+        self._running = False
+        if self._thread:
+            self._thread.join(timeout=2)
+        self.vision.release()
+        logger.info("Monitoring stopped")
+
+    def _loop(self):
+        while self._running:
+            frame = self.vision.read()
+            if frame is None:
+                time.sleep(0.05)
+                continue
+            detections = self.vision.detect(frame)
+            result = self.sequence.update(detections)
+            log_entry = {"ts": time.time(), "detections": detections, "result": result}
+            self._logs.append(log_entry)
+            logger.debug(str(log_entry))
+            time.sleep(0.05)
+
+    def status(self):
+        return {
+            "running": self._running,
+            "current_step": self.sequence.current_step,
+            "completed": self.sequence.completed,
+        }
+
+    def get_logs(self, limit: int = 50):
+        return self._logs[-limit:]

Domains/IoT/MiniProjects/syncro/backend/services/sequence_service.py

@@ -0,0 +1,25 @@
+from typing import List, Dict
+
+class SequenceService:
+    def __init__(self, steps: List[str]):
+        self.steps = steps
+        self.index = 0
+        self.completed = False
+
+    @property
+    def current_step(self):
+        if self.completed:
+            return None
+        return self.steps[self.index]
+
+    def update(self, detections: List[Dict]):
+        if self.completed:
+            return {"status": "done"}
+        labels = {d.get("label") for d in detections}
+        if self.current_step in labels:
+            self.index += 1
+            if self.index >= len(self.steps):
+                self.completed = True
+                return {"status": "completed"}
+            return {"status": "next", "current_step": self.current_step}
+        return {"status": "waiting", "current_step": self.current_step}