refined design

Utils.py

@@ -0,0 +1,7 @@
+from typing import List, Union
+from pydantic import BaseModel
+
+class CoordinatesPayload(BaseModel):
+    coords: Union[str, List[List[float]]]
+    state: str
+    time: int

app.py

@@ -1,70 +1,22 @@
 from fastapi import FastAPI, HTTPException,BackgroundTasks
-from typing import List, Union
-from pydantic import BaseModel
 import requests
 import ast
 import pymysql
 import time
-
+from database_manager import *
+from mqtt_manager import singlas_state 
+from mqtt_manager import * 
+from traffic_utils import *
+from Utils import * 
 
 # -------------------------
 # App Initialization
 # -------------------------
 app = FastAPI()
 
-# -------------------------
-# Database Setup (Pure PyMySQL)
-# -------------------------
-timeout = 20
-
-def get_db():
-    connection = pymysql.connect(
-        charset="utf8mb4",
-        connect_timeout=timeout,
-        cursorclass=pymysql.cursors.DictCursor,
-        db="trafficManagerSignals",
-        host="mysql-19285eb2-tahaelshrif1-7999.h.aivencloud.com",
-        password="AVNS_aT0RGFafs6_34WFegSF",
-        read_timeout=timeout,
-        port=11520,
-        user="avnadmin",
-        write_timeout=timeout,
-        ssl={"ssl": True},
-    )
-    return connection
-
-# -------------------------
-# Utilities
-# -------------------------
-def get_rectangle_container(coordinates):
-    lons = [i for i, _ in coordinates]
-    lats = [i for _, i in coordinates]
-    min_lat, max_lat = min(lats), max(lats)
-    min_lon, max_lon = min(lons), max(lons)
-    return min_lat, max_lat, min_lon, max_lon
-
-def get_traffic_in_container(coordinates):
-    (min_lat, max_lat, min_lon, max_lon) = get_rectangle_container(coordinates)
-    overpass_url = "http://overpass-api.de/api/interpreter"
-    query = f"""
-    [out:json];
-    node["highway"="traffic_signals"]({min_lat},{min_lon},{max_lat},{max_lon});
-    out body;
-    """
-    response = requests.get(overpass_url, params={'data': query})
-    data = response.json()
-    return [(el['lat'], el['lon']) for el in data['elements']]
 
 
-singlas_state = {}
-
-def open_signal(tl_id,state,time_):
-    singlas_state[tl_id] = state 
-    time.sleep(time_)
-    singlas_state.pop(tl_id)
-
-
-# -------------------------
+#--------------------------
 # Routes
 # -------------------------
 @app.get("/")
@@ -72,88 +24,6 @@ def read_root():
     return {"message": "Hello from FastAPI on Hugging Face Spaces!"}
 
 
-@app.post("/get_State")
-def get_State(tl_id):
-    if tl_id in singlas_state:
-        return singlas_state[tl_id]
-    else:
-        return "FREE"
-
-
-import paho.mqtt.client as mqtt
-import time
-import threading
-
-mqtt_broker = "broker.hivemq.com"  # أو بروكر محلي لو عاوز
-mqtt_port = 1883
-
-client = mqtt.Client()
-client.connect(mqtt_broker, mqtt_port)
-
-def mqtt_publisher_loop(tl_id):
-    import paho.mqtt.client as mqtt
-    import time
-
-    mqtt_broker = "broker.hivemq.com"
-    mqtt_port = 1883
-
-    client = mqtt.Client()
-    client.connect(mqtt_broker, mqtt_port)
-    client.loop_start()
-
-    last_state = None
-    while True:
-        current_state = get_State(tl_id)
-        if current_state != last_state:
-            topic = f"state/{tl_id}"
-            client.publish(topic, current_state)
-            print(f"📡 Published {current_state} to {topic}")
-            last_state = current_state
-        time.sleep(1)
-
-@app.post("/getstt")
-def start_mqtt_publishers(tl_id:str):
-    threading.Thread(target=mqtt_publisher_loop, args=(tl_id,), daemon=True).start()
-
-'''
-from fastapi import WebSocket, WebSocketDisconnect
-import asyncio
-
-@app.websocket("/ws/state")
-async def websocket_get_state(websocket: WebSocket):
-    await websocket.accept()
-    try:
-        tl_id = await websocket.receive_text()
-        print(f"✅ Client subscribed to {tl_id}")
-
-        last_state = None
-        while True:
-            current_state = get_State(tl_id)
-            if current_state != last_state:
-                await websocket.send_text(current_state)
-                last_state = current_state
-            await asyncio.sleep(1)  # كل ثانية يشيك التغيير
-    except WebSocketDisconnect:
-        print("❌ Client disconnected")
-
-        
-@app.get("/test-db")
-def test_db():
-    try:
-        connection = get_db()
-        with connection.cursor() as cursor:
-            cursor.execute("SELECT 1")
-            result = cursor.fetchone()
-        connection.close()
-        return {"status": "connected", "result": result}
-    except Exception as e:
-        return {"status": "error", "detail": str(e)}
-
-'''
-class CoordinatesPayload(BaseModel):
-    coords: Union[str, List[List[float]]]
-    state: str
-    time: int
 
 
 @app.post("/send-coordinates")
