Merge branch 'main' into feature/dtd-doc-updates

Author: sugnanprabhu

metro-ai-suite/smart-traffic-intersection-agent/pytest.ini

@@ -0,0 +1,23 @@
+[pytest]
+# Test discovery
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
+
+# Output configuration
+addopts = -v --tb=short --strict-markers
+
+# Async configuration
+asyncio_mode = auto
+
+# Markers
+markers =
+    unit: Unit tests (fast, isolated)
+
+# Logging
+log_cli = true
+log_cli_level = INFO
+
+# Coverage (optional, requires pytest-cov)
+# addopts = --cov=src --cov-report=term-missing --cov-report=html

metro-ai-suite/smart-traffic-intersection-agent/setup.sh

@@ -235,36 +235,36 @@ print_all_service_host_endpoints() {
             *dlstreamer-pipeline-server*)
                 SERVICE_NAME="DLStreamer Pipeline Server"
                 PORT=$(docker port "$CONTAINER_NAME" 8080 | cut -d: -f2)
-                echo -e "${BLUE}Access $SERVICE_NAME at -> http://$HOST_IP:$PORT${NC}/pipelines"
+                echo -e "${BLUE}Access $SERVICE_NAME -> http://$HOST_IP:$PORT${NC}/pipelines"
                 ;;
             *grafana*)
                 SERVICE_NAME="Grafana Dashboard"
                 PORT=$(docker port "$CONTAINER_NAME" 3000 | cut -d: -f2)
-                echo -e "${BLUE}Access $SERVICE_NAME at -> http://$HOST_IP:$PORT${NC}"
+                echo -e "${BLUE}Access $SERVICE_NAME -> http://$HOST_IP:$PORT${NC}"
                 ;;
             *node-red*)
                 SERVICE_NAME="Node-RED"
                 PORT=$(docker port "$CONTAINER_NAME" 1880 | cut -d: -f2)
-                echo -e "${BLUE}Access $SERVICE_NAME at -> http://$HOST_IP:$PORT${NC}"
+                echo -e "${BLUE}Access $SERVICE_NAME -> http://$HOST_IP:$PORT${NC}"
                 ;;
             *web*)
                 SERVICE_NAME="Scenescape Web UI"
                 PORT=$(docker port "$CONTAINER_NAME" 443 | cut -d: -f2)
-                echo -e "${BLUE}Access $SERVICE_NAME at -> https://$HOST_IP:$PORT${NC}"
+                echo -e "${BLUE}Access $SERVICE_NAME -> https://$HOST_IP:$PORT${NC}"
                 ;;
             *traffic-agent*)
                 BACKEND_SERVICE_NAME="Traffic Intersection Agent API Docs"
                 PORT=$(docker port "$CONTAINER_NAME" 8081 | cut -d: -f2)
-                echo -e "${CYAN}$BACKEND_SERVICE_NAME -> http://$HOST_IP:$PORT/docs${NC}"
+                echo -e "${BLUE}Access $BACKEND_SERVICE_NAME -> http://$HOST_IP:$PORT/docs${NC}"
 
                 UI_SERVICE_NAME="Traffic Intersection Agent UI"
                 PORT=$(docker port "$CONTAINER_NAME" 7860 | cut -d: -f2)
-                echo -e "${CYAN}$UI_SERVICE_NAME -> http://$HOST_IP:$PORT${NC}"
+                echo -e "${BLUE}Access $UI_SERVICE_NAME -> http://$HOST_IP:$PORT${NC}"
                 ;;
             *vlm*)
                 SERVICE_NAME="VLM OpenVINO Serving API"
                 PORT=$(docker port "$CONTAINER_NAME" 8000 | cut -d: -f2)
-                echo -e "${CYAN}$SERVICE_NAME -> http://$HOST_IP:$PORT/docs${NC}"
+                echo -e "${BLUE}Access $SERVICE_NAME -> http://$HOST_IP:$PORT/docs${NC}"
                 ;;
             *)
                 SERVICE_NAME="Unknown Service"

metro-ai-suite/smart-traffic-intersection-agent/src/services/data_aggregator.py

@@ -371,7 +371,7 @@ def get_service_status(self) -> Dict[str, Any]:
             "active_analyzed_cameras": len(self.vlm_analyzed_camera_images),
             "has_weather_data": self.vlm_analyzed_weather_data is not None,
             "has_vlm_analysis": self.current_vlm_analysis is not None,
-            "last_analysis_time": self.last_analysis_time.isoformat() if self.last_analysis_time else None,
+            "last_analysis_time": datetime.fromtimestamp(self.last_analysis_time, tz=timezone.utc).isoformat() if self.last_analysis_time else None,
             # "vlm_analyzed_history_count": len(self.traffic_history),
             # "analysis_tasks_running": self.analysis_task is not None and not self.analysis_task.done()
         }

metro-ai-suite/smart-traffic-intersection-agent/src/services/vlm_service.py

@@ -323,7 +323,9 @@ def _build_vlm_request(self, prompt: str, camera_images: List[CameraImage]) -> D
         }
 
         logger.debug("VLM request built", 
-                   request)
+                   model=request.get("model"),
+                   max_tokens=request.get("max_completion_tokens"),
+                   content_items=len(content))
 
         return request
 
