From f3705ebd979356f45198f9a4e5e0850f4ab0bff7 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Tue, 12 May 2026 16:11:32 +0300 Subject: [PATCH 01/20] feat: multi-camera support keyed by video_aim+camera_idx Interface previously held a single self.camera. Replace with self.cameras dict keyed by f"{video_aim}_{camera_idx}", so two cameras can coexist in the same setup (e.g. one openfield cam feeding DLC plus one passive eye cam recording to disk). Always including camera_idx in the key keeps the mapping "physical camera -> file" stable across config edits. - Interface._initialize_camera loops over all SetupConfiguration.Camera rows for the setup_conf_idx, builds Camera instances with per-camera filenames (f"{animal}_{session}_{aim}_{idx}"). - Interface.release iterates over self.cameras. - WebCam and PiCamera now accept camera_num (defaults to 0) and use it for cv2.VideoCapture / Picamera2 device selection instead of the hard-coded 0. - Camera.__init__ derives the h5 timestamp filename from self.filename (which carries the per-camera disambiguator), so two cameras in the same session no longer overwrite each other's h5. Co-Authored-By: Claude Opus 4.7 --- src/ethopy/core/interface.py | 64 ++++++++++++++++++++++----------- src/ethopy/interfaces/Camera.py | 22 +++++++----- 2 files changed, 57 insertions(+), 29 deletions(-) diff --git a/src/ethopy/core/interface.py b/src/ethopy/core/interface.py index 87d2c2e..180d5d4 100755 --- a/src/ethopy/core/interface.py +++ b/src/ethopy/core/interface.py @@ -70,7 +70,7 @@ def __init__( self.logger = exp.logger if exp else None self.position = Port() self.position_tmst: int = 0 - self.camera = None + self.cameras: Dict[str, Any] = {} self.ports: List[Port] = [] self.pulse_rew: Dict[int, Dict] = {} self.duration: Dict[int, float] = {} @@ -119,32 +119,54 @@ def _initialize_hardware(self) -> None: self._initialize_camera() def _initialize_camera(self) -> None: - """Initialize camera if configured in setup.""" + """Initialize all cameras configured for this setup, keyed by ``video_aim``. + + Each row in ``SetupConfiguration.Camera`` becomes one ``Camera`` instance, + stored in ``self.cameras`` under ``f"{video_aim}_{camera_idx}"``. Always + including camera_idx in the key keeps the mapping "physical camera -> + file" stable across config edits — if a setup later drops one of two + openfield cameras, the remaining one keeps its idx suffix instead of + silently becoming the new "first openfield". + """ setup_cameras = self.logger.get( schema="interface", table="SetupConfiguration.Camera", fields=["setup_conf_idx"], ) - if self.exp.setup_conf_idx in setup_cameras: - camera_params = self.logger.get( - schema="interface", - table="SetupConfiguration.Camera", - key=f"setup_conf_idx={self.exp.setup_conf_idx}", - as_dict=True, - )[0] + if self.exp.setup_conf_idx not in setup_cameras: + return + camera_rows = self.logger.get( + schema="interface", + table="SetupConfiguration.Camera", + key=f"setup_conf_idx={self.exp.setup_conf_idx}", + as_dict=True, + ) + + filename_base = ( + f"{self.logger.trial_key['animal_id']}" + f"_{self.logger.trial_key['session']}" + ) + + for camera in camera_rows: + video_aim = camera.pop("video_aim") + # camera_idx doubles as the OS /dev/videoN index (V4L2) or the + # libcamera index (Picamera2). Future work: allow string device + # paths (e.g. udev symlinks) so cameras are addressed by role + # rather than relying on stable kernel enumeration order. + camera_num = camera.pop("camera_idx") + key = f"{video_aim}_{camera_num}" camera_class = getattr( - import_module("ethopy.interfaces.Camera"), camera_params["discription"] + import_module("ethopy.interfaces.Camera"), camera["discription"] ) - - self.camera = camera_class( - filename=f"{self.logger.trial_key['animal_id']}" - f"_{self.logger.trial_key['session']}", + self.cameras[key] = camera_class( + filename=f"{filename_base}_{key}", logger=self.logger, logger_timer=self.logger.logger_timer, - video_aim=camera_params.pop("video_aim"), - **camera_params, + video_aim=video_aim, + camera_num=camera_num, + **camera, ) def give_liquid(self, port: int, duration: Optional[float] = 0) -> None: @@ -213,11 +235,11 @@ def cleanup(self) -> None: """Clean up interface resources.""" def release(self) -> None: - """Release hardware resources, especially camera.""" - if self.camera: - log.info("Releasing camera resources.") - if self.camera.recording.is_set(): - self.camera.stop_rec() + """Release hardware resources, especially cameras.""" + for aim, cam in self.cameras.items(): + log.info("Releasing camera resources (video_aim=%s).", aim) + if cam.recording.is_set(): + cam.stop_rec() def load_calibration(self) -> None: """Load port calibration data from database. diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 184cef7..74feb64 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -122,11 +122,10 @@ def __init__( ), block=True, ) - h5s_filename = ( - f"animal_id_{logger.trial_key['animal_id']}" - f"_session_{logger.trial_key['session']}.h5" - ) - self.filename_tmst = "videosssctmst" + h5s_filename + # Tie the h5 timestamp file to self.filename so per-camera disambiguators + # (video_aim / camera_idx suffix) propagate here too — otherwise two + # cameras in the same session would overwrite each other's h5. + self.filename_tmst = f"videotmst_{self.filename}.h5" logger.log_recording( dict( rec_aim="sync", @@ -346,6 +345,7 @@ def __init__( resolution_x: int = 1280, resolution_y: int = 720, fps: int = 30, + camera_num: int = 0, logger_timer: Optional["Timer"] = None, **kwargs, ): @@ -355,6 +355,7 @@ def __init__( Args: resolution (Tuple[int, int], optional): Resolution of the webcam. Defaults to (640, 480). + camera_num (int): /dev/videoN index used by V4L2. Defaults to 0. Raises: ImportError: If the cv2 package is not installed. @@ -362,6 +363,7 @@ def __init__( """ self.fps = fps + self.camera_num = camera_num self.video_output = None self.dataset = None self.tmst_output = None @@ -385,7 +387,7 @@ def __init__( "You can install cv2 using pip:\n" 'sudo pip3 install opencv-python"' ) - self.camera = cv2.VideoCapture(0, cv2.CAP_V4L2) + self.camera = cv2.VideoCapture(self.camera_num, cv2.CAP_V4L2) if not self.camera.isOpened(): raise RuntimeError( "No camera is available. Please check if the camera is connected and functional." @@ -484,7 +486,7 @@ def camera_opened(self, camera): return True def recording_init(self): - self.camera = cv2.VideoCapture(0, cv2.CAP_V4L2) + self.camera = cv2.VideoCapture(self.camera_num, cv2.CAP_V4L2) if not self.camera.isOpened(): raise RuntimeError( "No camera is available. Please check if the camera is connected and functional." @@ -589,6 +591,7 @@ def __init__( fps: int = 15, sensor_mode: int = 1, exposure: int = 10000, + camera_num: int = 0, file_format: str = "rgb", logger_timer: Optional["Timer"] = None, **kwargs, @@ -605,6 +608,7 @@ def __init__( self.sensor_mode = sensor_mode self.resolution = (resolution_x, resolution_y) self.exposure = exposure + self.camera_num = camera_num self.file_format = file_format self.tmst_output = None @@ -675,7 +679,9 @@ def recording_init(self) -> None: def init_cam(self) -> "Picamera2": """Initialize the camera.""" - picam2 = Picamera2() + # Future: support string device identifiers so cameras can be addressed + # by role (via udev/libcamera config) instead of enumeration order. + picam2 = Picamera2(camera_num=self.camera_num) _mode = picam2.sensor_modes[self.sensor_mode] config = picam2.create_video_configuration( raw={"size": _mode["size"], "format": _mode["format"].format}, From 9b9e40a5f57e5c6d553810e8a9d074b84b3edfcb Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Tue, 12 May 2026 16:13:12 +0300 Subject: [PATCH 02/20] fix: make Camera.stop_rec idempotent and drop release() TOCTOU race MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The previous release() guarded each stop_rec call with `if cam.recording.is_set()`. With multi-camera setups that race fires: between mp.Process.start() and the child reaching `recording.set()`, a spawned-but-not-yet-recording subprocess holds resources (open file handles, FFmpegWriter, possibly a /dev/videoN FD) but is_set() returns False, so stop_rec is skipped and the subprocess leaks. - Camera.stop_rec is now idempotent: returns early if camera_process is None, swallows the rare ValueError from close() racing with terminate, joins twice (timeout=30 + timeout=5 after terminate), and clears self.camera_process so a second call is a no-op. - Interface.release calls stop_rec unconditionally — the stop event is sufficient to signal shutdown regardless of subprocess lifecycle step. Co-Authored-By: Claude Opus 4.7 --- src/ethopy/core/interface.py | 10 +++++++--- src/ethopy/interfaces/Camera.py | 23 +++++++++++++++++------ 2 files changed, 24 insertions(+), 9 deletions(-) diff --git a/src/ethopy/core/interface.py b/src/ethopy/core/interface.py index 180d5d4..b413ac5 100755 --- a/src/ethopy/core/interface.py +++ b/src/ethopy/core/interface.py @@ -235,11 +235,15 @@ def cleanup(self) -> None: """Clean up interface resources.""" def release(self) -> None: - """Release hardware resources, especially cameras.""" + """Release hardware resources, especially cameras. + + Calls ``stop_rec`` unconditionally — checking ``recording.is_set()`` here + would race against cameras still inside their startup path (TOCTOU). + ``stop_rec`` is idempotent and safe to call before recording begins. + """ for aim, cam in self.cameras.items(): log.info("Releasing camera resources (video_aim=%s).", aim) - if cam.recording.is_set(): - cam.stop_rec() + cam.stop_rec() def load_calibration(self) -> None: """Load port calibration data from database. diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 74feb64..f82d9ee 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -294,18 +294,29 @@ def dequeue(self, frame_queue: Queue) -> None: time.sleep(0.01) def stop_rec(self) -> None: + """Stop the camera subprocess. + + Idempotent and safe to call before the camera has finished starting up + (i.e. before ``self.recording`` has been set). The ``stop`` event is + sufficient to signal shutdown regardless of which lifecycle step the + child process has reached — relying on ``recording.is_set()`` here + would race against the spawn. """ - Set the stop event and join the write runner. - """ + if self.camera_process is None: + return self.stop.set() time.sleep(3) - # TODO: use join and close (possible issue due to h5 files) self.camera_process.join(timeout=30) - # check if the process is still alive if self.camera_process.is_alive(): self.camera_process.terminate() - else: + self.camera_process.join(timeout=5) + try: self.camera_process.close() + except ValueError: + # Process still alive after terminate() — rare, but don't mask it + # by raising during cleanup. The OS will reap it once it exits. + pass + self.camera_process = None @abstractmethod def rec(self) -> None: @@ -345,7 +356,7 @@ def __init__( resolution_x: int = 1280, resolution_y: int = 720, fps: int = 30, - camera_num: int = 0, + camera_num: int = 0, logger_timer: Optional["Timer"] = None, **kwargs, ): From aebe6dc426da395e92e7d62e9e4d0c59c0b65f30 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Tue, 12 May 2026 16:15:04 +0300 Subject: [PATCH 03/20] fix: stat-based device probe and correct Recording row h5 paths MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two review-flagged issues, both now visible because of multi-camera: 1. WebCam.__init__ previously opened /dev/videoN via cv2.VideoCapture in the parent, released it, then forked a child that re-opened the same device. V4L2 drivers don't always tear down state synchronously on close(), so two cameras initialized back-to-back could race — second child opens before kernel finishes cleaning up first parent. Replaced with os.path.exists("/dev/videoN") — race-free, still catches the common config errors (missing device, wrong camera_num). Permission / device-busy errors surface from the child's open in recording_init() as before. 2. The h5 timestamp Recording row was logged with source_path/ target_path pointing at the video directory, but the file actually lives under logger.source_path + "Recordings/_/" (per Logger.createDataset). With per-camera h5 filenames now in play, the mismatch is more impactful — downstream consumers that resolve h5 paths via Recording rows would look in the wrong place. Mirror createDataset's path logic in Camera.__init__ so the row matches reality. Co-Authored-By: Claude Opus 4.7 --- src/ethopy/interfaces/Camera.py | 36 +++++++++++++++++++++++++++------ 1 file changed, 30 insertions(+), 6 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index f82d9ee..a36eca8 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -126,14 +126,28 @@ def __init__( # (video_aim / camera_idx suffix) propagate here too — otherwise two # cameras in the same session would overwrite each other's h5. self.filename_tmst = f"videotmst_{self.filename}.h5" + # The h5 is written later by Logger.createDataset (child process), + # but the DB row is inserted here in the parent. Mirror the path + # logic from createDataset (logger.py:950-961) so the Recording + # row points at where the file will actually land. + recordings_folder = ( + f"Recordings/{logger.trial_key['animal_id']}" + f"_{logger.trial_key['session']}/" + ) + h5_source_path = logger.source_path + recordings_folder + h5_target_path = ( + logger.target_path + recordings_folder + if os.path.isdir(logger.target_path) + else False + ) logger.log_recording( dict( rec_aim="sync", software="EthoPy", version="0.1", filename=self.filename_tmst, - source_path=self.source_path, - target_path=self.target_path, + source_path=h5_source_path, + target_path=h5_target_path, ), block=True, ) @@ -398,12 +412,22 @@ def __init__( "You can install cv2 using pip:\n" 'sudo pip3 install opencv-python"' ) - self.camera = cv2.VideoCapture(self.camera_num, cv2.CAP_V4L2) - if not self.camera.isOpened(): + # Stat-based probe: confirm the device node exists in the PARENT before + # spawning the subprocess. We deliberately do NOT open() the device here + # — opening + releasing leaves brief V4L2 driver state that races with + # the child's open in recording_init(), especially with two cameras + # initialized back-to-back. stat is cheap, race-free, and catches the + # common config errors (wrong camera_num, unplugged camera). Permission + # / device-busy errors still surface from the child's open later. + # Future: accept string device paths (e.g. udev symlinks like + # "/dev/cam_eye") so cameras can be addressed by role instead of + # relying on stable Linux enumeration order across reboots/replugs. + device_path = f"/dev/video{self.camera_num}" + if not os.path.exists(device_path): raise RuntimeError( - "No camera is available. Please check if the camera is connected and functional." + f"Camera device {device_path} not found. Check that the camera " + "is connected and that camera_num matches the intended /dev/videoN." ) - self.camera.release() super().__init__(kwargs["filename"], kwargs["logger"], kwargs["video_aim"]) def setup(self): From f80ded3602182906b8c74fb6fac98b3e1999adc0 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 14 May 2026 12:45:09 +0300 Subject: [PATCH 04/20] feat: improve clear_local_videos diagnostics cameras producing multiple source directories to drain: - copy_file: routine "transferred" and "deleted" logs demoted from info to debug (they fire per file; info-level for two cameras with hundreds of frame files each was overwhelming). - clear_local_videos: explicit error log when source/target paths are missing, with an actionable "Create it with: mkdir -p ..." hint; warning if no files were found to transfer; file count visible in the transfer start log; warning if the source folder isn't empty after transfer (indicates a partial failure that previously went silent). --- src/ethopy/interfaces/Camera.py | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index a36eca8..f2fc3f1 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -231,13 +231,13 @@ def copy_file(args): file, target = args try: shutil.copy(str(file), str(target / file.name)) - log.info(f"Transferred file: {file.name}") + log.debug(f"Transferred file: {file.name}") # Verify the file exists in the target directory if os.path.exists(str(target / file.name)) and os.path.getsize( str(file) ) == os.path.getsize(str(target / file.name)): os.remove(str(file)) - log.info(f"Deleted original file: {file.name}") + log.debug(f"Deleted original file: {file.name}") else: log.error(f"Failed to transfer file: {file.name}") except FileNotFoundError as ex: @@ -251,24 +251,38 @@ def clear_local_videos(self) -> None: target = Path(self.target_path) if not source.is_dir(): + log.error(f"Video source path does not exist: {source}") raise ValueError( f"Source path {source} does not exist or is not a directory." ) if not target.exists(): + log.error(f"Video target path does not exist: {target}") + log.error(f"Create it with: mkdir -p {target}") raise ValueError( f"Target path {target} does not exist or is not a directory." ) files = [(entry, target) for entry in source.iterdir() if entry.is_file()] - with Pool(processes=min(2, os.cpu_count() - 1)) as pool: - pool.map(self.copy_file, files) + if not files: + log.warning("No video files found to transfer") + else: + log.info( + f"Transferring {len(files)} video file(s) from {source} to {target}" + ) + with Pool(processes=min(2, os.cpu_count() - 1)) as pool: + pool.map(self.copy_file, files) # Clean up if the source directory is empty if not any(source.iterdir()): source.rmdir() - log.info(f"Deleted the empty folder: {source}") + log.info("Video transfer complete, cleaned up source folder") + else: + remaining = list(source.iterdir()) + log.warning( + f"Source folder not empty after transfer: {len(remaining)} items remaining" + ) def setup(self) -> None: """ From fd611435e2b89b7f64931b5d971a5d02600c67ef Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Wed, 24 Jun 2026 17:25:30 +0300 Subject: [PATCH 05/20] WebCam: add device_id support and resolver Add stable camera addressing and robustness fixes. - schema: add device_id (varchar) to SetupConfiguration with empty default, allowing cameras to be addressed by /dev/v4l/by-id symlink or a serial substring (falls back to camera index when empty). - WebCam: accept and store device_id - Replace simple /dev/videoN stat check with _resolve_device() that: returns camera_num when device_id is empty, accepts an existing path, or matches a single by-id index0 symlink under /dev/v4l/by-id. Sets self.device before spawning child to avoid race conditions. - Use self.device when opening cv2.VideoCapture and pin capture options (FOURCC=YUYV, CONVERT_RGB, BUFFERSIZE=1) to ensure consistent frames and reduce buffering/staleness. - Ensure captured grayscale frames are cast to uint8 and improve error/warning messages (fix typo, warn when property set() is rejected). These changes improve stable camera selection across reboots/USB re-plugs and make capture initialization more robust and predictable. --- src/ethopy/core/interface.py | 1 + src/ethopy/interfaces/Camera.py | 107 +++++++++++++++++++++++++------- 2 files changed, 86 insertions(+), 22 deletions(-) diff --git a/src/ethopy/core/interface.py b/src/ethopy/core/interface.py index b413ac5..dbb133d 100755 --- a/src/ethopy/core/interface.py +++ b/src/ethopy/core/interface.py @@ -467,6 +467,7 @@ class Camera(dj.Lookup, dj.Part): iso : smallint file_format : varchar(256) video_aim : enum('eye','body','openfield') + device_id="" : varchar(256) # stable /dev/v4l/by-id symlink or serial substring; empty = use camera_idx discription : varchar(256) """ diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index f2fc3f1..7dd864a 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -384,7 +384,7 @@ def __init__( resolution_x: int = 1280, resolution_y: int = 720, fps: int = 30, - camera_num: int = 0, + camera_num: int = 0, logger_timer: Optional["Timer"] = None, **kwargs, ): @@ -395,6 +395,14 @@ def __init__( resolution (Tuple[int, int], optional): Resolution of the webcam. Defaults to (640, 480). camera_num (int): /dev/videoN index used by V4L2. Defaults to 0. + Used only when ``device_id`` (kwarg) is empty. + + Keyword Args: + device_id (str): Stable hardware identifier for the camera. Either a + full path to a ``/dev/v4l/by-id/...`` symlink, or a substring of one + (e.g. a serial like "20231205_0001"). When set, it takes precedence + over ``camera_num`` and survives reboots / USB re-plugging. Empty + (default) falls back to the ``camera_num`` index. Raises: ImportError: If the cv2 package is not installed. @@ -418,6 +426,7 @@ def __init__( self.gain = kwargs.get("gain") self.contrast = kwargs.get("contrast") self.brightness = kwargs.get("brightness") + self.device_id = kwargs.get("device_id") or "" if not globals()["IMPORT_CV2"]: raise ImportError( @@ -426,24 +435,62 @@ def __init__( "You can install cv2 using pip:\n" 'sudo pip3 install opencv-python"' ) - # Stat-based probe: confirm the device node exists in the PARENT before - # spawning the subprocess. We deliberately do NOT open() the device here - # — opening + releasing leaves brief V4L2 driver state that races with - # the child's open in recording_init(), especially with two cameras - # initialized back-to-back. stat is cheap, race-free, and catches the - # common config errors (wrong camera_num, unplugged camera). Permission - # / device-busy errors still surface from the child's open later. - # Future: accept string device paths (e.g. udev symlinks like - # "/dev/cam_eye") so cameras can be addressed by role instead of - # relying on stable Linux enumeration order across reboots/replugs. - device_path = f"/dev/video{self.camera_num}" - if not os.path.exists(device_path): - raise RuntimeError( - f"Camera device {device_path} not found. Check that the camera " - "is connected and that camera_num matches the intended /dev/videoN." - ) + # Resolve + probe the device in the PARENT before spawning the + # subprocess. We deliberately do NOT open() the device here — opening + + # releasing leaves brief V4L2 driver state that races with the child's + # open in recording_init(), especially with two cameras initialized + # back-to-back. The resolver does a stat (cheap, race-free) and catches + # the common config errors (wrong device, unplugged camera) early. + # Permission / device-busy errors still surface from the child's open. + # self.device is set before super().__init__ spawns the child so the + # forked process inherits the resolved target. + self.device = self._resolve_device() super().__init__(kwargs["filename"], kwargs["logger"], kwargs["video_aim"]) + def _resolve_device(self) -> Union[int, str]: + """Resolve the configured camera to a target ``cv2.VideoCapture`` can open. + + Resolution order: + * no ``device_id`` -> the ``/dev/videoN`` index (``camera_num``); + legacy behaviour, fine when only one camera is connected. + * ``device_id`` is an existing path (e.g. a ``/dev/v4l/by-id/...`` + symlink) -> use it directly. + * otherwise ``device_id`` is treated as a substring (serial/model) + and matched against the capture-node (``index0``) symlinks under + ``/dev/v4l/by-id``. + + ``by-id`` symlinks are keyed on vendor/model/serial, so they survive + reboots and USB re-plugging — unlike the ``/dev/videoN`` index or + ``/dev/v4l/by-path`` (which encodes the physical port). + """ + if not self.device_id: + device_path = f"/dev/video{self.camera_num}" + if not os.path.exists(device_path): + raise RuntimeError( + f"Camera device {device_path} not found. Check that the " + "camera is connected and that camera_idx matches the " + "intended /dev/videoN." + ) + return self.camera_num + + if os.path.exists(self.device_id): + return self.device_id + + by_id = "/dev/v4l/by-id" + available = sorted(os.listdir(by_id)) if os.path.isdir(by_id) else [] + matches = [ + os.path.join(by_id, name) + for name in available + if self.device_id in name and name.endswith("index0") + ] + if len(matches) == 1: + return matches[0] + raise RuntimeError( + f"device_id {self.device_id!r} matched {len(matches)} capture " + f"device(s) under {by_id} (expected exactly 1). " + f"Available: {available}" + ) + def setup(self): """Setup the camera.""" out_vid_fn = self.source_path + self.filename + ".mp4" @@ -513,7 +560,7 @@ def get_frame(self) -> Tuple[bool, np.ndarray]: check, image = self.camera.read() if check: # If the capture was successful, convert the image to grayscale - image = np.squeeze(np.mean(image, axis=2)) + image = np.squeeze(np.mean(image, axis=2)).astype(np.uint8) return check, image def write_frame(self, item: Tuple[float, np.ndarray]) -> None: @@ -535,25 +582,34 @@ def camera_opened(self, camera): return True def recording_init(self): - self.camera = cv2.VideoCapture(self.camera_num, cv2.CAP_V4L2) + self.camera = cv2.VideoCapture(self.device, cv2.CAP_V4L2) if not self.camera.isOpened(): raise RuntimeError( "No camera is available. Please check if the camera is connected and functional." ) + # Pin a predictable capture format: YUYV decoded to 3-channel RGB (get_frame's + # grayscale mean over axis=2 relies on 3 channels), and BUFFERSIZE=1 so read() + # always returns the most recent frame instead of a stale buffered one. + self.camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc("Y", "U", "Y", "V")) + self.camera.set(cv2.CAP_PROP_CONVERT_RGB, 1) + self.camera.set(cv2.CAP_PROP_BUFFERSIZE, 1) self.camera.set(cv2.CAP_PROP_FPS, self.fps) self.res_set = self.set_resolution(self.resolution_x, self.resolution_y) if not self.res_set: logging.warning( - f"Camera resolution cannot be set tp {(self.resolution_x, self.resolution_y)}" - f",resize of frames will be used!!" + f"Camera resolution cannot be set to {(self.resolution_x, self.resolution_y)}" + f", resize of frames will be used!!" ) + # Every property below is opt-in. A camera that can't honour a setting + # (e.g. an analog frame-grabber) simply omits the key from its config, so + # the attribute is None and the setter is skipped. Provide real, non-zero + # values only for cameras that support them. if self.exposure: self.camera.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1) # Disable auto exposure self._set_camera_property(cv2.CAP_PROP_EXPOSURE, self.exposure) if self.wb_temperature: self.camera.set(cv2.CAP_PROP_AUTO_WB, 0.0) # Disable auto white balance self._set_camera_property(cv2.CAP_PROP_WB_TEMPERATURE, self.wb_temperature) - # If not provided in kwargs, they default to None and _set_camera_property skips them self._set_camera_property(cv2.CAP_PROP_SATURATION, self.saturation) self._set_camera_property(cv2.CAP_PROP_GAIN, self.gain) self._set_camera_property(cv2.CAP_PROP_CONTRAST, self.contrast) @@ -571,6 +627,13 @@ def _set_camera_property(self, property_id, value): f"Camera property {property_id} was set to " f"{actual_value}, not the requested {value}" ) + else: + # set() returned False: the driver rejected the property + # (common on analog grabbers / cameras that don't expose it). + logging.warning( + f"Camera property {property_id} is not supported by this " + f"camera; requested value {value} was ignored" + ) def rec(self): """ From 878f08702a8c2bfc09a4f911a5957615674bec10 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 25 Jun 2026 11:18:23 +0300 Subject: [PATCH 06/20] Refactor docs, logging, and camera handling Clean up docstrings and comments, reduce noisy prints, and add throttled logging and robustness fixes across interface, Camera, and DLC modules. - src/ethopy/core/interface.py: Simplified and clarified docstrings for camera initialization and release, emphasize stable camera->file mapping and explain unconditional stop_rec call to avoid races. - src/ethopy/interfaces/Camera.py: Clarified comments around camera naming and device resolution; made per-camera timestamp filename behavior explicit; added _last_frame_err_log to throttle per-frame read-error logs; simplified stop_rec and process-close handling; improved device-resolution error messages and camera property comments; ensured consistent behavior for capture format and buffer settings. - src/ethopy/interfaces/dlc.py: Added logging (logger import and module logger); convert frames to float32 before passing to DLCLive; replace prints with structured logging (debug/warning/exception); introduce a _throttled_log helper to avoid spamming logs in tight loops; add debug/warning checks around frame transfer delays, corner detection, and process lifecycle; handle case of no confident corner frames gracefully; use debug for shared memory unlink messages. Overall these changes improve observability, reduce console spam, make DLC inputs type-consistent, and clarify camera/device handling to make startup and shutdown more robust. --- src/ethopy/core/interface.py | 24 +++------- src/ethopy/interfaces/Camera.py | 84 ++++++++++++--------------------- src/ethopy/interfaces/dlc.py | 69 ++++++++++++++++++++------- 3 files changed, 87 insertions(+), 90 deletions(-) diff --git a/src/ethopy/core/interface.py b/src/ethopy/core/interface.py index dbb133d..f0e2849 100755 --- a/src/ethopy/core/interface.py +++ b/src/ethopy/core/interface.py @@ -119,14 +119,9 @@ def _initialize_hardware(self) -> None: self._initialize_camera() def _initialize_camera(self) -> None: - """Initialize all cameras configured for this setup, keyed by ``video_aim``. - - Each row in ``SetupConfiguration.Camera`` becomes one ``Camera`` instance, - stored in ``self.cameras`` under ``f"{video_aim}_{camera_idx}"``. Always - including camera_idx in the key keeps the mapping "physical camera -> - file" stable across config edits — if a setup later drops one of two - openfield cameras, the remaining one keeps its idx suffix instead of - silently becoming the new "first openfield". + """Initialize each configured camera into ``self.cameras``, keyed by + ``f"{video_aim}_{camera_idx}"`` so the camera->file mapping stays stable + across config edits. """ setup_cameras = self.logger.get( schema="interface", @@ -151,10 +146,7 @@ def _initialize_camera(self) -> None: for camera in camera_rows: video_aim = camera.pop("video_aim") - # camera_idx doubles as the OS /dev/videoN index (V4L2) or the - # libcamera index (Picamera2). Future work: allow string device - # paths (e.g. udev symlinks) so cameras are addressed by role - # rather than relying on stable kernel enumeration order. + # camera_idx is the /dev/videoN (V4L2) or libcamera index. camera_num = camera.pop("camera_idx") key = f"{video_aim}_{camera_num}" camera_class = getattr( @@ -235,12 +227,8 @@ def cleanup(self) -> None: """Clean up interface resources.""" def release(self) -> None: - """Release hardware resources, especially cameras. - - Calls ``stop_rec`` unconditionally — checking ``recording.is_set()`` here - would race against cameras still inside their startup path (TOCTOU). - ``stop_rec`` is idempotent and safe to call before recording begins. - """ + """Release hardware resources. stop_rec is idempotent, so it is called + unconditionally rather than racing a recording.is_set() check.""" for aim, cam in self.cameras.items(): log.info("Releasing camera resources (video_aim=%s).", aim) cam.stop_rec() diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 7dd864a..a8c2450 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -122,14 +122,10 @@ def __init__( ), block=True, ) - # Tie the h5 timestamp file to self.filename so per-camera disambiguators - # (video_aim / camera_idx suffix) propagate here too — otherwise two - # cameras in the same session would overwrite each other's h5. + # Per-camera name so two cameras in a session don't overwrite each + # other's h5; paths mirror Logger.createDataset (logger.py:950-961), + # which writes the file later in the child process. self.filename_tmst = f"videotmst_{self.filename}.h5" - # The h5 is written later by Logger.createDataset (child process), - # but the DB row is inserted here in the parent. Mirror the path - # logic from createDataset (logger.py:950-961) so the Recording - # row points at where the file will actually land. recordings_folder = ( f"Recordings/{logger.trial_key['animal_id']}" f"_{logger.trial_key['session']}/" @@ -322,13 +318,8 @@ def dequeue(self, frame_queue: Queue) -> None: time.sleep(0.01) def stop_rec(self) -> None: - """Stop the camera subprocess. - - Idempotent and safe to call before the camera has finished starting up - (i.e. before ``self.recording`` has been set). The ``stop`` event is - sufficient to signal shutdown regardless of which lifecycle step the - child process has reached — relying on ``recording.is_set()`` here - would race against the spawn. + """Stop the camera subprocess. Idempotent and safe to call before the + camera has finished starting up (the stop event alone signals shutdown). """ if self.camera_process is None: return @@ -341,9 +332,7 @@ def stop_rec(self) -> None: try: self.camera_process.close() except ValueError: - # Process still alive after terminate() — rare, but don't mask it - # by raising during cleanup. The OS will reap it once it exits. - pass + pass # still alive after terminate(); the OS reaps it once it exits self.camera_process = None @abstractmethod @@ -427,6 +416,7 @@ def __init__( self.contrast = kwargs.get("contrast") self.brightness = kwargs.get("brightness") self.device_id = kwargs.get("device_id") or "" + self._last_frame_err_log = 0.0 # throttles the per-frame read-error log if not globals()["IMPORT_CV2"]: raise ImportError( @@ -435,41 +425,25 @@ def __init__( "You can install cv2 using pip:\n" 'sudo pip3 install opencv-python"' ) - # Resolve + probe the device in the PARENT before spawning the - # subprocess. We deliberately do NOT open() the device here — opening + - # releasing leaves brief V4L2 driver state that races with the child's - # open in recording_init(), especially with two cameras initialized - # back-to-back. The resolver does a stat (cheap, race-free) and catches - # the common config errors (wrong device, unplugged camera) early. - # Permission / device-busy errors still surface from the child's open. - # self.device is set before super().__init__ spawns the child so the - # forked process inherits the resolved target. + # Probe in the parent (stat only, no open() — opening here races the + # child's open in recording_init). self.device is inherited by the fork. self.device = self._resolve_device() super().__init__(kwargs["filename"], kwargs["logger"], kwargs["video_aim"]) def _resolve_device(self) -> Union[int, str]: - """Resolve the configured camera to a target ``cv2.VideoCapture`` can open. - - Resolution order: - * no ``device_id`` -> the ``/dev/videoN`` index (``camera_num``); - legacy behaviour, fine when only one camera is connected. - * ``device_id`` is an existing path (e.g. a ``/dev/v4l/by-id/...`` - symlink) -> use it directly. - * otherwise ``device_id`` is treated as a substring (serial/model) - and matched against the capture-node (``index0``) symlinks under - ``/dev/v4l/by-id``. + """Resolve the camera to a target cv2.VideoCapture can open. - ``by-id`` symlinks are keyed on vendor/model/serial, so they survive - reboots and USB re-plugging — unlike the ``/dev/videoN`` index or - ``/dev/v4l/by-path`` (which encodes the physical port). + With no ``device_id`` this is the ``/dev/videoN`` index. Otherwise it is + an existing path (e.g. a ``/dev/v4l/by-id`` symlink), or a substring + matched against the ``index0`` symlinks under ``/dev/v4l/by-id`` — these + are keyed on vendor/model/serial, so they survive reboots and re-plugging. """ if not self.device_id: device_path = f"/dev/video{self.camera_num}" if not os.path.exists(device_path): raise RuntimeError( - f"Camera device {device_path} not found. Check that the " - "camera is connected and that camera_idx matches the " - "intended /dev/videoN." + f"Camera device {device_path} not found; check the camera is " + "connected and camera_idx matches the intended /dev/videoN." ) return self.camera_num @@ -483,12 +457,13 @@ def _resolve_device(self) -> Union[int, str]: for name in available if self.device_id in name and name.endswith("index0") ] + # Require exactly one: 0 means not found, 2+ means the substring is + # ambiguous and picking one would open an arbitrary camera. if len(matches) == 1: return matches[0] raise RuntimeError( - f"device_id {self.device_id!r} matched {len(matches)} capture " - f"device(s) under {by_id} (expected exactly 1). " - f"Available: {available}" + f"device_id {self.device_id!r} matched {len(matches)} device(s) " + f"under {by_id} (expected 1). Available: {available}" ) def setup(self): @@ -587,9 +562,8 @@ def recording_init(self): raise RuntimeError( "No camera is available. Please check if the camera is connected and functional." ) - # Pin a predictable capture format: YUYV decoded to 3-channel RGB (get_frame's - # grayscale mean over axis=2 relies on 3 channels), and BUFFERSIZE=1 so read() - # always returns the most recent frame instead of a stale buffered one. + # YUYV decoded to 3-channel RGB (get_frame averages over axis=2), and + # BUFFERSIZE=1 so read() returns the latest frame, not a stale buffered one. self.camera.set(cv2.CAP_PROP_FOURCC, cv2.VideoWriter_fourcc("Y", "U", "Y", "V")) self.camera.set(cv2.CAP_PROP_CONVERT_RGB, 1) self.camera.set(cv2.CAP_PROP_BUFFERSIZE, 1) @@ -600,10 +574,8 @@ def recording_init(self): f"Camera resolution cannot be set to {(self.resolution_x, self.resolution_y)}" f", resize of frames will be used!!" ) - # Every property below is opt-in. A camera that can't honour a setting - # (e.g. an analog frame-grabber) simply omits the key from its config, so - # the attribute is None and the setter is skipped. Provide real, non-zero - # values only for cameras that support them. + # Properties below are opt-in: omit the key from a camera's config (e.g. an + # analog grabber) to leave the value None and skip the setter. if self.exposure: self.camera.set(cv2.CAP_PROP_AUTO_EXPOSURE, 1) # Disable auto exposure self._set_camera_property(cv2.CAP_PROP_EXPOSURE, self.exposure) @@ -628,8 +600,7 @@ def _set_camera_property(self, property_id, value): f"{actual_value}, not the requested {value}" ) else: - # set() returned False: the driver rejected the property - # (common on analog grabbers / cameras that don't expose it). + # set() returned False: the camera doesn't expose this property. logging.warning( f"Camera property {property_id} is not supported by this " f"camera; requested value {value} was ignored" @@ -655,7 +626,10 @@ def rec(self): continue # Process the frame here except RuntimeError as error: - log.error(f"Failed to read frame from camera. Error: {error}") + now = time.time() + if now - self._last_frame_err_log >= 1.0: + self._last_frame_err_log = now + log.error(f"Failed to read frame from camera. Error: {error}") continue tmst = self.logger_timer.elapsed_time() if not self.res_set: diff --git a/src/ethopy/interfaces/dlc.py b/src/ethopy/interfaces/dlc.py index 544c330..14ba193 100644 --- a/src/ethopy/interfaces/dlc.py +++ b/src/ethopy/interfaces/dlc.py @@ -1,3 +1,4 @@ +import logging import multiprocessing as mp import os import time @@ -20,6 +21,8 @@ IMPORT_DLCLive = False from ethopy.utils.helper_functions import read_yalm, shared_memory_array +log = logging.getLogger(__name__) + np.set_printoptions(suppress=True) @@ -52,12 +55,15 @@ def __init__(self, path: str): self.joint_names = read_yalm(self.path, "pose_cfg.yaml", "all_joints_names") def setup_model(self, frame): + log.debug( + "DLC setup input: shape=%s, dtype=%s, min=%s, max=%s", + frame.shape, frame.dtype, frame.min(), frame.max(), + ) self.dlc_model = DLCLive(self.path, processor=self.dlc_processor) - self.dlc_model.init_inference(frame / 255) + self.dlc_model.init_inference((frame / 255).astype(np.float32)) def get_pose(self, frame): - return self.dlc_model.get_pose(frame / 255) - + return self.dlc_model.get_pose((frame / 255).astype(np.float32)) class DLCProcessor(ABC): """ @@ -82,7 +88,7 @@ def __init__( "Please install dlc_live before using DLCProcessor.\n" "sudo pip3 install deeplabcut-live" ) - print("model_path ", model_path) + log.debug("DLC model_path: %s", model_path) self.model = DLCModel(model_path) self.frame_queue = frame_queue self.frame_timeout = 1 @@ -94,12 +100,20 @@ def __init__( self.finish_signal.clear() self.current_frame = None + self._log_throttle = {} # per-key timestamps for the per-frame logs below self.dlc_process = mp.Process(target=self._setup_and_run) self.dlc_process.start() if wait_for_setup: self._wait_for_setup() + def _throttled_log(self, key, level, msg, *args, interval=1.0): + """Log at most once per ``interval`` seconds per ``key`` (for loop bodies).""" + now = time.time() + if now - self._log_throttle.get(key, 0.0) >= interval: + self._log_throttle[key] = now + log.log(level, msg, *args) + def _wait_for_setup(self): """Wait for the DLC model setup to complete.""" self.setup_complete.wait(timeout=30) @@ -133,13 +147,17 @@ def process_frames(self): if self.latest_frame is not None: frame_tranfer_delay = self.logger.logger_timer.elapsed_time()-latest_timestamp if frame_tranfer_delay > 100: - print(f"###############################frame transfer delay: {frame_tranfer_delay} ms") - # print('exception qsize', self.frame_queue.qsize(), self.frame_queue.empty()) + self._throttled_log( + "transfer_delay", logging.WARNING, + "DLC frame transfer delay: %s ms", frame_tranfer_delay, + ) if delay_time > 0.01: - print(f"------------------------------------------ DLC queue empty delay: {delay_time} sec") + self._throttled_log( + "drain_delay", logging.DEBUG, + "DLC queue drain delay: %s sec", delay_time, + ) pose = self.model.get_pose(self.latest_frame) self._process_frame(pose, latest_timestamp) - # print("time ", time.time()-start_t) else: # If stop signal is set wait until there is no new frames(Close camera) if self.stop_signal.is_set(): @@ -147,10 +165,10 @@ def process_frames(self): time.sleep(0.01) # Short sleep to prevent busy-waiting except Exception as e: # Log any exceptions that occur during frame processing - print(f"Frame processing error: {e}") + log.exception("DLC frame processing error: %s", e) finally: # Ensure cleanup is always executed, even if an error occurs - print("Frame process has been finished.") + log.debug("DLC frame process finished.") self._process_finish() self.finish_signal.clear() @@ -169,7 +187,7 @@ def stop(self): self.stop_signal.set() self.dlc_process.join(timeout=60) if self.dlc_process.is_alive(): - print("Terminate dlc process") + log.warning("DLC process did not stop in time; terminating.") self.dlc_process.terminate() # Force terminate if not stopping. @@ -209,15 +227,30 @@ def __init__( def _process_frame(self, pose, timestamp): """Detect arena corners and calculate perspective transform.""" - if np.all(pose[:, 2] > self.CONFIDENCE_THRESHOLD): + confident = np.all(pose[:, 2] > self.CONFIDENCE_THRESHOLD) + self._throttled_log( + "corner_scores", logging.DEBUG, + "DLC corner scores=%s all>%s? %s", pose[:, 2], self.CONFIDENCE_THRESHOLD, confident, + ) + if confident: self.detected_corners.append(pose) - else: - print("\rWait for high confidence corners scores", pose[:, 2], end="") + log.debug("DLC corner frame appended — total %s", len(self.detected_corners)) if len(self.detected_corners) >= self.MIN_CONFIDENT_FRAMES or self.stop_signal.is_set(): self.finish_signal.set() def _process_finish(self): + log.debug( + "DLC corner finish: %s confident frame(s), stop_signal=%s", + len(self.detected_corners), self.stop_signal.is_set(), + ) + if len(self.detected_corners) == 0: + log.warning( + "DLC corner detection found no high-confidence corners; " + "skipping perspective transform." + ) + return self.corners = np.mean(np.array(self.detected_corners), axis=0) + log.debug("DLC detected corners: %s", self.corners) self.affine_matrix, self.affine_matrix_inv = self._calculate_perspective_transform( self.corners, self.arena_size ) @@ -385,9 +418,11 @@ def _initialize_pose(self, confidence_threshold: float = 0.01) -> np.ndarray: if not self.frame_queue.empty(): _, frame = self.frame_queue.get_nowait() pose = self.model.get_pose(frame) - print("frame ", frame) scores = np.array(pose[0:3][:, 2]) - print("\rWait for high confidence pose scores ", scores, end="") + self._throttled_log( + "pose_wait", logging.DEBUG, + "Waiting for high-confidence pose scores: %s", scores, + ) if np.sum(scores >= confidence_threshold) == 3: return pose time.sleep(0.1) @@ -544,4 +579,4 @@ def stop(self): try: self.shared_memory.unlink() except FileNotFoundError: - print("Shared memory already unlinked or does not exist.") + log.debug("Shared memory already unlinked or does not exist.") From 422ebb3523d88d6bc00fb28ae5c069b84782c5e3 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 25 Jun 2026 11:23:50 +0300 Subject: [PATCH 07/20] Use explicit length check for recs Replace the truthiness check 'if not recs' with 'if len(recs) == 0' when computing rec_idx in Logger. This avoids ambiguous truth-value errors for array-like objects (e.g., numpy arrays) and ensures rec_idx is set to 1 for empty collections, otherwise max(recs) + 1. --- src/ethopy/core/logger.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/ethopy/core/logger.py b/src/ethopy/core/logger.py index 4249c99..47efded 100755 --- a/src/ethopy/core/logger.py +++ b/src/ethopy/core/logger.py @@ -1014,7 +1014,7 @@ def log_recording(self, rec_key: Dict, **kwargs) -> None: key=self.trial_key, fields=["rec_idx"], ) - rec_idx = 1 if not recs else max(recs) + 1 + rec_idx = 1 if len(recs) == 0 else max(recs) + 1 self.log("Recording", data={**rec_key, "rec_idx": rec_idx}, schema="recording", **kwargs) From 2e7399b9fedc89eb9506ef6916d250d587612d74 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 25 Jun 2026 11:56:48 +0300 Subject: [PATCH 08/20] co-locate camera video with session h5 files; write the .mp4 into the same Recordings/{animal}_{session}/ folder as the timestamp/DLC h5 (drops the now-unused video_source_path keys), and make clear_local_videos move only this camera's own files without rmdir, so it can't clobber the Writer-owned h5 files or other cameras' videos. --- src/ethopy/interfaces/Camera.py | 77 +++++++++++++++------------------ 1 file changed, 36 insertions(+), 41 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index a8c2450..1fd68b1 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -82,8 +82,22 @@ def __init__( else datetime.now().strftime("%Y-%m-%d_%H-%M-%S") ) - self.source_path = local_conf.get("video_source_path", "") + f"{self.filename}/" - self.target_path = local_conf.get("video_target_path", "") + f"{self.filename}/" + if logger: + # Co-locate the video with the timestamp/DLC h5 in the session + # Recordings folder (the path Logger.createDataset also uses). + recordings_folder = ( + f"Recordings/{logger.trial_key['animal_id']}" + f"_{logger.trial_key['session']}/" + ) + self.source_path = logger.source_path + recordings_folder + self.target_path = ( + logger.target_path + recordings_folder + if os.path.isdir(logger.target_path) + else self.source_path + ) + else: + self.source_path = local_conf.get("source_path", "") + f"{self.filename}/" + self.target_path = local_conf.get("target_path", "") + f"{self.filename}/" self.serve_port = local_conf.get("server.port", 0) if self.serve_port: @@ -123,14 +137,8 @@ def __init__( block=True, ) # Per-camera name so two cameras in a session don't overwrite each - # other's h5; paths mirror Logger.createDataset (logger.py:950-961), - # which writes the file later in the child process. + # other's h5; the file is written later by Logger.createDataset. self.filename_tmst = f"videotmst_{self.filename}.h5" - recordings_folder = ( - f"Recordings/{logger.trial_key['animal_id']}" - f"_{logger.trial_key['session']}/" - ) - h5_source_path = logger.source_path + recordings_folder h5_target_path = ( logger.target_path + recordings_folder if os.path.isdir(logger.target_path) @@ -142,7 +150,7 @@ def __init__( software="EthoPy", version="0.1", filename=self.filename_tmst, - source_path=h5_source_path, + source_path=self.source_path, target_path=h5_target_path, ), block=True, @@ -240,45 +248,32 @@ def copy_file(args): log.error(f"Failed to transfer file: {file.name}. Reason: {ex}") def clear_local_videos(self) -> None: - """ - Move all files from the source path to the target path. + """Move this camera's video file(s) to the target path. + + The source folder is shared with the timestamp/DLC h5 files (owned by the + Writer) and other cameras, so only this camera's own files are moved + (matched by filename, excluding .h5) and the folder is left in place. """ source = Path(self.source_path) target = Path(self.target_path) - if not source.is_dir(): - log.error(f"Video source path does not exist: {source}") - raise ValueError( - f"Source path {source} does not exist or is not a directory." - ) - - if not target.exists(): - log.error(f"Video target path does not exist: {target}") - log.error(f"Create it with: mkdir -p {target}") - raise ValueError( - f"Target path {target} does not exist or is not a directory." - ) - - files = [(entry, target) for entry in source.iterdir() if entry.is_file()] + if source == target or not target.is_dir(): + return # autocopy disabled; leave the video alongside the h5 files + files = [ + (entry, target) + for entry in source.iterdir() + if entry.is_file() + and self.filename in entry.name + and entry.suffix.lower() != ".h5" + ] if not files: log.warning("No video files found to transfer") - else: - log.info( - f"Transferring {len(files)} video file(s) from {source} to {target}" - ) - with Pool(processes=min(2, os.cpu_count() - 1)) as pool: - pool.map(self.copy_file, files) + return - # Clean up if the source directory is empty - if not any(source.iterdir()): - source.rmdir() - log.info("Video transfer complete, cleaned up source folder") - else: - remaining = list(source.iterdir()) - log.warning( - f"Source folder not empty after transfer: {len(remaining)} items remaining" - ) + log.info(f"Transferring {len(files)} video file(s) from {source} to {target}") + with Pool(processes=min(2, os.cpu_count() - 1)) as pool: + pool.map(self.copy_file, files) def setup(self) -> None: """ From fbb8b13663dd207cce92975676930dd460febb9b Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 25 Jun 2026 12:03:01 +0300 Subject: [PATCH 09/20] expand conditions on any iterable primary key, not just core[0] get_table_keys can return a scalar primary key first (e.g. response_port before response_loc_x), so the old core[0] check missed the iterable and inserted tuples into scalar columns. Scan all non-hash primary keys for a list/tuple instead. It used hasattr(..., "__iter__"), which treats strings as expandable, and the inner split used a negative scalar test (int/float/str) that misclassified numpy scalars as sequences and crashed on indexing. --- src/ethopy/core/experiment.py | 29 +++++++++++++++++++---------- 1 file changed, 19 insertions(+), 10 deletions(-) diff --git a/src/ethopy/core/experiment.py b/src/ethopy/core/experiment.py index 2918943..8f11861 100644 --- a/src/ethopy/core/experiment.py +++ b/src/ethopy/core/experiment.py @@ -591,19 +591,28 @@ def log_conditions( log.warning(f"Skipping {ctable}, Missing keys:{missing_keys}") continue - # check if there is a primary key which is not hash and it is iterable - if core and hasattr(condition[core[0]], "__iter__"): - # TODO make a function for this and clarify it - # If any of the primary keys is iterable all the rest should be. - # The first element of the iterable will be matched with the first - # element of the rest of the keys - for idx, _ in enumerate(condition[core[0]]): + # If any non-hash primary key is an indexable sequence (list, + # tuple, or numpy array), expand the condition into one row per + # element. Checking every core key (not just core[0]) avoids + # missing the sequence when a scalar key sorts first. Strings and + # numpy scalars are treated as single values, not expanded. + seq_types = (list, tuple, np.ndarray) + expandable_key = next( + (k for k in core if isinstance(condition.get(k), seq_types)), + None, + ) + if expandable_key is not None: + # Expand into one row per element, all sharing the same + # cond_hash (one condition, many rows). Scalar fields are + # repeated; every sequence field is split in parallel by + # index, so all sequences must have the same length. + for idx in range(len(condition[expandable_key])): cond_key = {} for k in fields: - if isinstance(condition[k], (int, float, str)): - cond_key[k] = condition[k] - else: + if isinstance(condition[k], seq_types): cond_key[k] = condition[k][idx] + else: + cond_key[k] = condition[k] self.logger.put( table=ctable, From a9d008ca8f125d37c06e59add723a8c8c1db70b8 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 25 Jun 2026 14:10:18 +0300 Subject: [PATCH 10/20] Update dlc.py --- src/ethopy/interfaces/dlc.py | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/src/ethopy/interfaces/dlc.py b/src/ethopy/interfaces/dlc.py index 14ba193..33a7698 100644 --- a/src/ethopy/interfaces/dlc.py +++ b/src/ethopy/interfaces/dlc.py @@ -63,7 +63,8 @@ def setup_model(self, frame): self.dlc_model.init_inference((frame / 255).astype(np.float32)) def get_pose(self, frame): - return self.dlc_model.get_pose((frame / 255).astype(np.float32)) + # DLCLive's exported graph does its own preprocessing and expects pixels in the 0-255 range. + return self.dlc_model.get_pose(frame) class DLCProcessor(ABC): """ From df9181491c521fa943811fd805ef9d069e42e9fb Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Mon, 20 Jul 2026 13:41:30 +0300 Subject: [PATCH 11/20] fix: pass configured serve_port and support logger-less WebCam timestamps The MJPEG stream ignored server.port: HTTPServerThread was constructed without serve_port, so it always bound 8000 even when a different port was configured. Pass self.serve_port through. WebCam wrote timestamps only via self.dataset, which setup() leaves None on the txt path (logger=None), so a logger-less WebCam crashed with AttributeError in write_frame and at the end of rec(). Guard both o tmst_type, mirroring PiCamera. --- src/ethopy/interfaces/Camera.py | 17 +++++++++++++---- 1 file changed, 13 insertions(+), 4 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 1fd68b1..7006315 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -542,8 +542,11 @@ def write_frame(self, item: Tuple[float, np.ndarray]) -> None: """ img = item[1].copy() self.video_output.writeFrame(img) - # Append the timestamp to the 'frame_tmst' h5 dataset - self.dataset.append("frame_tmst", [np.double(item[0])]) + # Record the timestamp: h5 dataset with a logger, plain text file without. + if self.tmst_type == "txt": + self.tmst_output.write(f"{item[0]}\n") + else: + self.dataset.append("frame_tmst", [np.double(item[0])]) def camera_opened(self, camera): """Check if the camera is opened.""" @@ -640,7 +643,10 @@ def rec(self): self.camera.release() self.recording.clear() - self.dataset.exit() + if self.tmst_type == "txt": + self.tmst_output.close() + else: + self.dataset.exit() def stop_rec(self): """ @@ -792,7 +798,10 @@ def init_cam(self) -> "Picamera2": output = FfmpegOutput(str(Path(self.source_path) / f"{self.filename}.mp4")) if self.serve_port > 0: self.httpthread = HTTPServerThread( - self, server_user=self.server_user, server_password=self.server_password + self, + serve_port=self.serve_port, + server_user=self.server_user, + server_password=self.server_password, ) self.httpthread.start() picam2.start_encoder(encoder, output) From af59f23b2e430a52fe1ef7814d0046837bbad6dd Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Mon, 20 Jul 2026 14:11:41 +0300 Subject: [PATCH 12/20] refactor: extract condition-row expansion into a documented helper The inline expansion in log_conditions mixed three concerns (detect a sequence primary key, validate lengths, split into rows) in a nested loop that was hard to follow. Move it to task_helper_funcs.expand_condition_rows with a docstring explaining the semantics, so the call site is just "build the rows, insert each." The helper also validates that all sequence fields share one length and raises a clear error instead of the previous raw IndexError / silent truncation on mismatched lengths. Co-Authored-By: Claude Opus 4.8 --- src/ethopy/core/experiment.py | 42 +++++---------------- src/ethopy/utils/task_helper_funcs.py | 53 +++++++++++++++++++++++++++ 2 files changed, 62 insertions(+), 33 deletions(-) diff --git a/src/ethopy/core/experiment.py b/src/ethopy/core/experiment.py index 8f11861..1196f2e 100644 --- a/src/ethopy/core/experiment.py +++ b/src/ethopy/core/experiment.py @@ -28,7 +28,11 @@ from ethopy.core.logger import Logger, experiment from ethopy.utils.helper_functions import factorize, make_hash -from ethopy.utils.task_helper_funcs import format_params_print, get_parameters +from ethopy.utils.task_helper_funcs import ( + expand_condition_rows, + format_params_print, + get_parameters, +) from ethopy.utils.timer import Timer log = logging.getLogger(__name__) @@ -591,39 +595,11 @@ def log_conditions( log.warning(f"Skipping {ctable}, Missing keys:{missing_keys}") continue - # If any non-hash primary key is an indexable sequence (list, - # tuple, or numpy array), expand the condition into one row per - # element. Checking every core key (not just core[0]) avoids - # missing the sequence when a scalar key sorts first. Strings and - # numpy scalars are treated as single values, not expanded. - seq_types = (list, tuple, np.ndarray) - expandable_key = next( - (k for k in core if isinstance(condition.get(k), seq_types)), - None, - ) - if expandable_key is not None: - # Expand into one row per element, all sharing the same - # cond_hash (one condition, many rows). Scalar fields are - # repeated; every sequence field is split in parallel by - # index, so all sequences must have the same length. - for idx in range(len(condition[expandable_key])): - cond_key = {} - for k in fields: - if isinstance(condition[k], seq_types): - cond_key[k] = condition[k][idx] - else: - cond_key[k] = condition[k] - - self.logger.put( - table=ctable, - tuple=cond_key, - schema=schema, - priority=_priority, - ) - - else: + # A condition normally maps to one row, but a sequence-valued + # primary key expands it into several (see expand_condition_rows). + for row in expand_condition_rows(condition, fields, core): self.logger.put( - table=ctable, tuple=condition, schema=schema, priority=_priority + table=ctable, tuple=row, schema=schema, priority=_priority ) # Increment the priority for each subsequent table diff --git a/src/ethopy/utils/task_helper_funcs.py b/src/ethopy/utils/task_helper_funcs.py index 70b51ca..9e269a8 100644 --- a/src/ethopy/utils/task_helper_funcs.py +++ b/src/ethopy/utils/task_helper_funcs.py @@ -1,5 +1,58 @@ +from typing import Any, Dict, List + import numpy as np +# Field values that split a single condition into several table rows. +# Strings and numpy scalars are deliberately excluded: they are single values. +_SEQUENCE_TYPES = (list, tuple, np.ndarray) + + +def expand_condition_rows( + condition: Dict[str, Any], fields: set, core: List[str] +) -> List[Dict[str, Any]]: + """Turn one condition into the list of table rows it describes. + + A condition usually maps to a single row. When a primary key holds a + sequence (list/tuple/array) it instead describes several rows at once, for + example one row per response port, all sharing the same ``cond_hash``. + Every sequence field is then split in parallel by index, and every scalar + field is repeated in each row. + + Expansion is triggered only by a sequence in a *primary* key (``core``): the + rows must differ in their primary key to be distinct, so a sequence in a + non-primary field alone is left untouched (it would create duplicate keys). + + Args: + condition: The condition; already holds every name in ``fields``. + fields: All column names of the target table. + core: The non-hash primary key names of the target table. + + Returns: + One dict per row, a single-element list when there is nothing to expand. + + Raises: + ValueError: if the sequence fields do not all share the same length. + """ + def is_sequence(value: Any) -> bool: + return isinstance(value, _SEQUENCE_TYPES) + + if not any(is_sequence(condition[k]) for k in core): + return [condition] + + lengths = {k: len(condition[k]) for k in fields if is_sequence(condition[k])} + if len(set(lengths.values())) > 1: + raise ValueError( + f"Condition has sequence fields of unequal length: {lengths}. " + "All sequence-valued fields in one condition must share one length." + ) + + n_rows = next(iter(lengths.values())) + return [ + {k: condition[k][idx] if is_sequence(condition[k]) else condition[k] + for k in fields} + for idx in range(n_rows) + ] + def get_parameters(_class): """Create a dictionary with required fields set to '...' and default values included. From e6b0c529abacefbfff4ad3b8dfdecef7fc94dcc1 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Mon, 20 Jul 2026 15:00:29 +0300 Subject: [PATCH 13/20] test: cover expand_condition_rows Unit tests for the condition-row expansion helper: scalar passthrough, single- and parallel-sequence expansion, trigger on any primary key, non-primary sequences left whole, strings/numpy scalars treated as single values, and the unequal-length ValueError. Co-Authored-By: Claude Opus 4.8 --- tests/test_task_helper_funcs.py | 98 +++++++++++++++++++++++++++++++++ 1 file changed, 98 insertions(+) create mode 100644 tests/test_task_helper_funcs.py diff --git a/tests/test_task_helper_funcs.py b/tests/test_task_helper_funcs.py new file mode 100644 index 0000000..976feb7 --- /dev/null +++ b/tests/test_task_helper_funcs.py @@ -0,0 +1,98 @@ +"""Tests for helpers in ethopy.utils.task_helper_funcs. + +expand_condition_rows is a pure function (no database), so these tests import +and call it directly. +""" + +import numpy as np +import pytest + +from ethopy.utils.task_helper_funcs import expand_condition_rows + + +class TestExpandConditionRows: + """Turn one condition into the table rows it describes.""" + + def test_scalar_only_returns_condition_unchanged(self): + """No sequence anywhere -> the single condition, untouched.""" + condition = {"cond_hash": "h", "difficulty": 3, "reward": 5} + rows = expand_condition_rows( + condition, {"cond_hash", "difficulty", "reward"}, ["difficulty"] + ) + assert rows == [condition] + + def test_single_sequence_primary_key_expands_and_repeats_scalars(self): + """A list primary key -> one row per element; scalar fields repeat.""" + condition = {"cond_hash": "h", "port": [1, 2, 3], "reward": 5} + rows = expand_condition_rows( + condition, {"cond_hash", "port", "reward"}, ["port"] + ) + assert rows == [ + {"cond_hash": "h", "port": 1, "reward": 5}, + {"cond_hash": "h", "port": 2, "reward": 5}, + {"cond_hash": "h", "port": 3, "reward": 5}, + ] + + def test_parallel_sequences_split_by_index(self): + """Two equal-length sequence keys split together, element by element.""" + condition = {"cond_hash": "h", "port": [1, 2], "loc_x": [0.1, 0.2]} + rows = expand_condition_rows( + condition, {"cond_hash", "port", "loc_x"}, ["port", "loc_x"] + ) + assert rows == [ + {"cond_hash": "h", "port": 1, "loc_x": 0.1}, + {"cond_hash": "h", "port": 2, "loc_x": 0.2}, + ] + + def test_scalar_primary_key_first_still_finds_the_sequence(self): + """Expansion triggers on any primary key, not just the first one.""" + condition = {"cond_hash": "h", "resp_port": 7, "loc_x": [0.1, 0.2, 0.3]} + rows = expand_condition_rows( + condition, {"cond_hash", "resp_port", "loc_x"}, ["resp_port", "loc_x"] + ) + assert [r["loc_x"] for r in rows] == [0.1, 0.2, 0.3] + assert all(r["resp_port"] == 7 for r in rows) + + def test_sequence_in_non_primary_field_is_not_expanded(self): + """A sequence outside the primary key stays whole (no duplicate keys).""" + condition = {"cond_hash": "h", "label": "a", "blob": [1, 2, 3]} + rows = expand_condition_rows( + condition, {"cond_hash", "label", "blob"}, ["label"] + ) + assert rows == [condition] + + def test_string_primary_key_is_a_single_value(self): + """Strings are not sequences here -> not expanded per character.""" + condition = {"cond_hash": "h", "stim_type": "grating", "reward": 5} + rows = expand_condition_rows( + condition, {"cond_hash", "stim_type", "reward"}, ["stim_type"] + ) + assert rows == [condition] + + def test_numpy_array_primary_key_expands(self): + """A numpy array primary key expands like a list.""" + condition = {"cond_hash": "h", "port": np.array([1, 2, 3])} + rows = expand_condition_rows(condition, {"cond_hash", "port"}, ["port"]) + assert [r["port"] for r in rows] == [1, 2, 3] + + def test_numpy_scalar_field_is_repeated_not_indexed(self): + """A numpy scalar is a single value, repeated across the expanded rows.""" + condition = {"cond_hash": "h", "port": [1, 2], "seed": np.int64(42)} + rows = expand_condition_rows( + condition, {"cond_hash", "port", "seed"}, ["port"] + ) + assert [r["seed"] for r in rows] == [np.int64(42), np.int64(42)] + + def test_tuple_primary_key_expands(self): + """Tuples count as sequences too.""" + condition = {"cond_hash": "h", "port": (1, 2)} + rows = expand_condition_rows(condition, {"cond_hash", "port"}, ["port"]) + assert [r["port"] for r in rows] == [1, 2] + + def test_mismatched_sequence_lengths_raise(self): + """Unequal sequence lengths raise a clear error naming the fields.""" + condition = {"cond_hash": "h", "port": [1, 2, 3], "loc_x": [0.1, 0.2]} + with pytest.raises(ValueError, match="unequal length"): + expand_condition_rows( + condition, {"cond_hash", "port", "loc_x"}, ["port", "loc_x"] + ) From 0ec33b52e5458f4a9d03869185c001b471b2388f Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Mon, 20 Jul 2026 15:24:17 +0300 Subject: [PATCH 14/20] test: initialize session_params in the behavior fixture test_is_hydrated assigns into behavior.session_params, but the fixture built a Behavior() without calling setup(), leaving session_params as None (its __init__ default). Initialize it to an empty dict in the fixture, matching how the other mocked attributes are set up. --- tests/test_behavior.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_behavior.py b/tests/test_behavior.py index a3fb667..a7d240a 100644 --- a/tests/test_behavior.py +++ b/tests/test_behavior.py @@ -38,6 +38,7 @@ def behavior(self): beh.logger.trial_key = {} # Empty dict for trial key beh.interface = Mock() beh.params = {} + beh.session_params = {} # normally populated by setup(); tests assign into it # Set logging explicitly since it may not be set correctly due to mocking beh.logging = True return beh @@ -113,14 +114,14 @@ def test_is_hydrated(self, behavior): assert behavior.is_hydrated(rew=6.0) is False # Test with params max_reward - behavior.params["max_reward"] = 4.0 + behavior.session_params["max_reward"] = 4.0 assert behavior.is_hydrated() is True - behavior.params["max_reward"] = 6.0 + behavior.session_params["max_reward"] = 6.0 assert behavior.is_hydrated() is False # Test with no max_reward set - behavior.params["max_reward"] = None + behavior.session_params["max_reward"] = None assert behavior.is_hydrated() is False def test_is_sleep_time(self, behavior): From d2d6e02af1a197a8fe1c100bb7d2da5afbd0aa0f Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Mon, 20 Jul 2026 16:23:28 +0300 Subject: [PATCH 15/20] test: pin empty-sequence and tuple-split behavior of expand_condition_rows Add cases for an empty sequence (zero rows) and a tuple secondary field (split element-by-element). Both lock in the existing behavior, which a differential check confirmed is identical to the pre-refactor inline logic; the tuple case also documents that it intentionally differs from factorize. --- tests/test_task_helper_funcs.py | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/tests/test_task_helper_funcs.py b/tests/test_task_helper_funcs.py index 976feb7..453e7c5 100644 --- a/tests/test_task_helper_funcs.py +++ b/tests/test_task_helper_funcs.py @@ -89,6 +89,24 @@ def test_tuple_primary_key_expands(self): rows = expand_condition_rows(condition, {"cond_hash", "port"}, ["port"]) assert [r["port"] for r in rows] == [1, 2] + def test_tuple_secondary_field_is_split_like_a_list(self): + """A tuple is split element-by-element, even as a secondary field. + + Note: this differs from factorize(), which keeps tuples as one composite + value. + """ + condition = {"cond_hash": "h", "port": [1, 2], "coord": (5, 6)} + rows = expand_condition_rows( + condition, {"cond_hash", "port", "coord"}, ["port"] + ) + assert [r["coord"] for r in rows] == [5, 6] + + def test_empty_sequence_expands_to_no_rows(self): + """An empty sequence primary key produces zero rows (silently).""" + condition = {"cond_hash": "h", "port": []} + rows = expand_condition_rows(condition, {"cond_hash", "port"}, ["port"]) + assert rows == [] + def test_mismatched_sequence_lengths_raise(self): """Unequal sequence lengths raise a clear error naming the fields.""" condition = {"cond_hash": "h", "port": [1, 2, 3], "loc_x": [0.1, 0.2]} From 8dc817f08840d00c5a8b11325dbe7b4039918c20 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 13 Aug 2026 15:33:42 +0300 Subject: [PATCH 16/20] Ensure source/target paths end with separator Wrap source_path and target_path with os.path.join(..., "") to ensure they end with a path separator. This is necessary because both paths are concatenated with subfolders/filenames in multiple places (Logger, Writer, and Camera classes), so they must have a trailing separator. --- src/ethopy/core/logger.py | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/ethopy/core/logger.py b/src/ethopy/core/logger.py index 47efded..3053feb 100755 --- a/src/ethopy/core/logger.py +++ b/src/ethopy/core/logger.py @@ -153,9 +153,12 @@ def __init__(self, task: bool = False) -> None: self.update_status.clear() # source path is the local path that data are saved - self.source_path = local_conf.get("source_path") # target path is the path that data will be moved after the session ends - self.target_path = local_conf.get("target_path") + # Both are joined to subfolders/filenames by string concatenation (here, in + # Writer and in Camera), so they must end with a separator; os.path.join + # with "" appends one only when it is missing. + self.source_path = os.path.join(local_conf.get("source_path"), "") + self.target_path = os.path.join(local_conf.get("target_path"), "") # inserter_thread read the queue and insert the data in the database self.thread_end, self.thread_lock = threading.Event(), threading.Lock() From ce245c886b7833837dc839fcf960814358d09d2e Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Thu, 13 Aug 2026 15:36:33 +0300 Subject: [PATCH 17/20] Harden file transfer and camera error handling Make file transfer resilient: copy_file now returns a bool, verifies the destination file size before deleting the local copy, and handles OSError clear_local_videos collects worker results and logs any failed transfers while keeping local copies. Replace string-based raises with RuntimeError (preserving traceback) in setup and recording loops. PiCamera improvements: explicit cv2 import check with a helpful message, guard against cam being None before stopping/closing, and ensure self.cam is nulled during cleanup. These changes reduce data-loss risk and clarify errors. --- src/ethopy/interfaces/Camera.py | 72 +++++++++++++++++++++++---------- 1 file changed, 51 insertions(+), 21 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 7006315..1340188 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -218,7 +218,7 @@ def _create_and_set_path(self, path: str) -> str: return path @staticmethod - def copy_file(args): + def copy_file(args) -> bool: """ Copy a file from the source path to the target path. @@ -226,26 +226,32 @@ def copy_file(args): args (tuple): A tuple containing the source file path and the target directory path. Returns: - None - - Raises: - FileNotFoundError: If the source file is not found. + bool: True if the file was copied, verified and removed locally. On + False the local copy is kept, so the recording is never lost. """ file, target = args + destination = target / file.name try: - shutil.copy(str(file), str(target / file.name)) + shutil.copy(str(file), str(destination)) log.debug(f"Transferred file: {file.name}") - # Verify the file exists in the target directory - if os.path.exists(str(target / file.name)) and os.path.getsize( - str(file) - ) == os.path.getsize(str(target / file.name)): - os.remove(str(file)) - log.debug(f"Deleted original file: {file.name}") - else: - log.error(f"Failed to transfer file: {file.name}") - except FileNotFoundError as ex: + # Verify the copy before deleting the only other copy of the data + if ( + not destination.exists() + or destination.stat().st_size != file.stat().st_size + ): + log.error( + f"Size mismatch after transferring {file.name}; " + "keeping the local copy" + ) + return False + os.remove(str(file)) + log.debug(f"Deleted original file: {file.name}") + return True + except OSError as ex: + # OSError also covers shutil.SameFileError and a dropped network mount log.error(f"Failed to transfer file: {file.name}. Reason: {ex}") + return False def clear_local_videos(self) -> None: """Move this camera's video file(s) to the target path. @@ -273,7 +279,14 @@ def clear_local_videos(self) -> None: log.info(f"Transferring {len(files)} video file(s) from {source} to {target}") with Pool(processes=min(2, os.cpu_count() - 1)) as pool: - pool.map(self.copy_file, files) + results = pool.map(self.copy_file, files) + + failed = [entry.name for (entry, _), ok in zip(files, results) if not ok] + if failed: + log.error( + f"Failed to transfer {len(failed)} of {len(files)} video file(s): " + f"{', '.join(failed)}. They are kept in {source}" + ) def setup(self) -> None: """ @@ -297,7 +310,9 @@ def start_rec(self) -> None: self.capture_runner.join() self.write_runner.join() except Exception as cam_error: - raise f"Exception occurred during recording: {cam_error}" + raise RuntimeError( + f"Exception occurred during recording: {cam_error}" + ) from cam_error def dequeue(self, frame_queue: Queue) -> None: """ @@ -687,6 +702,16 @@ def __init__( raise ImportError( "the picamera package could not be imported, install it before use!" ) + # PicameraOutput annotates every frame with cv2, so a missing cv2 would + # otherwise surface as a NameError inside the recording thread. + if not globals()["IMPORT_CV2"]: + raise ImportError( + "The cv2 package could not be imported. " + "Please install it before using PiCamera.\n" + "On Raspberry Pi OS install it from apt so it links against the " + "system numpy:\n" + "sudo apt install python3-opencv" + ) self.initialized = threading.Event() self.initialized.clear() self.cam = None @@ -754,7 +779,9 @@ def rec(self) -> None: while not self.stop.is_set(): time.sleep(1) except Exception as rec_error: - raise f"Error during camera recording: {rec_error}" + raise RuntimeError( + f"Error during camera recording: {rec_error}" + ) from rec_error finally: self._stop_recording() @@ -813,8 +840,11 @@ def _stop_recording(self) -> None: if self.recording.is_set(): if self.httpthread: self.httpthread.stop_serving() - self.cam.stop_recording() - self.cam.close() + # cam is None when init_cam() raised; without this the AttributeError + # here would replace the real initialisation error. + if self.cam is not None: + self.cam.stop_recording() + self.cam.close() if self.tmst_type == "txt": self.tmst_output.close() @@ -822,7 +852,7 @@ def _stop_recording(self) -> None: self.dataset.exit() self.recording.clear() - self._cam = None + self.cam = None self.clear_local_videos() def write_frame(self, item: Union[List, tuple]) -> None: From f627124938d66d23aad912a9efd9649e0f291c69 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Wed, 19 Aug 2026 14:20:53 +0300 Subject: [PATCH 18/20] Start the camera HTTP server after the camera is initialised --- src/ethopy/interfaces/Camera.py | 26 ++++++++++++++++++-------- 1 file changed, 18 insertions(+), 8 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 1340188..1befef8 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -790,6 +790,24 @@ def recording_init(self) -> None: self.stop.clear() self.recording.set() self.cam = self.init_cam() + self._start_http_server() + + def _start_http_server(self) -> None: + """Serve the camera over HTTP, if a port is configured. + + Must run after self.cam is assigned: start_serving() dereferences it + from the HTTP handler thread, so the server cannot accept a client any + earlier without racing the camera being ready. + """ + if self.serve_port <= 0: + return + self.httpthread = HTTPServerThread( + self, + serve_port=self.serve_port, + server_user=self.server_user, + server_password=self.server_password, + ) + self.httpthread.start() def init_cam(self) -> "Picamera2": """Initialize the camera.""" @@ -823,14 +841,6 @@ def init_cam(self) -> "Picamera2": ) # pylint: disable=all encoder = H264Encoder(10000000) output = FfmpegOutput(str(Path(self.source_path) / f"{self.filename}.mp4")) - if self.serve_port > 0: - self.httpthread = HTTPServerThread( - self, - serve_port=self.serve_port, - server_user=self.server_user, - server_password=self.server_password, - ) - self.httpthread.start() picam2.start_encoder(encoder, output) return picam2 From 65e864adaf102776ec73c61d7439da594775d653 Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Wed, 19 Aug 2026 14:33:26 +0300 Subject: [PATCH 19/20] Throttle the MJPEG stream and stop holding the frame lock during writes --- src/ethopy/interfaces/Camera.py | 23 +++++++++++++++++++---- 1 file changed, 19 insertions(+), 4 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 1befef8..baa2607 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -103,6 +103,8 @@ def __init__( if self.serve_port: self.server_user = local_conf.get("server.user", "") self.server_password = local_conf.get("server.password", "") + # Frames per second to stream; 0 streams every encoded frame. + self.serve_fps = local_conf.get("server.fps", 0) self.httpthread = None self.tmst_type = None self.dataset = None @@ -806,6 +808,7 @@ def _start_http_server(self) -> None: serve_port=self.serve_port, server_user=self.server_user, server_password=self.server_password, + serve_fps=self.serve_fps, ) self.httpthread.start() @@ -946,6 +949,7 @@ def __init__( serve_port: int = 8000, server_user: Optional[str] = None, server_password: Optional[str] = None, + serve_fps: float = 0, ): super().__init__() self.python_logger = logging.getLogger(self.__class__.__name__) @@ -953,6 +957,8 @@ def __init__( ("", serve_port), self.CameraHTTPRequestHandler ) self.server.cam = cam + # 0 (the default) streams every frame the encoder produces. + self.server.serve_interval = 1 / serve_fps if serve_fps > 0 else 0 self.server.auth = None if server_user and server_password: str_auth = f"{server_user}:{server_password}" @@ -991,12 +997,16 @@ def check_auth(self) -> bool: def send_jpeg(self, output: StreamingOutput) -> None: """Send a JPEG image.""" + # Take a reference under the lock but write outside it: the encoder + # holds the same condition in StreamingOutput.write, so a slow + # client must not block frame production. with output.condition: output.condition.wait() - self.send_header("Content-Type", "image/jpeg") - self.send_header("Content-Length", len(output.frame)) - self.end_headers() - self.wfile.write(output.frame) + frame = output.frame + self.send_header("Content-Type", "image/jpeg") + self.send_header("Content-Length", len(frame)) + self.end_headers() + self.wfile.write(frame) def do_GET(self) -> None: """Handle a GET request.""" @@ -1015,6 +1025,11 @@ def do_GET(self) -> None: self.send_jpeg(output) self.wfile.write(b"\r\n") self.wfile.flush() + # Throttle the stream: send_jpeg blocks until the + # next frame, so sleeping here drops the ones in + # between instead of pushing them over the network. + if self.server.serve_interval: + time.sleep(self.server.serve_interval) except IOError as err: self.logger().error( "Exception while serving client %s: %s", From a64a6ce4eec9de05793eb1805a4a4940c5325e7e Mon Sep 17 00:00:00 2001 From: Alexandros Evangelou Date: Fri, 21 Aug 2026 13:10:12 +0300 Subject: [PATCH 20/20] add log in errors in the camera threads --- src/ethopy/interfaces/Camera.py | 57 +++++++++++++++++++++++++++------ 1 file changed, 47 insertions(+), 10 deletions(-) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index baa2607..7cd2d6f 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -296,11 +296,32 @@ def setup(self) -> None: """ self.frame_queue = Queue() # self.process_queue.cancel_join_thread() - self.capture_runner = threading.Thread(target=self.rec) + self.capture_runner = threading.Thread( + target=self._run_guarded, args=(self.rec,) + ) self.write_runner = threading.Thread( - target=self.dequeue, args=(self.frame_queue,) + target=self._run_guarded, args=(self.dequeue, self.frame_queue) ) + def _run_guarded(self, func: Any, *args: Any) -> None: + """Run a recording thread target, logging whatever it raises. + + An unhandled exception in a thread only reaches threading.excepthook, + so it never lands in the ethopy log, and self.stop stays clear - which + leaves dequeue() spinning and the whole camera subprocess alive with a + closed camera (and still holding the streaming port). + """ + try: + func(*args) + except Exception: + log.exception( + "Camera %s: %s failed, stopping recording.", + self.filename, + getattr(func, "__name__", func), + ) + finally: + self.stop.set() + def start_rec(self) -> None: """ Start the capture and write runners with exception handling. @@ -312,6 +333,7 @@ def start_rec(self) -> None: self.capture_runner.join() self.write_runner.join() except Exception as cam_error: + log.exception("Camera %s: recording setup failed.", self.filename) raise RuntimeError( f"Exception occurred during recording: {cam_error}" ) from cam_error @@ -789,7 +811,6 @@ def rec(self) -> None: def recording_init(self) -> None: """Initialize the recording.""" - self.stop.clear() self.recording.set() self.cam = self.init_cam() self._start_http_server() @@ -803,13 +824,29 @@ def _start_http_server(self) -> None: """ if self.serve_port <= 0: return - self.httpthread = HTTPServerThread( - self, - serve_port=self.serve_port, - server_user=self.server_user, - server_password=self.server_password, - serve_fps=self.serve_fps, - ) + # One port per camera, so several cameras in one setup do not all try + # to bind server.port. + port = self.serve_port + self.camera_num + try: + self.httpthread = HTTPServerThread( + self, + serve_port=port, + server_user=self.server_user, + server_password=self.server_password, + serve_fps=self.serve_fps, + ) + except OSError: + # Streaming is an accessory: a port that is busy (usually a camera + # process left over from an earlier run) must not stop the + # recording. + self.httpthread = None + log.exception( + "Camera %s: could not serve on port %s, continuing without " + "the video stream.", + self.filename, + port, + ) + return self.httpthread.start() def init_cam(self) -> "Picamera2":