feat: use database as app source

src/app.py

@@ -1,13 +1,37 @@
 """App to display images in a gallery"""
 
-import os
 import numpy as np
 import streamlit as st
 from image_preprocessing import get_image_caption, get_images, resize_image
-from typing import Any
+
+from data_models.sql_connection import get_db_connection
 from data_models.park_manager import ParkManager
+from data_models.image_manager import ImageManager
+from data_models.openai_manager import OpenAIManager
+
+
+engine, session = get_db_connection()
+park_manager = ParkManager()
+image_manager = ImageManager()
+openai_manager = OpenAIManager()
+
+
+def get_image_predictions(images: list[dict]) -> dict:
+    """Get predictions for a list of images
+
+    Args:
+        images (list): list of images to get predictions for
+
+    Returns:
+        list: list of predictions for the images
+    """
+    predictions = {}
+    for image in images:
+        predictions[image["id"]] = openai_manager.get_predictions(image["id"])
+    return predictions
+
 
-def image_gallery(images: list) -> None:
+def image_gallery(images: list[dict], predictions:dict) -> None:
     """Display a gallery of images in a streamlit app
 
     Args:
@@ -18,52 +42,53 @@ def image_gallery(images: list) -> None:
     columns = st.columns(3)
     for index, image in enumerate(images):
         with columns[index % 3]:
-            resized_image = resize_image(image, target_w=1080, target_h=720)
-            image_caption = get_image_caption(image)
+            resized_image = resize_image(image["name"], target_w=1080, target_h=720)
             st.image(
-                np.array(resized_image),
-                width=200,
-                caption=image_caption,
+                np.array(resized_image), width=200, caption=f"Image id #{image['id']}"
             )
+            if not image["id"] in predictions or predictions[image["id"]] is None:
+                st.write("No predictions available")
+                continue
+            st.json(predictions[image["id"]])
 
 
-# def get_park_list() -> list:
-#     """Get the list of parks from the images folder
-
-#     Returns:
-#         list: list of parks
-#     """
+def get_park_list() -> list:
+    return park_manager.get_parks()
 
-#     path = "./images/"
-#     parks = os.listdir(path)
-#     return sorted(parks)
 
+def get_park_images(park: str) -> list:
+    return image_manager.get_images_by_park(park)
 
-def get_park_list() -> list:
-    park_manager = ParkManager()
-    # return [park["name"] for park in park_manager.get_parks()]
-    parks = [park[1] for park in park_manager.get_parks()]
-    return parks
 
-
-def sidebar() -> Any | None:
+def sidebar() -> dict | None:
     """Create a sidebar to select the park
 
     Returns:
         selected_park: selected park in the sidebar
     """
-    return st.sidebar.selectbox("Park List", get_park_list(), None)
+    park_list = get_park_list()
+
+    def _park_list_formatter(park):
+        return park["name"]
+
+    return st.sidebar.selectbox(
+        label="Park List",
+        options=park_list,
+        index=None,
+        format_func=_park_list_formatter,
+    )
 
 
 def main() -> None:
     """Main function to run the app"""
-    path = "./images/"
     park = sidebar()
     if not park:
         st.write("Please select a park")
         return
-    images = get_images(path + park)
-    image_gallery(images)
+
+    images = get_park_images(park["id"])
+    predictions = get_image_predictions(images)
+    image_gallery(images, predictions)
 
 
 if __name__ == "__main__":

src/data_models/image_manager.py

@@ -56,7 +56,7 @@ class ImageManager:
         except Exception as e:
             raise Exception(f"An error occurred while getting the image ID: {e}")
 
-    def get_images(self, park_id):
+    def get_images_by_park(self, park_id):
         """
         Get all images from the `images` table.
         Args:
@@ -71,7 +71,7 @@ class ImageManager:
             result = self.session.execute(
                 query, {"park_id": park_id} if park_id else {}
             )
-            return [dict(row) for row in result]
+            return [row._asdict() for row in result]
         except Exception as e:
             raise Exception(f"Erreur lors de la récupération des images : {e}")
 

src/data_models/openai_manager.py

@@ -60,6 +60,23 @@ class OpenAIManager:
             self.session.rollback()
             raise Exception(f"An error occurred while adding the predictions: {e}")
 
+    def get_predictions(self, img_id):
+        """
+        Get predictions from the `openai_predictions` table.
+
+        Args:
+            img_id (int): ID of the image where the bounding box was detected.
+
+        Returns:
+            dict: Predictions of the image.
+        """
+        query = text("SELECT * FROM openai_predictions WHERE img_id = :img_id")
+        try:
+            result = self.session.execute(query, {"img_id": img_id}).fetchone()
+            return result._asdict() if result else None
+        except Exception as e:
+            raise Exception(f"An error occurred while getting the predictions: {e}")
+
     def close_connection(self):
         """Close the connection."""
         self.session.close()

src/data_models/park_manager.py

@@ -36,7 +36,7 @@ class ParkManager:
         try:
             query = text("SELECT * FROM parks")
             result = self.session.execute(query)
-            return [row for row in result.fetchall()]
+            return [row._asdict() for row in result.fetchall()]
         except Exception as e:
             raise Exception(f"An error occurred while fetching the parks: {e}")
 

src/image_preprocessing.py

@@ -22,7 +22,7 @@ def get_images(folder: str) -> list:
     return images
 
 
-def resize_image(img_path: str, target_w: int = 200, target_h: int = 200) -> Image:
+def resize_image(img_path: str, target_w: int = 200, target_h: int = 200) -> Image.Image:
     """Resize an image
 
     Args: