16457410971

Committed 22 Jul 2025 06:50AM UTC coverage: 77.726% (-2.1%) from 79.825%

Build # 16457410971

Build Type

push

github

Committed by

web-flow

Commit Message

Merge pull request #83 from Aharoni-Lab/feat-preprocess

Add video preprocessing (denoising) feature

Run Details

478 of 675 new or added lines in 11 files covered. (70.81%)

1 existing line in 1 file now uncovered.

1825 of 2348 relevant lines covered (77.73%)

11.35 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

68.52

/mio/models/frames.py

"""
Pydantic models for storing frames and videos.
"""

from pathlib import Path
from typing import List, Optional, Union

import cv2
import numpy as np
from pydantic import BaseModel, ConfigDict, Field

from mio.io import VideoWriter
from mio.logging import init_logger

logger = init_logger("model.frames")


class NamedBaseFrame(BaseModel):
    """
    Pydantic model to store an an image or a video together with a name.
    """

    name: str = Field(
        ...,
        description="Name of the video.",
    )
    model_config = ConfigDict(
        arbitrary_types_allowed=True,
    )

    def export(self, output_path: Union[Path, str], fps: int, suffix: bool) -> None:
        """
        Export the frame data to a file.
        The implementation needs to be defined in the derived classes.
        """
        raise NotImplementedError("Method not implemented.")


class NamedFrame(NamedBaseFrame):
    """
    Pydantic model to store an image or a video together with a name.
    """

    frame: Optional[np.ndarray] = Field(
        None,
        description="Frame data, if provided.",
    )

    def export(self, output_path: Union[Path, str], suffix: bool = False) -> None:
        """
        Export the frame data to a file.
        The file name will be a concatenation of the output path and the name of the frame.
        """
        output_path = Path(output_path)
        if self.frame is None:
            logger.warning(f"No frame data provided for {self.name}. Skipping export.")
            return
        if suffix:
            output_path = output_path.with_name(output_path.stem + f"_{self.name}")
        cv2.imwrite(str(output_path.with_suffix(".png")), self.frame)
        logger.info(
            f"Writing frame to {output_path}.png: {self.frame.shape[1]}x{self.frame.shape[0]}"
        )

    def display(self, binary: bool = False) -> None:
        """
        Display the frame data in a opencv window. Press ESC to close the window.

        Parameters
        ----------
        binary : bool
            If True, the frame will be scaled to the full range of uint8.
        """
        if self.frame is None:
            logger.warning(f"No frame data provided for {self.name}. Skipping display.")
            return

        frame_to_display = self.frame
        if binary:
            frame_to_display = cv2.normalize(
                self.frame, None, 0, np.iinfo(np.uint8).max, cv2.NORM_MINMAX
            ).astype(np.uint8)
        cv2.imshow(self.name, frame_to_display)
        while True:
            if cv2.waitKey(1) == 27:
                break
        cv2.destroyAllWindows()
        cv2.waitKey(1)  # Extra waitKey to properly close the window


class NamedVideo(NamedBaseFrame):
    """
    Pydantic model to store a video together with a name.
    """

    video: Optional[List[np.ndarray]] = Field(
        None,
        description="List of frames.",
    )

    def export(self, output_path: Union[Path, str], suffix: bool = False, fps: float = 20) -> None:
        """
        Export the frame data to a file.
        """
        if self.video is None or self.video == []:
            logger.warning(f"No frame data provided for {self.name}. Skipping export.")
            return
        output_path = Path(output_path)
        if suffix:
            output_path = output_path.with_name(output_path.stem + f"_{self.name}")
        if not all(isinstance(frame, np.ndarray) for frame in self.video):
            raise ValueError("Not all frames are numpy arrays.")
        writer = VideoWriter(
            path=output_path.with_suffix(".avi"),
            fps=fps,
        )
        logger.info(
            f"Writing video to {output_path}.avi:"
            f"{self.video[0].shape[1]}x{self.video[0].shape[0]}"
        )
        try:
            for frame in self.video:
                picture = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)
                writer.write_frame(picture)
        finally:
            writer.close()

