Skip to content

Engines

Info

The engines are adapter classes that facilitate the interaction with underlying libraries, simplifying the processing of multimedia resources. Each engine establishes a contract through a protocol, which allows for the extension to new engines, as well as smooth uniform communication and collaboration.

ImageEngine

Bases: Engine

Engine that adapts the Pillow library to support image processing.

Usage:

# adapt pillow Image
library = PIL.open(media.path)
return ImageEngine(library)
Source code in nucleus/sdk/processing/engines.py 
 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
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class ImageEngine(Engine):
    """Engine that adapts the Pillow library to support image processing.

    Usage:

        # adapt pillow Image
        library = PIL.open(media.path)
        return ImageEngine(library)
    """

    def __init__(self, lib: Pillow):
        # compile the pattern to avoid overhead in loop and bind underlying lib
        self._pattern = re.compile(r'(?<!^)(?=[A-Z])')
        super().__init__(lib)

    def _to_snake_case(self, class_name: str) -> str:
        """Transform PascalCase class definition to snake_case method name.

        :para name: The class name to parse
        :return: The snake case version for class name
        """
        return self._pattern.sub('_', class_name).lower()

    def _setup_methods(self):
        """Call and chain methods based on settings"""
        for class_name, params in self.compile():
            # The method to call should be the same as the option name.
            method = self._to_snake_case(class_name)
            func = getattr(self._library, method)
            # pillow chaining features
            # all methods return a new instance of the Image class, holding the resulting image
            # ref:
            # https://pillow.readthedocs.io/en/stable/reference/Image.html#PIL.Image.Image
            self._library = func(**dict(params))

    def introspect(self, path: Path) -> Introspection:
        # fet the mime type from file path
        (mime_type, _) = mimetypes.guess_type(path)
        # get attributes from PIL.Image object
        members = inspect.getmembers(PIL.Image.open(path))
        filter_private = filter(lambda x: not x[0].startswith('_'), members)
        filter_method = filter(lambda x: not inspect.ismethod(x[1]), filter_private)
        image_introspection = _to_object(dict(filter_method))
        # patch to avoid size conflict keyword
        delattr(image_introspection, 'size')

        # extend introspection with custom PIL.Image attributes
        return Introspection(
            size=path.size(),
            type=str(mime_type),
            **vars(image_introspection),
        )

    def save(self, path: Path) -> File:
        # We generate the expected path after processing
        try:
            self._setup_methods()
            self._library.save(path)

            # after low level processing happen!!
            i8t = self.introspect(path)
            return File(path=path, meta=i8t)
        except Exception as e:
            # Standard exceptions raised
            raise ProcessingEngineError(f'error while trying to save image output: {str(e)}')

VideoEngine

Bases: Engine

Engine that adapts the FFMPEG Python library to support low-level transcoding.

Usage:

# adapt ffmpeg lib
library = ffmpeg.input(media.path)
return VideoEngine(library)
Source code in nucleus/sdk/processing/engines.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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
class VideoEngine(Engine):
    """Engine that adapts the FFMPEG Python library to support low-level transcoding.

    Usage:

        # adapt ffmpeg lib
        library = ffmpeg.input(media.path)
        return VideoEngine(library)
    """

    def __init__(self, lib: FFMPEG):
        super().__init__(lib)

    def _build_output_args(self) -> ChainMap[Any, Any]:
        """Join config as output arguments for ffmpeg"""
        mapped_args = [y for _, y in self.compile()]
        return ChainMap(*mapped_args)

    def introspect(self, path: Path) -> Introspection:
        # process the arg path or use the current media file path
        (mime_type, _) = mimetypes.guess_type(path)
        video_introspection = _to_object(processing.probe(path))

        # extend introspection with custom video ffprobe
        return Introspection(
            size=path.size(),
            type=str(mime_type),
            **vars(video_introspection),
        )

    def save(self, path: Path) -> File:
        # TODO allow see ffmpeg progress
        # TODO pubsub? Observer: Keep reading on event?
        try:
            output_args = self._build_output_args()
            # We generate the expected path after transcode
            self._library.output(path, **output_args).run()

            # after low level processing happen!!
            i8t = self.introspect(path)
            return File(path=path, meta=i8t)
        except Exception as e:
            # Standard exceptions raised
            raise ProcessingEngineError(f'error while trying to save video output: {str(e)}')