16432267384

Committed 22 Jul 2025 01:33AM UTC coverage: 77.702% (-2.1%) from 79.825%

Build # 16432267384

Build Type

Pull #83

github

Committed by

web-flow

Commit Message

Merge 891a19f80 into 455a51694

Pull Request Pull Request #83: Add video preprocessing (denoising) feature

Run Details

473 of 669 new or added lines in 11 files covered. (70.7%)

35 existing lines in 2 files now uncovered.

1819 of 2341 relevant lines covered (77.7%)

11.34 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.
        """
        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.

        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	"""
53	output_path = Path(output_path)	15✔
54	if self.frame is None:	15✔
NEW 55	logger.warning(f"No frame data provided for {self.name}. Skipping export.")	×
NEW 56	return	×
57	if suffix:	15✔
58	output_path = output_path.with_name(output_path.stem + f"_{self.name}")	15✔
59	cv2.imwrite(str(output_path.with_suffix(".png")), self.frame)	15✔
60	logger.info(	15✔
61	f"Writing frame to {output_path}.png: {self.frame.shape[1]}x{self.frame.shape[0]}"
62	)
63
64	def display(self, binary: bool = False) -> None:	15✔
65	"""
66	Display the frame data.
67
68	Parameters
69	----------
70	binary : bool
71	If True, the frame will be scaled to the full range of uint8.
72	"""
NEW 73	if self.frame is None:	×
NEW 74	logger.warning(f"No frame data provided for {self.name}. Skipping display.")	×
NEW 75	return	×
76
NEW 77	frame_to_display = self.frame	×
NEW 78	if binary:	×
NEW 79	frame_to_display = cv2.normalize(	×
80	self.frame, None, 0, np.iinfo(np.uint8).max, cv2.NORM_MINMAX
81	).astype(np.uint8)
NEW 82	cv2.imshow(self.name, frame_to_display)	×
83	while True:
NEW 84	if cv2.waitKey(1) == 27:	×
NEW 85	break	×
NEW 86	cv2.destroyAllWindows()	×
NEW 87	cv2.waitKey(1) # Extra waitKey to properly close the window	×
88
89
90	class NamedVideo(NamedBaseFrame):	15✔
91	"""
92	Pydantic model to store a video together with a name.
93	"""
94
95	video: Optional[List[np.ndarray]] = Field(	15✔
96	None,
97	description="List of frames.",
98	)
99
100	def export(self, output_path: Union[Path, str], suffix: bool = False, fps: float = 20) -> None:	15✔
101	"""
102	Export the frame data to a file.
103	"""
104	if self.video is None or self.video == []:	15✔
NEW 105	logger.warning(f"No frame data provided for {self.name}. Skipping export.")	×
NEW 106	return	×
107	output_path = Path(output_path)	15✔
108	if suffix:	15✔
109	output_path = output_path.with_name(output_path.stem + f"_{self.name}")	15✔
110	if not all(isinstance(frame, np.ndarray) for frame in self.video):	15✔
NEW 111	raise ValueError("Not all frames are numpy arrays.")	×
112	writer = VideoWriter(	15✔
113	path=output_path.with_suffix(".avi"),
114	fps=fps,
115	)
116	logger.info(	15✔
117	f"Writing video to {output_path}.avi:"
118	f"{self.video[0].shape[1]}x{self.video[0].shape[0]}"
119	)
120	try:	15✔
121	for frame in self.video:	15✔
122	picture = cv2.cvtColor(frame, cv2.COLOR_GRAY2BGR)	15✔
123	writer.write_frame(picture)	15✔
124	finally:
125	writer.close()	15✔

Aharoni-Lab / mio / 16432267384

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