Backend

pyproject.toml

@@ -14,6 +14,7 @@ authors = [
 ]
 
 dependencies = [
+
 ]
 
 [project.optional-dependencies]
@@ -64,4 +65,9 @@ rpi = [
     "matplotlib>=3.10.5",
     "opencv-python>=4.12.0.88",
     "scipy>=1.15.3",
+    "websockets>=16.0",
+    "fastapi>=0.128.0",
+    "uvicorn>=0.40.0",
+    "websocket>=0.2.1",
+    "websockets>=16.0",
 ]

scripts/commande_PS4.py

@@ -18,7 +18,7 @@ def envoie_donnee(Voiture): #si utilisation de la voiture directement
     bus = smbus.SMBus(1)
     while True:
             try :
-                data = struct.pack('<ff', float(round(Voiture.vitesse_mms)), float(round(Voiture.direction_d)))
+                data = struct.pack('<ff', float(round(Voiture.vitesse_mms)), float(round(Voiture.direction)))
                 bus.write_i2c_block_data(SLAVE_ADDRESS, 0, list(data))
                 #time.sleep(0.00005)
             except Exception as e:
@@ -64,20 +64,20 @@ def kill(self):
         self.running = False
 
     @property
-    def vitesse_d(self):
+    def target_speed(self):
         return self.controller.vitesse_mms
 
     @property
-    def direction_d(self):
-        return self.controller.direction_d
+    def direction(self):
+        return self.controller.direction
 
 class MyController(Controller):
 
     def __init__(self, **kwargs):
         super().__init__(**kwargs)
         self.log = logging.getLogger(__name__)
         self.vitesse_mms = 0 # vitesse initiale en métre par milliseconde
-        self.direction_d = 0 # angle initiale des roues en degrés
+        self.direction = 0 # angle initiale des roues en degrés
         self.filtered = 0
         self.alpha = 0.3
         self.running = 0
@@ -120,12 +120,12 @@ def on_R1_release(self):
     def on_L3_right(self,value):
         # print("x_r :", value, "degré : ",map_range(value,-32767, 32767, 60, 120))
         dir = map_range(value, 0, 32767, 0, MAX_ANGLE)
-        self.direction_d = dir
+        self.direction = dir
 
     def on_L3_left(self,value):
         #print("x_r :", value, "degré : ",map_range(value,-32767, 0, -MAX_ANGLE, 0 ))
         dir = self.stable_direction(value)
-        self.direction_d = dir
+        self.direction = dir
 
 
     def on_L2_press(self, value):

scripts/remote_control_controller.py

@@ -10,16 +10,16 @@
 sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
 
 def envoie_donnee():
-    global vitesse_m, direction_d
+    global vitesse_m, direction
     while True:
-        packet = struct.pack("ff", vitesse_m, direction_d)
+        packet = struct.pack("ff", vitesse_m, direction)
         sock.sendto(packet, ("192.168.1.10", 5556))
         time.sleep(0.05)
 
 ###################################################
 # Paramètres véhicule
 ###################################################
-direction_d = 0
+direction = 0
 vitesse_m = 0
 
 vitesse_max_m_s_soft = 2
@@ -30,9 +30,9 @@ def map_range(x, in_min, in_max, out_min, out_max):
     return (x - in_min) * (out_max - out_min) / (in_max - in_min) + out_min
 
 def set_direction_degre(angle_degre):
-    global direction_d
-    direction_d = angle_degre
-    print(direction_d, vitesse_m)
+    global direction
+    direction = angle_degre
+    print(direction, vitesse_m)
 
 def set_vitesse_m_ms(vit):
     global vitesse_m

src/HL/Autotech_constant.py

@@ -42,6 +42,7 @@
 TEXT_LEFT_OFFSET = 3 # Offset from the left of the screen to ensure no cuttoff
 
 
+
 script_dir = os.path.dirname(os.path.abspath(__file__))
 MODEL_PATH = os.path.join(script_dir, "model_CNN1D.onnx")  # Allows the model to be loaded from the same directory as the script regardless of the current working directory (aka where the script is run from)
 
@@ -59,4 +60,9 @@
 # In our case Higher temperature means less agressive driving and lower temperature means more aggressive driving.
 
 
-LOGGING_LEVEL = logging.DEBUG # can be either NOTSET, DEBUG, INFO, WARNING, ERROR, CRITICAL
+LOGGING_LEVEL = logging.DEBUG # can be either NOTSET, DEBUG, INFO, WARNING, ERROR, CRITICAL
+
+#Startup
+CAMERA_STREAM_ON_START = True #If True the camera stream will start at the launch of the car
+BACKEND_ON_START = True #If True the backend will start at the launch of the car
+LIDAR_STREAM_ON_START = True #If True the lidar stream will start at the launch of the car