1	"""
2	Pydantic models for storing frames and videos.
3	"""
4
5	from pathlib import Path	15✔
6	from typing import List, Optional, Union	15✔
7
8	import cv2	15✔
9	import numpy as np	15✔
10	from pydantic import BaseModel, ConfigDict, Field	15✔
11
12	from mio.io import VideoWriter	15✔
13	from mio.logging import init_logger	15✔
14
15	logger = init_logger("model.frames")	15✔
16
17
18	class NamedBaseFrame(BaseModel):	15✔
19	"""
20	Pydantic model to store an an image or a video together with a name.
21	"""
22
23	name: str = Field(	15✔
24	...,
25	description="Name of the video.",
26	)
27	model_config = ConfigDict(	15✔
28	arbitrary_types_allowed=True,
29	)
30
31	def export(self, output_path: Union[Path, str], fps: int, suffix: bool) -> None:	15✔
32	"""
33	Export the frame data to a file.
34	The implementation needs to be defined in the derived classes.
35	"""
NEW 36	raise NotImplementedError("Method not implemented.")	×
37
38
39	class NamedFrame(NamedBaseFrame):	15✔
40	"""
41	Pydantic model to store an image or a video together with a name.
42	"""
43
44	frame: Optional[np.ndarray] = Field(	15✔
45	None,
46	description="Frame data, if provided.",
47	)
48
49	def export(self, output_path: Union[Path, str], suffix: bool = False) -> None:	15✔
50	"""
51	Export the frame data to a file.
52	The file name will be a concatenation of the output path and the name of the frame.
53	"""
54	output_path = Path(output_path)	15✔
55	if self.frame is None:	15✔
NEW 56	logger.warning(f"No frame data provided for {self.name}. Skipping export.")	×
NEW 57	return	×
58	if suffix:	15✔
59	output_path = output_path.with_name(output_path.stem + f"_{self.name}")	15✔
60	cv2.imwrite(str(output_path.with_suffix(".png")), self.frame)	15✔
61	logger.info(	15✔
62	f"Writing frame to {output_path}.png: {self.frame.shape[1]}x{self.frame.shape[0]}"
63	)
64
65	def display(self, binary: bool = False) -> None:	15✔
66	"""
67	Display the frame data in a opencv window. Press ESC to close the window.
68
69	Parameters
70	----------
71	binary : bool
72	If True, the frame will be scaled to the full range of uint8.
73	"""
NEW 74	if self.frame is None:	×
NEW 75	logger.warning(f"No frame data provided for {self.name}. Skipping display.")	×
NEW 76	return	×
77
NEW 78	frame_to_display = self.frame	×
NEW 79	if binary:	×
NEW 80	frame_to_display = cv2.normalize(	×
81	self.frame, None, 0, np.iinfo(np.uint8).max, cv2.NORM_MINMAX
82	).astype(np.uint8)
NEW 83	cv2.imshow(self.name, frame_to_display)	×
84	while True:
NEW 85	if cv2.waitKey(1) == 27:	×
NEW 86	break	×
NEW 87	cv2.destroyAllWindows()	×
NEW 88	cv2.waitKey(1) # Extra waitKey to properly close the window	×
89
90
91	class NamedVideo(NamedBaseFrame):	15✔
92	"""
93	Pydantic model to store a video together with a name.
94	"""
95
96	video: Optional[List[np.ndarray]] = Field(	15✔
97	None,
98	description="List of frames.",
99	)
100
101	def export(self, output_path: Union[Path, str], suffix: bool = False, fps: float = 20) -> None:	15✔
102	"""
103	Export the frame data to a file.
104	"""
105	if self.video is None or self.video == []:	15✔
NEW 106	logger.warning(f"No frame data provided for {self.name}. Skipping export.")	×
NEW 107	return	×
108	output_path = Path(output_path)	15✔
109	if suffix:	15✔
110	output_path = output_path.with_name(output_path.stem + f"_{self.name}")	15✔
111	if not all(isinstance(frame, np.ndarray) for frame in self.video):	15✔
NEW 112	raise ValueError("Not all frames are numpy arrays.")	×
113	writer = VideoWriter(	15✔
114	path=output_path.with_suffix(".avi"),
115	fps=fps,
116	)
117	logger.info(	15✔
118	f"Writing video to {output_path}.avi:"
119	f"{self.video[0].shape[1]}x{self.video[0].shape[0]}"
120	)
121	try:	15✔
122	for frame in self.video:	15✔
123	picture = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)	15✔
124	writer.write_frame(picture)	15✔
125	finally:
126	writer.close()	15✔

Aharoni-Lab / mio / 16457410971

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous