Fix/workflows and imaging

requirements.txt

@@ -30,3 +30,4 @@ yapf==0.40.1
 zipp==3.17.0
 ultralytics==8.3.128
 websockets==15.0.1
+depthai==2.32.0.0

samples/analysis.py

@@ -13,10 +13,11 @@
 cam = camera.RPiCamera()
 debugger = debug.ImageAnalysisDebugger()
 debugger = None
-img_analysis = analysis.ImageAnalysisDelegate(detector, cam, location_provider,
-                                              debugger)
+img_analysis = analysis.ImageAnalysisDelegate(
+    detector, cam, location_provider, debugger
+)
 
-#img_analysis.subscribe(lambda _, __, ___: debugger.show())
+# img_analysis.subscribe(lambda _, __, ___: debugger.show())
 img_analysis.subscribe(lambda _, lon, lat: print(lon, lat))
 img_analysis.start()
 

samples/battery_status.py

@@ -14,7 +14,7 @@ def wait_for_voltage():
             print("voltage: ", voltage)
         except ValueError:
             pass
-            #print("no data yet")
+            # print("no data yet")
         time.sleep(0.5)
 
 

samples/bucket_test.py

@@ -1,23 +1,20 @@
-from src.modules.imaging.bucket_detector import BucketDetector
-from src.modules.imaging.camera import DebugCameraFromDir
-from src.modules.imaging.location import DebugLocationProvider
-# from src.modules.imaging.analysis import ImageAnalysisDelegate
-from src.modules.imaging.debug_analysis import DebugImageAnalysisDelegate
+if __name__ == "__main__":
+    from src.modules.imaging.bucket_detector import BucketDetector
+    from src.modules.imaging.camera import DebugCameraFromDir
+    from src.modules.imaging.location import DebugLocationProvider
+    from src.modules.imaging.debug_analysis import DebugImageAnalysisDelegate
 
+    cam = DebugCameraFromDir("images")
+    det = BucketDetector("samples/models/n640.pt")
+    location = DebugLocationProvider()
+    analysis = DebugImageAnalysisDelegate(det, cam, location)
+    # analysis = ImageAnalysisDelegate(det, cam, location)
 
+    def test(a1, a2):
+        if a1 is not None and a2 is not None:
+            print(a1)
+            print(a2)
 
-cam = DebugCameraFromDir("images")
-det = BucketDetector("samples/models/n640.pt")
-location = DebugLocationProvider()
-analysis = DebugImageAnalysisDelegate(det, cam, location)
-# analysis = ImageAnalysisDelegate(det, cam, location)
-
-def test(a1, a2):
-    if a1 != None and a2 != None:
-        print(a1)
-        print(a2)
-
-analysis.subscribe(test)
-
-analysis.start()
+    analysis.subscribe(test)
 
+    analysis.start()

samples/bucket_test_simple.py

@@ -1,47 +1,42 @@
-from src.modules.imaging.bucket_detector import BucketDetector
-from src.modules.imaging.camera import DebugCameraFromDir, RPiCamera, DebugCamera
-from src.modules.imaging.location import DebugLocationProvider
-from src.modules.imaging.analysis import ImageAnalysisDelegate
-
-from ultralytics import YOLO
 import os
-
 import cv2
-from PIL import Image
 import time
 
-cam = RPiCamera(0)
-model_path = 'samples/models'
-models = os.listdir(model_path)
-models = ["best.pt"]
-for file in models:
-    model = YOLO(os.path.join(model_path, file))
-    i = len(os.listdir("photos"))
-    while True: 
-        image = cam.capture()
-        image.save(f"photos/{i}.png")
-        results = model(image)
+if __name__ == "__main__":
+    from src.modules.imaging.camera import RPiCamera
+    from ultralytics import YOLO
 