@@ -555,18 +557,10 @@ def _create_fallback_analysis(self,
                 if weather_data:
                     weather_impact = True
 
-                    # Handle both WeatherType enum and string (defensive programming)
-                    weather_type = weather_data.weather_type
-                    if isinstance(weather_type, str):
-                        weather_type = WeatherType(weather_type)
-                    
-                    CRITICAL_WEATHER = {WeatherType.FIRES, WeatherType.STORM, WeatherType.FLOOD}
-                    alert_level = AlertLevel.CRITICAL if weather_type in CRITICAL_WEATHER else AlertLevel.WARNING
-                    
                     alert = VLMAlert(
                         alert_type=AlertType.WEATHER_RELATED,
-                        level=alert_level,
-                        description=self.weather_service.get_current_weather_description(weather_type),
+                        level=AlertLevel.INFO,
+                        description=self.weather_service.get_current_weather_description(),
                         weather_related=weather_impact
                     )
                     alerts.append(alert)

metro-ai-suite/smart-traffic-intersection-agent/src/services/weather_service.py

@@ -202,12 +202,22 @@ def _sync_fetch():
 
                 logger.debug("Forecast API response successful", status_code=forecast_resp.status_code)
 
-                # Step 4: Process the first forecast period (most current hour) and format according to specified structure
-                first_period = forecast_data["properties"]["periods"][0]
-                logger.info("First forecast period data", period=first_period)
+                # Step 4: Find the current hour's forecast period
+                periods = forecast_data["properties"]["periods"]
+                now = datetime.now(timezone.utc)
+                current_period = periods[0]  # Default to first period
+                
+                for period in periods:
+                    start_time = datetime.fromisoformat(period["startTime"].replace('Z', '+00:00'))
+                    end_time = datetime.fromisoformat(period["endTime"].replace('Z', '+00:00'))
+                    if start_time <= now <= end_time:
+                        current_period = period
+                        break
+                
+                logger.info("Current forecast period data", period=current_period)
 
                 # Process weather data into WeatherData object
-                weather_data = self._process_weather_data(first_period)
+                weather_data = self._process_weather_data(current_period)
 
                 logger.debug("Weather data fetched successfully", 
                            conditions=weather_data.detailed_forecast,

metro-ai-suite/smart-traffic-intersection-agent/src/ui/data_loader.py

@@ -8,10 +8,16 @@
 from typing import Optional
 from datetime import datetime, timezone
 
-from models import (
-    MonitoringData, IntersectionData, RegionCount, 
-    VLMAnalysis, WeatherData, CameraData, TrafficContext
-)
+try:
+    from ui.models import (
+        MonitoringData, IntersectionData, RegionCount,
+        VLMAnalysis, WeatherData, CameraData, TrafficContext
+    )
+except (ModuleNotFoundError, ImportError):
+    from models import (
+        MonitoringData, IntersectionData, RegionCount,
+        VLMAnalysis, WeatherData, CameraData, TrafficContext
+    )
 
 logger = logging.getLogger(__name__)
 

metro-ai-suite/smart-traffic-intersection-agent/src/ui/ui_components.py

@@ -8,8 +8,15 @@
 from PIL import Image
 from typing import Any, Optional, List, Tuple
 
-from models import MonitoringData
-from config import Config
+try:
+    from ui.models import MonitoringData
+except (ModuleNotFoundError, ImportError):
+    from models import MonitoringData
+
+try:
+    from ui.config import Config
+except (ModuleNotFoundError, ImportError):
+    from config import Config
 
 
 class ThemeColors:

metro-ai-suite/smart-traffic-intersection-agent/tests/__init__.py

@@ -0,0 +1,3 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Smart Traffic Intersection Agent Test Suite."""

metro-ai-suite/smart-traffic-intersection-agent/tests/conftest.py

@@ -0,0 +1,74 @@
+# Copyright (C) 2025 Intel Corporation
+# SPDX-License-Identifier: Apache-2.0
+"""Shared test fixtures and configuration for pytest."""
+
+import os
+import sys
+import pytest
+
+# Add src directory to Python path for imports
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src'))
+
+
+@pytest.fixture
+def mock_env_vars(monkeypatch):
+    """Set up mock environment variables for testing."""
+    monkeypatch.setenv("INTERSECTION_NAME", "Test-Intersection")
+    monkeypatch.setenv("INTERSECTION_LATITUDE", "37.7749")
+    monkeypatch.setenv("INTERSECTION_LONGITUDE", "-122.4194")
+    monkeypatch.setenv("MQTT_HOST", "localhost")
+    monkeypatch.setenv("MQTT_PORT", "1883")
+    monkeypatch.setenv("WEATHER_MOCK", "true")
+    monkeypatch.setenv("HIGH_DENSITY_THRESHOLD", "10")
+    monkeypatch.setenv("MODERATE_DENSITY_THRESHOLD", "5")
+
+
+@pytest.fixture
+def sample_traffic_data():
+    """Sample traffic data for testing."""
+    return {
+        "intersection_id": "test-intersection-001",
+        "timestamp": "2026-01-22T10:00:00Z",
+        "directions": {
+            "north": {"vehicle_count": 5, "density": "NORMAL"},
+            "south": {"vehicle_count": 8, "density": "MODERATE"},
+            "east": {"vehicle_count": 12, "density": "HIGH"},
+            "west": {"vehicle_count": 3, "density": "NORMAL"}
+        },
+        "total_vehicles": 28
+    }
+
+
+@pytest.fixture
+def sample_weather_data():
+    """Sample weather data for testing."""
+    return {
+        "temperature": 55,
+        "short_forecast": "Sunny",
+        "conditions": "Cloudy conditions with 5 mph winds from the NW. 10% chance of precipitation.",
+        "is_precipitation": False,
+        "precipitation": False,
+        "wind_speed": "5 mph",
+        "wind_direction": "NW"
+    }
+
+
+@pytest.fixture
+def sample_vlm_response():
+    """Sample VLM analysis response for testing."""
+    return {
+        "analysis": "Traffic flow is moderate with higher density on the east approach.",
+        "alerts": [
+            {
+                "alert_type": "congestion",
+                "level": "warning",
+                "description": "East approach experiencing high traffic density",
+                "weather_related": False
+            }
+        ],
+        "recommendations": [
+            {
+                "recommendation": "Consider extending green light duration for east-west traffic"
+            }
+        ]
+    }

metro-ai-suite/smart-traffic-intersection-agent/tests/unit/README.md

@@ -0,0 +1,130 @@
+# Running Unit Tests for Smart Traffic Intersection Agent
+
+This guide will help you run the unit tests for the Smart Traffic Intersection Agent project using the pytest framework.
+
+---
+
+## Table of Contents
+
+- [Prerequisites](#prerequisites)
+- [Running Tests in a Virtual Environment](#running-tests-in-a-virtual-environment)
+- [Test Files](#test-files)
+
+---
+
+## Prerequisites
+
+Before running the tests, ensure you have the following installed:
+
+- Python 3.10+
+- `pip` (Python package installer)
+- `uv` (Python package and project manager) - Optional but recommended
+
+You can install `uv` using the following command:
+
+```bash
+curl -LsSf https://astral.sh/uv/install.sh | sh
+```
+
+---
+
+## Running Tests in a Virtual Environment
+
+Follow these steps to run the tests:
+
+1. **Clone the Repository**
+
+   Clone the repository to your local machine:
+
+   ```bash
+   git clone https://github.com/open-edge-platform/edge-ai-suites.git
+   cd edge-ai-suites/metro-ai-suite/smart-traffic-intersection-agent
+   ```
+
+2. **Create and Activate a Virtual Environment**
+
+   Navigate to the project directory and create a virtual environment:
+
+   ```bash
+   python -m venv .venv
+   source .venv/bin/activate
+   ```
+
+3. **Install Dependencies**
+
+   Install application and dev dependencies:
+
+   ```bash
+   # Using uv (recommended)
+   cd src
+   uv pip install -e ".[dev]"
+   
+   # Or using pip
+   cd src
+   pip install -e ".[dev]"
+   ```
+
+4. **Navigate to the Project Root**
+
+   Change to the project root directory (where `pytest.ini` is located):
+
+   ```bash
+   cd /path/to/edge-ai-suites/metro-ai-suite/smart-traffic-intersection-agent
+   ```
+
+5. **Run the Tests**
+
+   **Important:** Make sure the virtual environment is activated before running tests. You should see `(.venv)` in your terminal prompt.
+
+   ```bash
+   # Verify you're in the virtual environment (should show .venv path)
+   which python
+   # Expected: /path/to/smart-traffic-intersection-agent/.venv/bin/python
+   
+   # If not activated, activate it first:
+   source .venv/bin/activate
+   ```
+
+   Use the below command to run all unit tests:
+
+   ```bash
+   pytest tests/unit -v
+   ```
+
+   To run a specific test file:
+
+   ```bash
+   pytest tests/unit/test_data_aggregator.py -v
+   ```
+
+   To run tests with coverage report:
+
+   ```bash
+   pytest tests/unit --cov=src --cov-report=term-missing
+   ```
+
+6. **Deactivate Virtual Environment**
+
+   When you are done with testing:
+
+   ```bash
+   deactivate
+   ```
+
+---
+
+## Test Files
+
+The following unit test files are available:
+
+| Test File | Description |
+|-----------|-------------|
+| `test_api_routes.py` | Tests for FastAPI route endpoints |
+| `test_config_service.py` | Tests for configuration service |
+| `test_data_aggregator.py` | Tests for data aggregation service |
+| `test_main.py` | Tests for main application entry point |
+| `test_mqtt_service.py` | Tests for MQTT communication service |
+| `test_run.py` | Tests for application runner |
+| `test_ui_components.py` | Tests for UI components |
+| `test_vlm_service.py` | Tests for VLM (Vision Language Model) service |
+| `test_weather_service.py` | Tests for weather data service |