@@ -176,23 +46,3 @@ def receive_coordinates(payload: CoordinatesPayload,background_tasks: Background
     else:
         return {"error": "No traffic signals found in database"+str(traffic_lights)}
 
-# -------------------------
-# Signal Database Check
-# -------------------------
-def check_signals(coords):
-    connection = get_db()
-    found_signals = []
-    try:
-        with connection.cursor() as cursor:
-            for lat, lon in coords:
-                cursor.execute("""
-                    SELECT tl_id_sumo FROM traffic_signals
-                    WHERE ABS(lat - %s) < 0.0001 AND ABS(lon - %s) < 0.0001
-                """, (lon,lat))
-                result = cursor.fetchone()
-                if result:
-                    found_signals.append(result['tl_id_sumo'])
-    finally:
-        connection.close()
-
-    return found_signals

database_manager.py

@@ -0,0 +1,30 @@
+import pymysql
+from dotenv import load_dotenv
+import os
+
+load_dotenv(dotenv_path="keys.env")
+
+db_name = os.getenv("DATABASE_NAME")
+db_user = os.getenv("User")
+mysql_host = os.getenv("MYSQL_HOST")
+mysql_password = os.getenv("MYSQL_PASSWORD")
+db_port = int(os.getenv("MYSQL_PORT"))
+
+
+timeout = 20
+
+def get_db():
+    connection = pymysql.connect(
+        charset="utf8mb4",
+        connect_timeout=timeout,
+        cursorclass=pymysql.cursors.DictCursor,
+        db=db_name,
+        host=mysql_host,
+        password=mysql_password,
+        read_timeout=timeout,
+        port=db_port,
+        user=db_user,
+        write_timeout=timeout,
+        ssl={"ssl": True},
+    )
+    return connection

keys.env

@@ -0,0 +1,13 @@
+MQTT_BROKER="broker.hivemq.com"
+MQTT_PORT=1883
+
+MQTT_PARTOPIC_state="state"
+MQTT_SUBTOPIC_reply="reply"
+MQTT_SUBTOPIC_msg="msg"
+
+# database info
+MYSQL_HOST="mysql-19285eb2-tahaelshrif1-7999.h.aivencloud.com"
+MYSQL_PORT=11520
+MYSQL_USER="avnadmin"
+MYSQL_PASSWORD="AVNS_aT0RGFafs6_34WFegSF"
+DATABASE_NAME="trafficManagerSignals"

mqtt_manager.py

@@ -0,0 +1,98 @@
+
+import paho.mqtt.client as mqtt
+import time
+import threading
+from dotenv import load_dotenv
+import os
+from traffic_utils import status_acc,status_emr,status_free
+import paho.mqtt.client as mqtt
+import time
+
+load_dotenv(dotenv_path="keys.env")
+
+mqtt_broker = os.getenv("MQTT_BROKER")
+mqtt_port = int(os.getenv("MQTT_PORT"))
+MQTT_PARTOPIC_state = os.getenv("MQTT_PARTOPIC_state")
+MQTT_SUBTOPIC_reply = os.getenv("MQTT_SUBTOPIC_reply")
+MQTT_SUBTOPIC_msg = os.getenv("MQTT_SUBTOPIC_msg")
+topic_rep = lambda x: MQTT_PARTOPIC_state+"/"+x +"/"+ MQTT_SUBTOPIC_reply
+topic_msg = lambda x:MQTT_PARTOPIC_state+"/"+x +"/"+ MQTT_SUBTOPIC_msg
+
+
+singlas_state = {
+    "1698478721" :status_free,
+    "6082411793":status_free}
+
+client = mqtt.Client()
+
+def intialize_mqtt_brokers():
+    client.connect(mqtt_broker, mqtt_port)
+    for tl_id in singlas_state.keys():
+        client.subscribe(topic_rep(tl_id))
+
+def start_mqtt_publishers():
+    for tl_id in singlas_state.keys():
+        threading.Thread(target=mqtt_publisher_loop, args=(tl_id,), daemon=True).start()
+
+
+def get_State(tl_id):
+    if tl_id in singlas_state:
+        return singlas_state[tl_id]
+    else:
+        return "FREE"
+    
+
+def mqtt_publisher_loop(tl_id):
+    #client = mqtt.Client()
+    #client.connect(mqtt_broker, mqtt_port)
+    client.loop_start()
+
+    last_state = None
+    while True:
+        current_state = get_State(tl_id)
+        if current_state != last_state:
+            topic = topic_msg(tl_id)
+            client.publish(topic, current_state)
+            print(f"📡 Published {current_state} to {topic}")
+            last_state = current_state
+        time.sleep(1)
+
+intialize_mqtt_brokers()
+
+
+
+'''
+from fastapi import WebSocket, WebSocketDisconnect
+import asyncio
+
+@app.websocket("/ws/state")
+async def websocket_get_state(websocket: WebSocket):
+    await websocket.accept()
+    try:
+        tl_id = await websocket.receive_text()
+        print(f"✅ Client subscribed to {tl_id}")
+
+        last_state = None
+        while True:
+            current_state = get_State(tl_id)
+            if current_state != last_state:
+                await websocket.send_text(current_state)
+                last_state = current_state
+            await asyncio.sleep(1)  # كل ثانية يشيك التغيير
+    except WebSocketDisconnect:
+        print("❌ Client disconnected")
+
+        
+@app.get("/test-db")
+def test_db():
+    try:
+        connection = get_db()
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT 1")
+            result = cursor.fetchone()
+        connection.close()
+        return {"status": "connected", "result": result}
+    except Exception as e:
+        return {"status": "error", "detail": str(e)}
+
+'''

requirements.txt

@@ -4,3 +4,4 @@ requests
 pymysql
 sqlalchemy
 paho-mqtt
+python-dotenv

traffic_utils.py

@@ -0,0 +1,52 @@
+import requests 
+from database_manager import *
+import time
+from mqtt_manager import singlas_state
+
+status_free = "FREE"
+status_emr = "EMR"
+status_acc = "ACC"
+
+def get_traffic_in_container(coordinates):
+    (min_lat, max_lat, min_lon, max_lon) = get_rectangle_container(coordinates)
+    overpass_url = "http://overpass-api.de/api/interpreter"
+    query = f"""
+    [out:json];
+    node["highway"="traffic_signals"]({min_lat},{min_lon},{max_lat},{max_lon});
+    out body;
+    """
+    response = requests.get(overpass_url, params={'data': query})
+    data = response.json()
+    return [(el['lat'], el['lon']) for el in data['elements']]
+
+
+def get_rectangle_container(coordinates):
+    lons = [i for i, _ in coordinates]
+    lats = [i for _, i in coordinates]
+    min_lat, max_lat = min(lats), max(lats)
+    min_lon, max_lon = min(lons), max(lons)
+    return min_lat, max_lat, min_lon, max_lon
+
+def open_signal(tl_id,state,time_):
+    singlas_state[tl_id] = state 
+    time.sleep(time_)
+    singlas_state[tl_id] = status_free 
+
+
+def check_signals(coords):
+    connection = get_db()
+    found_signals = []
+    try:
+        with connection.cursor() as cursor:
+            for lat, lon in coords:
+                cursor.execute("""
+                    SELECT tl_id_sumo FROM traffic_signals
+                    WHERE ABS(lat - %s) < 0.0001 AND ABS(lon - %s) < 0.0001
+                """, (lon,lat))
+                result = cursor.fetchone()
+                if result:
+                    found_signals.append(result['tl_id_sumo'])
+    finally:
+        connection.close()
+
+    return found_signals