-        result = results[0] # because one image
-        a = result.plot()
-        # b = Image.fromarray(a)
-        # b.save(f"results/{i}.png")
-        # 
-        cv2.imshow(f'Result for model: {file}', a)
-        cv2.waitKey(0)
-        cv2.destroyAllWindows()
-        time.sleep(0.5)
-        i += 1
-        # boxes = result.boxes
+    cam = RPiCamera(0)
+    model_path = "samples/models"
+    models = os.listdir(model_path)
+    models = ["best.pt"]
+    for file in models:
+        model = YOLO(os.path.join(model_path, file))
+        i = len(os.listdir("photos"))
+        while True:
+            image = cam.capture()
+            image.save(f"photos/{i}.png")
+            results = model(image)
 
-        # if boxes is not None and len(boxes) > 0:
-        #     best_box = boxes[boxes.conf.argmax()]
-        #     (x1, y1, x2, y2) = best_box.xyxy[0].tolist()  # box [x1, y1, x2, y2]
-        #     conf = best_box.conf.item() # confidence threshold
-        #     if conf < 0.5:
-        #         print("not confident")
-        #     else:
-        #         print("confident")
-        # else:
-        #     print("no bounding box")
+            result = results[0]  # because one image
+            a = result.plot()
+            # b = Image.fromarray(a)
+            # b.save(f"results/{i}.png")
+            #
+            cv2.imshow(f"Result for model: {file}", a)
+            cv2.waitKey(0)
+            cv2.destroyAllWindows()
+            time.sleep(0.5)
+            i += 1
+            # boxes = result.boxes
 
+            # if boxes is not None and len(boxes) > 0:
+            #     best_box = boxes[boxes.conf.argmax()]
+            #     (x1, y1, x2, y2) = best_box.xyxy[0].tolist()  # box [x1, y1, x2, y2]
+            #     conf = best_box.conf.item() # confidence threshold
+            #     if conf < 0.5:
+            #         print("not confident")
+            #     else:
+            #         print("confident")
+            # else:
+            #     print("no bounding box")

samples/camera.py

@@ -7,7 +7,7 @@
 
 cam = RPiCamera()
 
-#cam.set_size((1000, 1000))
+# cam.set_size((1000, 1000))
 
 print("Taking a picture in 3s.")
 

samples/emu_connection.py

@@ -1,6 +1,5 @@
 from src.modules.emu import Emu
 import time
-import json
 
 emu = Emu("res")
 
@@ -25,9 +24,12 @@
 # test different logs
 for i in range(6):
     print(f"sending log {i}")
-    if i % 3 == 0: severity = "normal"
-    elif i % 3 == 1: severity = "warning"
-    else: severity = "error"
+    if i % 3 == 0:
+        severity = "normal"
+    elif i % 3 == 1:
+        severity = "warning"
+    else:
+        severity = "error"
     emu.send_log(f"log text {i}", severity)
     time.sleep(1)
 

samples/geofence_test.py

@@ -1,41 +1,38 @@
 import json
-from src.modules.autopilot.lander import Lander
-from src.modules.georeference import inference_georeference
-
-with open('./samples/geofence.json', 'r') as f:
-    geofence = json.load(f)['features'][0]['geometry']['coordinates'][0]
-
-
-origin = [-113.54815575690603, 53.495546666117804]
-
-geofence = inference_georeference.Geofence_to_XY(origin, geofence)
-print(geofence)
-
-lander = Lander(0, 0, geofence)
-'''
-point1 = [53.4958, -113.5477]
-point2 = [53.4956, -113.5481]
-point3 = [53.4956, -113.5475]
-point4 = [53.4955, -113.5488]
-point5 = [53.4955, -113.5488]
-'''
-point1 = [0, 0]
-point2 = [5, 5]
-point3 = [10, 10]
-point4 = [-5, -5]
-point5 = [-10, 10]
-point6 = [30.4,
-          30.4]
-point7 = [-113.54827877970348,
-          53.49573842746909]
-
-
-print(lander.geofence_check(point1))
-print(lander.geofence_check(point2))
-print(lander.geofence_check(point3))
-print(lander.geofence_check(point4))
-print(lander.geofence_check(point5))
-print(lander.geofence_check(point6))
-print(lander.geofence_check(point7))
-
 
