Merge pull request #127 from openUC2/mergemaster

enable sending camera triggers from board

Author: beniroquai

uc2rest/UC2Client.py

@@ -28,6 +28,7 @@
 from .message import Message
 from .can import CAN
 from .canota import CANOTA
+from .camera_trigger import CameraTrigger
 try:
     import requests
 except:
@@ -151,6 +152,9 @@ def __init__(self, host=None, port=31950, serialport=None, identity="UC2_Feather
 
         # initialize messaging
         self.message = Message(self)
+        
+        # initialize camera trigger callback handler
+        self.camera_trigger = CameraTrigger(self)
 
         # initialize module controller
         self.modules = Modules(parent=self)

uc2rest/camera_trigger.py

@@ -0,0 +1,197 @@
+"""
+Camera trigger callback module for UC2-REST.
+
+This module handles camera trigger signals from the firmware ({"cam":1})
+to enable software triggering based on hardware events during stage scanning.
+"""
+
+import numpy as np
+import time
+import json
+
+
+gTIMEOUT = 1  # seconds to wait for a response from the ESP32
+
+
+class CameraTrigger(object):
+    """
+    This class parses incoming camera trigger signals from the ESP32 firmware.
+    
+    When the firmware sends {"cam":1}, this module triggers registered callbacks
+    which can be used for software-triggered image acquisition during stage scanning.
+    
+    Example usage:
+        import uc2rest
+        
+        ESP32 = uc2rest.UC2Client(serialport=port, baudrate=500000)
+        
+        # Register callback for camera trigger
+        def my_camera_callback(data):
+            print(f"Camera trigger received: {data}")
+            # Trigger image acquisition here
+            
+        ESP32.camera_trigger.register_callback(0, my_camera_callback)
+    """
+    
+    def __init__(self, parent=None, nCallbacks=10):
+        """
+        Initialize camera trigger handler.
+        
+        Args:
+            parent: Parent UC2Client instance
+            nCallbacks: Maximum number of callback functions to register
+        """
+        self._parent = parent
+        self.nCallbacks = nCallbacks
+        
+        # Track trigger count for diagnostics
+        self._trigger_count = 0
+        self._last_trigger_time = None
+        
+        # Initialize callback functions
+        self._callbackPerKey = {}
+        self.init_callback_functions(self.nCallbacks)
+        
+        # Register callback for camera trigger on serial loop
+        if hasattr(self._parent, "serial"):
+            self._parent.serial.register_callback(self._callback_camera_trigger, pattern="cam")
+
+    def _callback_camera_trigger(self, data):
+        """
+        Parse camera trigger message from firmware.
+        
+        Expected JSON format:
+        {
+            "cam": 1  # Trigger signal
+        }
+        
+        or with additional data:
+        {
+            "cam": {
+                "trigger": 1,
+                "frame_id": 123,
+                "illumination": 0
+            }
+        }
+        
+        Args:
+            data: JSON data dictionary from firmware
+        """
+        try:
+            # Update trigger statistics
+            self._trigger_count += 1
+            self._last_trigger_time = time.time()
+            
+            # Extract trigger information
+            cam_data = data.get("cam", {})
+            
+            # Handle simple trigger ({"cam": 1})
+            if isinstance(cam_data, (int, float)):
+                trigger_info = {
+                    "trigger": int(cam_data),
+                    "frame_id": self._trigger_count,
+                    "timestamp": self._last_trigger_time
+                }
+                self._parent.logger.debug(f"Camera trigger received: {trigger_info}")
+            else:
+                # Handle extended trigger data
+                trigger_info = {
+                    "trigger": cam_data.get("trigger", 1),
+                    "frame_id": cam_data.get("frame_id", self._trigger_count),
+                    "illumination": cam_data.get("illumination", -1),
+                    "timestamp": self._last_trigger_time
+                }
+                self._parent.logger.debug(f"Camera trigger with data received: {trigger_info}") 
+            
+            # Call all registered callbacks
+            for key, callback in self._callbackPerKey.items():
+                if callback is not None and callable(callback):
+                    try:
+                        callback(trigger_info)
+                    except Exception as callback_error:
+                        print(f"Error in camera trigger callback {key}: {callback_error}")
+                        
+        except Exception as e:
+            print(f"Error in _callback_camera_trigger: {e}")
+
+    def init_callback_functions(self, nCallbacks=10):
+        """
+        Initialize callback function dictionary.
+        
+        Args:
+            nCallbacks: Number of callback slots to create
+        """
+        self._callbackPerKey = {}
+        self.nCallbacks = nCallbacks
+        for i in range(nCallbacks):
+            self._callbackPerKey[i] = None
+
+    def register_callback(self, key, callback):
+        """
+        Register a callback function for camera trigger events.
+        
+        Args:
+            key: Integer key (0 to nCallbacks-1) for this callback
+            callback: Function to call when trigger is received.
+                     Function signature: callback(trigger_info: dict)
+                     
+        Example:
+            def on_camera_trigger(info):
+                print(f"Frame {info['frame_id']} triggered at {info['timestamp']}")
+                camera.snap_image()
+                
+            ESP32.camera_trigger.register_callback(0, on_camera_trigger)
+        """
+        if key < 0 or key >= self.nCallbacks:
+            raise ValueError(f"Callback key must be between 0 and {self.nCallbacks-1}")
+        self._callbackPerKey[key] = callback
+
+    def unregister_callback(self, key):
+        """
+        Remove a registered callback.
+        
+        Args:
+            key: Integer key of callback to remove
+        """
+        if key in self._callbackPerKey:
+            self._callbackPerKey[key] = None
+
+    def clear_all_callbacks(self):
+        """Remove all registered callbacks."""
+        self.init_callback_functions(self.nCallbacks)
+
+    def get_trigger_count(self):
+        """
+        Get the total number of triggers received since initialization.
+        
+        Returns:
+            int: Number of camera triggers received
+        """
+        return self._trigger_count
+
+    def get_last_trigger_time(self):
+        """
+        Get the timestamp of the last trigger received.
+        
+        Returns:
+            float or None: Unix timestamp of last trigger, or None if no triggers received
+        """
+        return self._last_trigger_time
+
+    def reset_trigger_count(self):
+        """Reset the trigger counter to zero."""
+        self._trigger_count = 0
+        self._last_trigger_time = None
+
+    def get_trigger_stats(self):
+        """
+        Get statistics about trigger events.
+        
+        Returns:
+            dict: Dictionary with trigger statistics
+        """
+        return {
+            "total_triggers": self._trigger_count,
+            "last_trigger_time": self._last_trigger_time,
+            "callbacks_registered": sum(1 for cb in self._callbackPerKey.values() if cb is not None)
+        }

uc2rest/motor.py

@@ -950,8 +950,8 @@ def stop_stage_scanning(self):
 
 
 
-    def start_stage_scanning(self, xstart=0, xstep=1000, nx=20, ystart=0, ystep=1000, ny=10, tsettle=5, tExposure=50, illumination=(0,0,0,0), led=0, speed=20000, acceleration=None):
-		#	{"task": "/motor_act", "stagescan": {"xStart": 0, "yStart": 0, "xStep": 500, "yStep": 500, "nX": 10, "nY": 10, "tPre": 50, "tPost": 50, "illumination": [0, 1, 0, 0], "led": 255}}
+    def start_stage_scanning(self, xstart=0, xstep=1000, nx=20, ystart=0, ystep=1000, ny=10, zstart=0, zstep=1000, nz=10, tsettle=5, tExposure=50, illumination=(0,0,0,0), led=0, speed=20000, acceleration=None):
+		#	{"task": "/motor_act", "stagescan": {"xStart": 0, "yStart": 0, "zStart": 0, "xStep": 500, "yStep": 500, "zStep": 500, "nX": 10, "nY": 10, "nZ": 10, "tPre": 50, "tPost": 50, "illumination": [0, 1, 0, 0], "led": 255}}
         if acceleration is None:
             acceleration = self.DEFAULT_ACCELERATION
         path = "/motor_act"
@@ -970,16 +970,19 @@ def start_stage_scanning(self, xstart=0, xstep=1000, nx=20, ystart=0, ystep=1000
                 "led": led,
                 "accel": self.DEFAULT_ACCELERATION,  # default acceleration
                 "speed": speed,  # default speed
+                "zStart": zstart / self.stepSizeZ,
+                "zStep": zstep / self.stepSizeZ,
+                "nZ": nz,
             }
         }
         r = self._parent.post_json(path, payload)
         return r
 
     def start_stage_scanning_by_coordinates(self, coordinates, tPre=50, tPost=50, led=100, illumination=[50, 75, 100, 125], stopped=0): 
         '''
-        Example: {"task": "/motor_act", "stagescan": {"coordinates": [{"x": 100, "y": 200}, {"x": 300, "y": 400}, {"x": 500, "y": 600}], "tPre": 50, "tPost": 50, "led": 100, "illumination": [50, 75, 100, 125], "stopped": 0}}
+        Example: {"task": "/motor_act", "stagescan": {"coordinates": [{"x": 100, "y": 200, "z": 0}, {"x": 300, "y": 400, "z": 0}, {"x": 500, "y": 600, "z": 0}], "tPre": 50, "tPost": 50, "led": 100, "illumination": [50, 75, 100, 125], "stopped": 0}}
         
-        coordinates: list of dictionaries with x and y coordinates
+        coordinates: list of dictionaries with x, y and z coordinates
         tPre: time before exposure in ms
         tPost: exposure time - time after action
         led: led value for illumination