diff --git a/src/ethopy/core/experiment.py b/src/ethopy/core/experiment.py index 2918943..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,30 +595,11 @@ 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]]): - cond_key = {} - for k in fields: - if isinstance(condition[k], (int, float, str)): - cond_key[k] = condition[k] - else: - cond_key[k] = condition[k][idx] - - 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/core/interface.py b/src/ethopy/core/interface.py index 87d2c2e..f0e2849 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,46 @@ def _initialize_hardware(self) -> None: self._initialize_camera() def _initialize_camera(self) -> None: - """Initialize camera if configured in setup.""" + """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", 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 is the /dev/videoN (V4L2) or libcamera index. + 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 +227,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. 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() def load_calibration(self) -> None: """Load port calibration data from database. @@ -441,6 +455,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/core/logger.py b/src/ethopy/core/logger.py index 4249c99..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() @@ -1014,7 +1017,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) diff --git a/src/ethopy/interfaces/Camera.py b/src/ethopy/interfaces/Camera.py index 184cef7..7cd2d6f 100644 --- a/src/ethopy/interfaces/Camera.py +++ b/src/ethopy/interfaces/Camera.py @@ -82,13 +82,29 @@ 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: 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 @@ -122,11 +138,14 @@ def __init__( ), block=True, ) - h5s_filename = ( - f"animal_id_{logger.trial_key['animal_id']}" - f"_session_{logger.trial_key['session']}.h5" + # Per-camera name so two cameras in a session don't overwrite each + # other's h5; the file is written later by Logger.createDataset. + self.filename_tmst = f"videotmst_{self.filename}.h5" + h5_target_path = ( + logger.target_path + recordings_folder + if os.path.isdir(logger.target_path) + else False ) - self.filename_tmst = "videosssctmst" + h5s_filename logger.log_recording( dict( rec_aim="sync", @@ -134,7 +153,7 @@ def __init__( version="0.1", filename=self.filename_tmst, source_path=self.source_path, - target_path=self.target_path, + target_path=h5_target_path, ), block=True, ) @@ -201,7 +220,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. @@ -209,53 +228,67 @@ 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)) - log.info(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}") - else: - log.error(f"Failed to transfer file: {file.name}") - except FileNotFoundError as ex: + shutil.copy(str(file), str(destination)) + log.debug(f"Transferred file: {file.name}") + # 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 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(): - raise ValueError( - f"Source path {source} does not exist or is not a directory." - ) - - if not target.exists(): - 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") + return + + 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) - # Clean up if the source directory is empty - if not any(source.iterdir()): - source.rmdir() - log.info(f"Deleted the empty folder: {source}") + 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: """ @@ -263,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. @@ -279,7 +333,10 @@ 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}" + log.exception("Camera %s: recording setup failed.", self.filename) + raise RuntimeError( + f"Exception occurred during recording: {cam_error}" + ) from cam_error def dequeue(self, frame_queue: Queue) -> None: """ @@ -295,18 +352,22 @@ 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 (the stop event alone signals shutdown). """ - 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: + pass # still alive after terminate(); the OS reaps it once it exits + self.camera_process = None @abstractmethod def rec(self) -> None: @@ -346,6 +407,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 +417,15 @@ 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. + 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. @@ -362,6 +433,7 @@ def __init__( """ self.fps = fps + self.camera_num = camera_num self.video_output = None self.dataset = None self.tmst_output = None @@ -377,6 +449,8 @@ 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 "" + self._last_frame_err_log = 0.0 # throttles the per-frame read-error log if not globals()["IMPORT_CV2"]: raise ImportError( @@ -385,14 +459,47 @@ def __init__( "You can install cv2 using pip:\n" 'sudo pip3 install opencv-python"' ) - self.camera = cv2.VideoCapture(0, cv2.CAP_V4L2) - if not self.camera.isOpened(): - raise RuntimeError( - "No camera is available. Please check if the camera is connected and functional." - ) - self.camera.release() + # 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 camera to a target cv2.VideoCapture can open. + + 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 the camera is " + "connected and 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") + ] + # 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)} device(s) " + f"under {by_id} (expected 1). Available: {available}" + ) + def setup(self): """Setup the camera.""" out_vid_fn = self.source_path + self.filename + ".mp4" @@ -462,7 +569,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: @@ -474,8 +581,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.""" @@ -484,25 +594,31 @@ def camera_opened(self, camera): return True def recording_init(self): - self.camera = cv2.VideoCapture(0, 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." ) + # 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) 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!!" ) + # 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) 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) @@ -520,6 +636,12 @@ 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 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" + ) def rec(self): """ @@ -541,7 +663,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: @@ -557,7 +682,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): """ @@ -589,6 +717,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, @@ -597,6 +726,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 @@ -605,6 +744,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 @@ -663,19 +803,57 @@ 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() def recording_init(self) -> None: """Initialize the recording.""" - 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 + # 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": """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}, @@ -703,11 +881,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, server_user=self.server_user, server_password=self.server_password - ) - self.httpthread.start() picam2.start_encoder(encoder, output) return picam2 @@ -717,8 +890,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() @@ -726,7 +902,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: @@ -810,6 +986,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__) @@ -817,6 +994,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}" @@ -855,12 +1034,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.""" @@ -879,6 +1062,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", diff --git a/src/ethopy/interfaces/dlc.py b/src/ethopy/interfaces/dlc.py index 544c330..33a7698 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,16 @@ 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) - + # 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): """ @@ -82,7 +89,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 +101,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 +148,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 +166,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 +188,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 +228,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 +419,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 +580,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.") 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. 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): diff --git a/tests/test_task_helper_funcs.py b/tests/test_task_helper_funcs.py new file mode 100644 index 0000000..453e7c5 --- /dev/null +++ b/tests/test_task_helper_funcs.py @@ -0,0 +1,116 @@ +"""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_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]} + with pytest.raises(ValueError, match="unequal length"): + expand_condition_rows( + condition, {"cond_hash", "port", "loc_x"}, ["port", "loc_x"] + )