Integrate detection service

rpi/services/detection_service.py

@@ -3,6 +3,7 @@ import numpy as np
 from pathlib import Path
 from ultralytics.engine.results import Results
 
+from hardware.camera.camera import Camera
 from models.detection.detector import Detector
 from models.detection.board_manager import BoardManager
 from models.detection.pieces_manager import PiecesManager
@@ -18,6 +19,8 @@ class DetectionService:
 
     scale_size : tuple[int, int]
 
+    camera :  Camera
+
     def __init__(self):
         current_file = Path(__file__).resolve()
         project_root = current_file.parent.parent
@@ -28,11 +31,23 @@ class DetectionService:
         self.pieces_manager = PiecesManager()
         self.board_manager = BoardManager()
         self.scale_size = (800, 800)
+        self.camera = Camera()
+
+    def start(self):
+        self.camera.open()
+
+    def stop(self):
+        self.camera.close()
+
+    def analyze_single_frame(self) -> str | None:
+        frame = self.camera.take_photo()
+        fen = self.__get_fen(frame)
+        return fen
 
 
-    def run_complete_detection(self, frame : np.ndarray, display=False) -> dict[str, list[Results]] :
-        pieces_prediction = self.run_pieces_detection(frame)
-        edges_prediction = self.run_edges_detection(frame)
+    def __run_complete_detection(self, frame : np.ndarray, display=False) -> dict[str, list[Results]] :
+        pieces_prediction = self.__run_pieces_detection(frame)
+        edges_prediction = self.__run_edges_detection(frame)
 
         if display:
             edges_annotated_frame = edges_prediction[0].plot()
@@ -42,22 +57,22 @@ class DetectionService:
         return { "edges" : edges_prediction, "pieces" : pieces_prediction}
 
 
-    def run_pieces_detection(self, frame : np.ndarray, display=False) -> list[Results]:
+    def __run_pieces_detection(self, frame : np.ndarray, display=False) -> list[Results]:
         prediction = self.pieces_detector.make_prediction(frame)
         if display:
             self.__display_frame(prediction[0].plot())
         return prediction
 
 
-    def run_edges_detection(self, frame : np.ndarray, display=False) -> list[Results]:
+    def __run_edges_detection(self, frame : np.ndarray, display=False) -> list[Results]:
         prediction = self.edges_detector.make_prediction(frame)
         if display:
             self.__display_frame(prediction[0].plot())
         return prediction
 
 
-    def get_fen(self, frame : np.ndarray) -> str | None:
-        result = self.run_complete_detection(frame)
+    def __get_fen(self, frame : np.ndarray) -> str | None:
+        result = self.__run_complete_detection(frame)
 
         edges_prediction = result["edges"]
         pieces_prediction = result["pieces"]
@@ -75,7 +90,6 @@ class DetectionService:
 
         return self.pieces_manager.board_to_fen(board)
 
-
     def __display_frame(self, frame : np.ndarray):
         cv2.namedWindow("Frame", cv2.WINDOW_NORMAL)
         cv2.resizeWindow("Frame", self.scale_size[0], self.scale_size[1])
@@ -83,24 +97,3 @@ class DetectionService:
         cv2.waitKey(0)
         cv2.destroyAllWindows()
         return
-
-
-if __name__ == "__main__" :
-    import os
-    import random
-
-    service = DetectionService()
-
-    img_folder = "../training/datasets/pieces/unified/test/images/"
-
-    test_images = os.listdir(img_folder)
-
-    rnd = random.randint(0, len(test_images) - 1)
-    img_path = os.path.join(img_folder, test_images[rnd])
-
-    image = cv2.imread(img_path)
-
-    fen = service.get_fen(image)
-    print(fen)
-
-    service.run_complete_detection(image, display=True)

rpi/services/game_service.py

@@ -0,0 +1,23 @@
+from services.clock_service import ClockService
+from services.detection_service import DetectionService
+
+
+class GameService:
+
+    detection_service : DetectionService
+    clock_service : ClockService
+
+    def __init__(self):
+        self.detection_service = DetectionService()
+        self.clock_service = ClockService()
+
+    def start(self, time_control : int, increment : int ) -> None:
+        self.clock_service.start(time_control, increment)
+        self.clock_service.set_on_terminated(self.stop)
+
+    def stop(self):
+        self.clock_service.stop()
+
+    def make_move(self) -> None:
+        fen = self.detection_service.analyze_single_frame()
+        print(fen)