feat: implementation of FSM, and invokation for first phases

- fsm implementation uses the `transitions` package.
- added unique keys to the input forms, so can check when all are filled
- included a basic viz/feedback on the state

requirements.txt

@@ -10,7 +10,8 @@ streamlit_folium==0.23.1
 
 # backend 
 datasets==3.0.2
-
+## FSM
+transitions==0.9.2 
 
 # running ML models
 

src/input/input_handling.py

@@ -30,6 +30,43 @@ spoof_metadata = {
     "time": None,
 }
 
+def check_inputs_are_set(empty_ok:bool=False, debug:bool=False) -> bool:
+    """
+    Checks if all expected inputs have been entered 
+    
+    Implementation: via the Streamlit session state.
+
+    Args:
+        empty_ok (bool): If True, returns True if no inputs are set. Default is False.
+        debug (bool): If True, prints and logs the status of each expected input key. Default is False.
+    Returns:
+        bool: True if all expected input keys are set, False otherwise.
+    """
+    filenames = st.session_state.image_filenames
+    if len(filenames) == 0:
+        return empty_ok
+    
+
+
+    exp_input_key_stubs = ["input_latitude", "input_longitude"]
+    #exp_input_key_stubs = ["input_latitude", "input_longitude", "input_author_email", "input_date", "input_time", "input_image_selector"]
+    vals = []
+    for image_filename in filenames:
+        for stub in exp_input_key_stubs:
+            key = f"{stub}_{image_filename}"
+            val = None
+            if key in st.session_state:
+                val = st.session_state[key]
+            vals.append(val)
+            if debug:
+                msg = f"{key:15}, {(val is not None):8}, {val}"
+                m_logger.debug(msg)
+                print(msg)
+    
+    return all([v is not None for v in vals])
+
+        
+
 def setup_input(
     viewcontainer: DeltaGenerator=None,
     _allowed_image_types: list=None, ) -> InputObservation:
@@ -66,7 +103,8 @@ def setup_input(
     uploaded_files = viewcontainer.file_uploader("Upload an image", type=allowed_image_types, accept_multiple_files=True)
     observations = {}
     images = {}
-    image_hashes =[]
+    image_hashes = []
+    filenames = []
     if uploaded_files is not None:
         for file in uploaded_files:
 
@@ -76,6 +114,7 @@ def setup_input(
             # load image using cv2 format, so it is compatible with the ML models
             file_bytes = np.asarray(bytearray(file.read()), dtype=np.uint8)
             filename = file.name
+            filenames.append(filename) 
             image = cv2.imdecode(file_bytes, 1)
             # Extract and display image date-time
             image_datetime = None  # For storing date-time from image
@@ -84,12 +123,18 @@ def setup_input(
         
 
             # 3. Latitude Entry Box
-            latitude = viewcontainer.text_input("Latitude for "+filename, spoof_metadata.get('latitude', ""))
+            latitude = viewcontainer.text_input(
+                "Latitude for "+filename, 
+                spoof_metadata.get('latitude', ""),
+                key=f"input_latitude_{filename}")
             if latitude and not is_valid_number(latitude):
                 viewcontainer.error("Please enter a valid latitude (numerical only).")
                 m_logger.error(f"Invalid latitude entered: {latitude}.")
             # 4. Longitude Entry Box
-            longitude = viewcontainer.text_input("Longitude for "+filename, spoof_metadata.get('longitude', ""))
+            longitude = viewcontainer.text_input(
+                "Longitude for "+filename, 
+                spoof_metadata.get('longitude', ""),
+                key=f"input_longitude_{filename}")
             if longitude and not is_valid_number(longitude):
                 viewcontainer.error("Please enter a valid longitude (numerical only).")
                 m_logger.error(f"Invalid latitude entered: {latitude}.")
@@ -118,4 +163,6 @@ def setup_input(
     st.session_state.files = uploaded_files
     st.session_state.observations = observations
     st.session_state.image_hashes = image_hashes
+    st.session_state.image_filenames = filenames
+
 

src/main.py

@@ -15,10 +15,11 @@ disable_caching()
 
 import whale_gallery as gallery
 import whale_viewer as viewer
-from input.input_handling import setup_input
+from input.input_handling import setup_input, check_inputs_are_set
 from maps.alps_map import present_alps_map
 from maps.obs_map import present_obs_map
 from utils.st_logs import setup_logging, parse_log_buffer
+from utils.workflow_state import WorkflowFSM, FSM_STATES
 from classifier.classifier_image import cetacean_classify
 from classifier.classifier_hotdog import hotdog_classify
 
@@ -48,6 +49,11 @@ if "handler" not in st.session_state:
 if "image_hashes" not in st.session_state:
     st.session_state.image_hashes = []
 
+# TODO: ideally just use image_hashes, but need a unique key for the ui elements
+# to track the user input phase; and these are created before the hash is generated. 
+if "image_filenames" not in st.session_state:
+    st.session_state.image_filenames = []
+
 if "observations" not in st.session_state:
     st.session_state.observations = {}
 
@@ -69,6 +75,23 @@ if "whale_prediction1" not in st.session_state:
 if "tab_log" not in st.session_state:
     st.session_state.tab_log = None
     
+if "workflow_fsm" not in st.session_state:
+    # create and init the state machine
+    st.session_state.workflow_fsm = WorkflowFSM(FSM_STATES)
+    
+# add progress indicator to session_state
+if "progress" not in st.session_state:
+    with st.sidebar:
+        st.session_state.disp_progress = [st.empty(), st.empty()]
+
+def refresh_progress():
+    with st.sidebar:
+        tot = st.session_state.workflow_fsm.num_states
+        cur_i = st.session_state.workflow_fsm.current_state_index
+        cur_t = st.session_state.workflow_fsm.current_state
+        st.session_state.disp_progress[0].markdown(f"*Progress: {cur_i}/{tot}. Current: {cur_t}.*")
+        st.session_state.disp_progress[1].progress(cur_i/tot)
+        
 
 def main() -> None:
     """
@@ -102,6 +125,10 @@ def main() -> None:
         st.tabs(["Cetecean classifier", "Hotdog classifier", "Map", "*:gray[Dev:coordinates]*", "Log", "Beautiful cetaceans"])
     st.session_state.tab_log = tab_log
 
+    refresh_progress()    
+    # add button to sidebar, with the callback to refesh_progress
+    st.sidebar.button("Refresh Progress", on_click=refresh_progress)
+    
 
     # create a sidebar, and parse all the input (returned as `observations` object)
     setup_input(viewcontainer=st.sidebar)
@@ -181,14 +208,25 @@ def main() -> None:
         
 
     # Display submitted observation
-    if st.sidebar.button("Validate"):
-        # create a dictionary with the submitted observation
-        tab_log.info(f"{st.session_state.observations}")
-        df = pd.DataFrame(st.session_state.observations, index=[0])
-        with tab_coords:
-            st.table(df)
+    all_inputs_set = check_inputs_are_set(debug=True)
+    if not all_inputs_set:
+        st.sidebar.button(":gray[*Validate*]", disabled=True, help="Please fill in all fields.")
+
+    else:
+        if st.session_state.workflow_fsm.is_in_state('init'):
+            st.session_state.workflow_fsm.complete_current_state()
         
+        if st.sidebar.button("**Validate**"):
+            if st.session_state.workflow_fsm.is_in_state('data_entry_complete'):
+                st.session_state.workflow_fsm.complete_current_state()
         
+            # create a dictionary with the submitted observation
+            tab_log.info(f"{st.session_state.observations}")
+            df = pd.DataFrame(st.session_state.observations, index=[0])
+            with tab_coords:
+                st.table(df)
+            
+            
 
         
     # inside the inference tab, on button press we call the model (on huggingface hub)
@@ -240,6 +278,9 @@ def main() -> None:
             hotdog_classify(pipeline_hot_dog, tab_hotdogs)
             
             
+    # after all other processing, we can show the stage/state
+    refresh_progress()
+
 
 if __name__ == "__main__":
     main()

src/utils/workflow_state.py

@@ -0,0 +1,92 @@
+from transitions import Machine
+from typing import List
+
+OKBLUE = '\033[94m'
+OKGREEN = '\033[92m'
+OKCYAN = '\033[96m'
+FAIL = '\033[91m'
+ENDC = '\033[0m'
+
+
+FSM_STATES = ['init', 'data_entry_complete', 'data_entry_validated', 'ml_classification_started', 'ml_classification_completed', 'manual_inspection_completed', 'data_uploaded']
+
+
+class WorkflowFSM:
+    def __init__(self, state_sequence: List[str]):
+        self.state_sequence = state_sequence
+        self.state_dict = {state: i for i, state in enumerate(state_sequence)}
+        
+        # Create state machine
+        self.machine = Machine(
+            model=self,
+            states=state_sequence,
+            initial=state_sequence[0],
+        )
+
+        # For each state (except the last), add a completion transition to the next state
+        for i in range(len(state_sequence) - 1):
+            current_state = state_sequence[i]
+            next_state = state_sequence[i + 1]
+            
+            self.machine.add_transition(
+                trigger=f'complete_{current_state}',
+                source=current_state,
+                dest=next_state,
+                conditions=[f'is_in_{current_state}']
+            )
+
+            # Dynamically add a condition method for each state
+            setattr(self, f'is_in_{current_state}',
+                   lambda s=current_state: self.is_in_state(s))
+
+        # Add callbacks for logging
+        self.machine.before_state_change = self._log_transition
+        self.machine.after_state_change = self._post_transition
+
+    def is_in_state(self, state_name: str) -> bool:
+        """Check if we're currently in the specified state"""
+        return self.state == state_name
+
+    def complete_current_state(self) -> bool:
+        """
+        Signal that the current state is complete.
+        Returns True if state transition occurred, False otherwise.
+        """
+        current_state = self.state
+        trigger_name = f'complete_{current_state}'
+        
+        if hasattr(self, trigger_name):
+            try:
+                trigger_func = getattr(self, trigger_name)
+                trigger_func()
+                return True
+            except:
+                return False
+        return False
+
+    @property
+    def current_state(self) -> str:
+        """Get the current state name"""
+        return self.state
+    
+    @property
+    def current_state_index(self) -> int:
+        """Get the current state index"""
+        return self.state_dict[self.state]
+
+    @property
+    def num_states(self) -> int:
+        return len(self.state_sequence)
+ 
+
+    def _log_transition(self):
+        # TODO: use logger, not printing. 
+        self._cprint(f"[FSM] -> Transitioning from {self.current_state}")
+
+    def _post_transition(self):
+        # TODO: use logger, not printing. 
+        self._cprint(f"[FSM] -| Transitioned to {self.current_state}")
+
+    def _cprint(self, msg:str, color:str=OKCYAN):
+        """Print colored message"""
+        print(f"{color}{msg}{ENDC}")