Skip to content

slice

Bases: FileOperation, FileRemoverMixin

An operation to extract images (frames) from video files.

This class processes video files in source directory that match a specific pattern and saves their frames into the target directory. It can also automatically delete the source video file after the slicing process is finished.

Attributes:

Name Type Description
step_sec float

The time interval in seconds between extracted frames.
suffix str

The file extension for the output images (e.g., '.jpg').
remove bool

If True, the source video is deleted after processing.
slicer VideoSlicer

The tool used to perform the actual video slicing.
Source code in file_operations/slice.py 
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
class SliceOperation(FileOperation, FileRemoverMixin):
    """
    An operation to extract images (frames) from video files.

    This class processes video files in source directory that match a specific
    pattern and saves their frames into the target directory. It can also
    automatically delete the source video file after the slicing process is finished.

    Attributes:
        step_sec (float): The time interval in seconds between extracted frames.
        suffix (str): The file extension for the output images (e.g., '.jpg').
        remove (bool): If True, the source video is deleted after processing.
        slicer (VideoSlicer): The tool used to perform the actual video slicing.
    """
    def __init__(self, **kwargs):
        """
        Initializes the slice operation with the required parameters.

        Args:
            **kwargs (dict): Arguments including 'step_sec', 'type', and 'remove'
                flags, usually passed from the command line or settings.
        """
        super().__init__(**kwargs)
        self.step_sec: float = kwargs.get("step_sec", self.settings.step_sec)
        self.suffix: str = kwargs.get('type', self.settings.suffix)
        self.remove: bool = kwargs.get('remove', self.settings.remove)
        self.slicer: VideoSlicer = VideoSlicer()


    @staticmethod
    def add_arguments(settings, parser: argparse.ArgumentParser) -> None:
        """
        Defines CLI arguments for the video slicing task.

        Args:
            settings (AppSettings): Global configuration for default values.
            parser (argparse.ArgumentParser): The parser to which arguments are added.
        """
        parser.add_argument(Arguments.dst, help=HelpStrings.dst)
        parser.add_argument(
            Arguments.remove, Arguments.rm,
            help=HelpStrings.remove,
            action='store_true'
        )
        parser.add_argument(
            Arguments.type, Arguments.t,
            help=HelpStrings.type,
            default=settings.suffix
        )
        parser.add_argument(
            Arguments.step_sec, Arguments.step,
            help=HelpStrings.step_sec,
            default=settings.step_sec
        )


    def do_task(self):
        """
        Processes the collected video files one by one.

        This method uses the 'VideoSlicer' tool to extract frames. It logs
        how many images were created for each video. If the 'remove' flag
        is enabled, it deletes the source video using 'FileRemoverMixin'.
        """
        for file_path in self.files_for_task:
            if file_path.is_file():
                ret, sliced_count = self.slicer.slice(
                    source_file=file_path,
                    target_dir=self.target_directory,
                    suffix=self.suffix,
                    step=self.step_sec
                )

                if ret:
                    self.logger.info(f"{file_path} sliced to {sliced_count} images")
                else:
                    self.logger.warning(f"Unable to read {file_path}. Not sliced.")
                    continue

                if self.remove:
                    self.remove_all(file_path)


    @property
    def step_sec(self) -> float:
        """float: Returns the time interval between frames."""
        return self._step_sec


    @step_sec.setter
    def step_sec(self, value: Union[float, int, str]) -> None:
        """
        Sets the time interval and ensures it is a float value.

        Args:
            value (Union[int, float, str]): The interval value to be converted.
        """
        if not isinstance(value, float):
            value = float(value)

        self._step_sec = value

step_sec property writable

float: Returns the time interval between frames.

__init__(**kwargs)

Initializes the slice operation with the required parameters.

Parameters:

Name Type Description Default
**kwargs dict

Arguments including 'step_sec', 'type', and 'remove' flags, usually passed from the command line or settings.

 {}
Source code in file_operations/slice.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
def __init__(self, **kwargs):
    """
    Initializes the slice operation with the required parameters.

    Args:
        **kwargs (dict): Arguments including 'step_sec', 'type', and 'remove'
            flags, usually passed from the command line or settings.
    """
    super().__init__(**kwargs)
    self.step_sec: float = kwargs.get("step_sec", self.settings.step_sec)
    self.suffix: str = kwargs.get('type', self.settings.suffix)
    self.remove: bool = kwargs.get('remove', self.settings.remove)
    self.slicer: VideoSlicer = VideoSlicer()

add_arguments(settings, parser) staticmethod

Defines CLI arguments for the video slicing task.

Parameters:

Name Type Description Default
settings AppSettings

Global configuration for default values.

 required
parser ArgumentParser

The parser to which arguments are added.

 required
Source code in file_operations/slice.py 
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
@staticmethod
def add_arguments(settings, parser: argparse.ArgumentParser) -> None:
    """
    Defines CLI arguments for the video slicing task.

    Args:
        settings (AppSettings): Global configuration for default values.
        parser (argparse.ArgumentParser): The parser to which arguments are added.
    """
    parser.add_argument(Arguments.dst, help=HelpStrings.dst)
    parser.add_argument(
        Arguments.remove, Arguments.rm,
        help=HelpStrings.remove,
        action='store_true'
    )
    parser.add_argument(
        Arguments.type, Arguments.t,
        help=HelpStrings.type,
        default=settings.suffix
    )
    parser.add_argument(
        Arguments.step_sec, Arguments.step,
        help=HelpStrings.step_sec,
        default=settings.step_sec
    )

do_task()

Processes the collected video files one by one.

This method uses the 'VideoSlicer' tool to extract frames. It logs how many images were created for each video. If the 'remove' flag is enabled, it deletes the source video using 'FileRemoverMixin'.

Source code in file_operations/slice.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
def do_task(self):
    """
    Processes the collected video files one by one.

    This method uses the 'VideoSlicer' tool to extract frames. It logs
    how many images were created for each video. If the 'remove' flag
    is enabled, it deletes the source video using 'FileRemoverMixin'.
    """
    for file_path in self.files_for_task:
        if file_path.is_file():
            ret, sliced_count = self.slicer.slice(
                source_file=file_path,
                target_dir=self.target_directory,
                suffix=self.suffix,
                step=self.step_sec
            )

            if ret:
                self.logger.info(f"{file_path} sliced to {sliced_count} images")
            else:
                self.logger.warning(f"Unable to read {file_path}. Not sliced.")
                continue

            if self.remove:
                self.remove_all(file_path)