Video preprocessor

This module contains functions for pre-processing video data.

class mio.process.video.BaseVideoProcessor(name: str, output_dir: Path, force: bool = False)

Base class for defining an abstract video processor.

Attributes: name (str): The name of the video processor. output_frames (list): A list of output frames. named_frame (NamedFrame): A NamedFrame object.

append_output_frame(input_frame: ndarray) → None

Append a frame to the output_frames list.

Parameters: frame (np.ndarray): The frame to append.

batch_export_videos() → None

Batch export the videos to a file. This method should be overridden in the subclass.

export_output_video() → None

Export the video to a file.

property output_named_video: NamedVideo

Get the output NamedFrame object.

Returns: NamedVideo: The output NamedVideo object.

process_frame(input_frame: ndarray, index: int) → None

Process a single frame. This method should be implemented in the subclass.

Parameters: frame (np.ndarray): The frame to process.

class mio.process.video.FreqencyMaskProcessor(name: str, freq_mask_config: FrequencyMaskingConfig, width: int, height: int, output_dir: Path, force: bool = False)

A class to apply frequency masking to a video.

batch_export_videos() → None

Batch export the videos to a file. Whether to export or not is controlled in each method.

export_freq_domain_frames() → None

Export the frequency domain to a file.

export_freq_mask() → None

Export the frequency mask to a file.

property freq_domain_named_video: NamedVideo

Get the NamedFrame object for the frequency domain.

property freq_mask: ndarray

Get the frequency mask.

property freq_mask_named_frame: NamedFrame

Get the NamedFrame object for the frequency mask.

process_frame(input_frame: ndarray, index: int) → ndarray | None

Process a single frame.

Parameters: frame (np.ndarray): The frame to process.

Returns: Optional[np.ndarray]: The processed frame. If the input is none, a None is returned.

class mio.process.video.MinProjSubtractProcessor(name: str, minimum_projection_config: MinimumProjectionConfig, output_dir: Path, video_frames: list[ndarray], force: bool = False)

A class to apply minimum projection to a video.

batch_export_videos() → None

Batch export the videos to a file. Whether to export or not is controlled in each method.

property min_proj_named_frame: NamedFrame

Get the NamedFrame object for the minimum projection.

normalize_stack() → None

Normalize the stack of images.

class mio.process.video.NoisePatchProcessor(name: str, noise_patch_config: NoisePatchConfig, output_dir: Path, force: bool = False)

A class to apply noise patching to a video.

batch_export_videos() → None

Batch export the videos to a file. Whether to export or not is controlled in each method.

export_noise_patch() → None

Export the noise patch to a file.

export_noisy_video() → None

Export the noisy frames to a file.

property noise_patch_named_video: NamedVideo

Get the NamedFrame object for the noise patch.

property noisy_frames_named_video: NamedVideo

Get the NamedFrame object for the noisy frames.

process_frame(input_frame: ndarray, index: int) → ndarray | None

Process a single frame.

Parameters:

  • input_frame (np.ndarray) – The frame to process.

  • index (int) – The frame number within the video.

Returns:

The processed frame. If the frame is noisy, a None is returned.

Return type:

Optional[np.ndarray]

class mio.process.video.PassThroughProcessor(name: str, output_dir: Path, force: bool = False)

A class to pass through a video.

batch_export_videos() → None

Batch export the videos to a file. Whether to export or not is controlled in each method.

property pass_through_named_video: NamedVideo

Get the NamedFrame object for the pass through.

process_frame(input_frame: ndarray, index: int) → ndarray

Process a single frame.

Parameters: frame (np.ndarray): The frame to process.

Returns: np.ndarray: The processed frame.

mio.process.video._drop_frames_by_index(df: DataFrame, dropped_frame_indices: Collection[int], index_column: str = 'reconstructed_frame_index') → DataFrame

Drop rows whose index_column value is in the dropped frame indices, and then recreate index_column monotonically increasing from 0.

mio.process.video._make_frame_timestamp_csv(csv_df: DataFrame[StreamBufferTable]) → DataFrame[TimestampTable]

Export a frame-timestamp CSV file mapping reconstructed_frame_index to unix timestamps.

The CSV includes both the first and last buffer timestamps for each frame.

Todo

Find a better place for this! the metadata csv operations deserve their own place!

Parameters:

csv_df (pd.DataFrame) – The modified CSV DataFrame with reconstructed_frame_index and buffer_recv_unix_time.

mio.process.video.denoise(video_path: Path, config: DenoiseConfig, csv_df: DataFrame | None = None, debug_dir: Path | None = None, progress: bool = False, force: bool = False) → Path

Preprocess a video file and display the results.

Parameters:

  • video_path (str) – The path to the video file.

  • config (DenoiseConfig) – The denoise configuration.

  • csv_df (Optional[pd.DataFrame], optional) – Pre-validated CSV DataFrame. If provided, uses this instead of reading from disk.

  • debug_dir (Optional[Path], optional) – Directory for intermediate/debug files. If provided, intermediate files (noisy frames, patches, frequency masks, etc.) will be saved here instead of the main output_dir. Main output files (output video and CSV) will still be saved to output_dir.

  • progress (bool) – Display a progress bar

  • force (bool) – Overwrite any existing files

Returns:

Path to the output video file.

Return type:

Path

mio.process.video.remove_frames(recording: Recording | Path, remove_indices: list[int], output_path: Path, force: bool = False, progress: bool = False) → Recording

Remove specific frames by index from a video.

Parameters:

  • recording (Recording | Path) – Recording object, or path to video.

  • remove_indices (list[int]) – 0-based frame indices to remove.

  • output_path (Optional[str], optional) – Path to the output video file. If None, defaults to input path with “_removed” suffix.

  • force (bool) – If True, overwrite any existing file. Default False.

  • progress (bool) – If True, show a progress bar. Default False.

Returns:

The video and generated metadata

Return type:

Recording

mio.process.video.score_noise(recording: Recording, config: NoisePatchConfig, progress: bool = False) → DataFrame[NoiseTable]

Score framewise noise from a recording, yielding a dataframe with columns for each kind of noise

  • reconstructed_frame_index: the index of the frame in the video

  • gradient: the number of pixels that are part of noise patches, as determined by the second row-wise derivative being above the configured threshold

  • black_area: the number of pixels in contiguous black regions (and thus missing)

mio.process.video.trim(video_path: Path, output_path: Path | None = None, csv_df: DataFrame | None = None, start: int = 0, end: int = 0, progress: bool = False, force: bool = False) → Path

Crop a video file by trimming frames from both ends.

Parameters:

  • video_path (str) – The path to the input video file.

  • output_path (Optional[str], optional) – The path to the output video file. If None, defaults to input path with “_trimmed” suffix.

  • csv_df (Optional[pd.DataFrame], optional) – Pre-validated CSV DataFrame. If provided, uses this instead of reading from disk.

  • start (int) – Number of frames to remove from the beginning. Default 0.

  • end (int) – Number of frames to remove from the end. Default 0.

  • progress (bool) – Display a progress bar

  • force (bool) – If True, overwrite any existing file.

Returns:

Path to the output video file.

Return type:

Path