+if __name__ == "__main__":
+    from src.modules.autopilot.lander import Lander
+    from src.modules.georeference import inference_georeference
+
+if __name__ == "__main__":
+    with open("./samples/geofence.json", "r") as f:
+        geofence = json.load(f)["features"][0]["geometry"]["coordinates"][0]
+
+    origin = [-113.54815575690603, 53.495546666117804]
+
+    geofence = inference_georeference.Geofence_to_XY(origin, geofence)
+    print(geofence)
+
+    lander = Lander(0, 0, geofence)
+    """
+    point1 = [53.4958, -113.5477]
+    point2 = [53.4956, -113.5481]
+    point3 = [53.4956, -113.5475]
+    point4 = [53.4955, -113.5488]
+    point5 = [53.4955, -113.5488]
+    """
+    point1 = [0, 0]
+    point2 = [5, 5]
+    point3 = [10, 10]
+    point4 = [-5, -5]
+    point5 = [-10, 10]
+    point6 = [30.4, 30.4]
+    point7 = [-113.54827877970348, 53.49573842746909]
+
+    print(lander.geofence_check(point1))
+    print(lander.geofence_check(point2))
+    print(lander.geofence_check(point3))
+    print(lander.geofence_check(point4))
+    print(lander.geofence_check(point5))
+    print(lander.geofence_check(point6))
+    print(lander.geofence_check(point7))

samples/img_test.py

@@ -1,20 +1,16 @@
-from src.modules.imaging.detector import IrDetector
-from src.modules.imaging.analysis import ImageAnalysisDelegate
-from src.modules.imaging.camera import RPiCamera
-from src.modules.imaging.location import DebugLocationProvider
-
-
-camera = RPiCamera()
-detector = IrDetector()
-location = DebugLocationProvider()
-
-
-def test(img, _):
-    print("Image taken")
-
-
-
-analysis = ImageAnalysisDelegate(detector, camera, location)
-analysis.subscribe(test)
-
-analysis.start()
+if __name__ == "__main__":
+    from src.modules.imaging.detector import IrDetector
+    from src.modules.imaging.analysis import ImageAnalysisDelegate
+    from src.modules.imaging.camera import RPiCamera
+    from src.modules.imaging.location import DebugLocationProvider
+    camera = RPiCamera()
+    detector = IrDetector()
+    location = DebugLocationProvider()
+
+    def test(img, _):
+        print("Image taken")
+
+    analysis = ImageAnalysisDelegate(detector, camera, location)
+    analysis.subscribe(test)
+
+    analysis.start()

samples/kml_mock.py

@@ -3,9 +3,9 @@
 
 kml = KMLGenerator()
 
-hotspot_1 = LatLong(1,1)
-hotspot_2 = LatLong(1,2)
-hotspot_3 = LatLong(0,0)
+hotspot_1 = LatLong(1, 1)
+hotspot_2 = LatLong(1, 2)
+hotspot_3 = LatLong(0, 0)
 
 kml.push(hotspot_1)
 kml.push(hotspot_2)

samples/oakd.py

@@ -16,9 +16,11 @@
 emu = Emu("tmp/log")
 i = 0
 
+
 def print_conn():
     print("connecton made")
 
+
 emu.set_on_connect(print_conn)
 latest_capture: DepthCapture | None = None
 
@@ -31,6 +33,7 @@ def print_conn():
 camera_thread = threading.Thread(target=camera.start(), daemon=True)
 camera_thread.start()
 
+
 def send_img(message):
     global latest_capture, i, ft_num
 
@@ -48,6 +51,7 @@ def send_img(message):
 
         i += 1
 
+
 def measure(message):
     global latest_capture
 
@@ -63,12 +67,12 @@ def measure(message):
             distPoint = latest_capture.get_point(point["x"], point["y"])
             send = {
                 "requestId": requestId,
-                "type": "point", 
+                "type": "point",
                 "message": {
                     "x": distPoint[0],
                     "y": distPoint[1],
                     "z": distPoint[2]
-                    }
+                }
             }
 
             # If point is (0,0,0), measurement is invalid