src/HL/Serveur_mq.py

@@ -33,6 +33,7 @@
 from src.HL.actionneur_capteur.Camera import ProgramStreamCamera
 from src.HL.programme.module_initialisation import Initialisation
 from src.HL.programme.Car import Ai_Programme
+from backend import BackendAPI
 
 from Autotech_constant import I2C_NUMBER_DATA_RECEIVED, I2C_SLEEP_RECEIVED, I2C_SLEEP_ERROR_LOOP, TEXT_HEIGHT, TEXT_LEFT_OFFSET
 
@@ -77,7 +78,14 @@ def __init__(self):
 
         self.initialisation_module = Initialisation(self,Camera,Lidar,ToF, I2c_arduino)
 
-        self.programme = [SshProgramme(), self.initialisation_module, Ai_Programme(self), PS4ControllerProgram(), RemoteControl(), ProgramStreamCamera(self), Poweroff()]
+        self.programme = [SshProgramme(),
+                         self.initialisation_module, 
+                         Ai_Programme(self), 
+                         PS4ControllerProgram(), 
+                         RemoteControl(), 
+                         ProgramStreamCamera(self), 
+                         BackendAPI(self, host="0.0.0.0", port=8001, site_dir="/home/intech/CoVAPSy/src/HL/site_controle"),
+                         Poweroff()]
         self.log.debug("Programmes chargés: %s", [type(p).__name__ for p in self.programme])
 
         # donnée de l'écran
@@ -101,6 +109,15 @@ def tof(self):
     @property
     def arduino_I2C(self):
         return self.initialisation_module.arduino_I2C
+
+    @property
+    def target_speed(self):
+        return self.programme[self.last_programme_control].target_speed
+
+    @property
+    def direction(self):
+        return self.programme[self.last_programme_control].direction
+
     #-----------------------------------------------------------------------------------------------------
     # affichage de l'écrans
     #-----------------------------------------------------------------------------------------------------

src/HL/actionneur_capteur/Camera.py

@@ -41,7 +41,7 @@ def outputframe(self, frame, keyframe=True):
 
 from src.HL.programme.Camera_serv import StreamServer, StreamHandler, StreamOutput, frame_buffer
 from src.HL.programme.programme import Program
-from src.HL.Autotech_constant import PORT_STREAMING_CAMERA, SIZE_CAMERA_X, SIZE_CAMERA_Y, FRAME_RATE, CAMERA_QUALITY, STREAM_PATH
+from src.HL.Autotech_constant import PORT_STREAMING_CAMERA, SIZE_CAMERA_X, SIZE_CAMERA_Y, FRAME_RATE, CAMERA_QUALITY, STREAM_PATH, CAMERA_STREAM_ON_START
 
 class ProgramStreamCamera(Program):
     def __init__(self,serveur):
@@ -50,6 +50,10 @@ def __init__(self,serveur):
         self.serveur = serveur
         self.running = False
         self.controls_car = False
+
+        if CAMERA_STREAM_ON_START:
+            self.start()
+
 
     @property
     def camera(self):

src/HL/actionneur_capteur/masterI2C.py

@@ -9,7 +9,7 @@ class I2c_arduino:
     def __init__(self,serveur):
         self.log = logging.getLogger(__name__)
         self.serveur = serveur
-        self.vitesse_r = 0
+        self.current_speed = 0
         self.send_running = True
         self.receive_running = True
 
@@ -32,8 +32,9 @@ def start_send(self):
         self.log.info("Thread I2C loop démarré")
         while self.send_running:
             try :
-                data = struct.pack('<ff', float(round(self.serveur.programme[self.serveur.last_programme_control].vitesse_d)), float(round(self.serveur.programme[self.serveur.last_programme_control].direction_d)))
+                data = struct.pack('<ff', float(round(self.serveur.target_speed)), float(round(self.serveur.direction)))
                 self.bus.write_i2c_block_data(SLAVE_ADDRESS, 0, list(data))
+                time.sleep(1e-5) # Short delay to prevent overwhelming the bus
             except Exception as e:
                 self.log.error("Erreur I2C write: %s", e, exc_info=True)
                 time.sleep(I2C_SLEEP_ERROR_LOOP)
@@ -52,7 +53,7 @@ def start_received(self):
                 # on enregistre les valeur
                 self.voltage_lipo = list_valeur[0]
                 self.voltage_nimh = list_valeur[1]
-                self.vitesse_r = list_valeur[2]
+                self.current_speed = list_valeur[2]
             else:
                 self.log.warning("I2C: taille inattendue (%d au lieu de %d)", len(data), length)
             time.sleep(I2C_SLEEP_RECEIVED)