Skip to content

Utils

MgProgressbar

MgProgressbar(total=100, time_limit=0.5, prefix='Progress', suffix='Complete', decimals=1, length=40, fill='█')

Calls in a loop to create terminal progress bar.

Initialize the MgProgressbar object.

Parameters:

Name Type Description Default
total int

Total iterations. Defaults to 100.

100
time_limit float

The minimum refresh rate of the progressbar in seconds. Defaults to 0.5.

0.5
prefix str

Prefix string. Defaults to 'Progress'.

'Progress'
suffix str

Suffix string. Defaults to 'Complete'.

'Complete'
decimals int

Positive number of decimals in process percent. Defaults to 1.

1
length int

Character length of the status bar. Defaults to 40.

40
fill str

Bar fill character. Defaults to '█'.

'█'
Source code in musicalgestures/_utils.py
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
def __init__(
        self,
        total: int = 100,
        time_limit: float = 0.5,
        prefix: str = 'Progress',
        suffix: str = 'Complete',
        decimals: int = 1,
        length: int = 40,
        fill: str = '█'):
    """
    Initialize the MgProgressbar object.

    Args:
        total (int, optional): Total iterations. Defaults to 100.
        time_limit (float, optional): The minimum refresh rate of the progressbar in seconds. Defaults to 0.5.
        prefix (str, optional): Prefix string. Defaults to 'Progress'.
        suffix (str, optional): Suffix string. Defaults to 'Complete'.
        decimals (int, optional): Positive number of decimals in process percent. Defaults to 1.
        length (int, optional): Character length of the status bar. Defaults to 40.
        fill (str, optional): Bar fill character. Defaults to '█'.
    """

    self.total = total - 1
    self.time_limit = time_limit
    self.prefix = prefix
    self.suffix = suffix
    self.decimals = decimals
    self.length = length
    self.fill = fill
    self.now = self.get_now()
    self.finished = False
    self.could_not_get_terminal_window = False
    self.tw_width = 0
    self.tw_height = 0
    self.display_only_percent = False

get_now

get_now()

Gets the current time.

Returns:

Type Description

datetime.datetime.timestamp: The current time.

Source code in musicalgestures/_utils.py
77
78
79
80
81
82
83
84
85
def get_now(self):
    """
    Gets the current time.

    Returns:
        datetime.datetime.timestamp: The current time.
    """
    from datetime import datetime
    return datetime.timestamp(datetime.now())

over_time_limit

over_time_limit()

Checks if we should redraw the progress bar at this moment.

Returns:

Name Type Description
bool bool

True if equal or more time has passed than self.time_limit since the last redraw.

Source code in musicalgestures/_utils.py
87
88
89
90
91
92
93
94
95
def over_time_limit(self) -> bool:
    """
    Checks if we should redraw the progress bar at this moment.

    Returns:
        bool: True if equal or more time has passed than `self.time_limit` since the last redraw.
    """
    callback_time = self.get_now()
    return callback_time - self.now >= self.time_limit

progress

progress(iteration)

Progresses the progress bar to the next step.

Parameters:

Name Type Description Default
iteration float

The current iteration. For example, the 57th out of 100 steps, or 12.3s out of the total 60s.

required
Source code in musicalgestures/_utils.py
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
def progress(self, iteration: float) -> None:
    """
    Progresses the progress bar to the next step.

    Args:
        iteration (float): The current iteration. For example, the 57th out of 100 steps, or 12.3s out of the total 60s.
    """
    if self.finished:
        return
    if not _SHOW_PROGRESS:
        if iteration >= self.total:
            self.finished = True
        return
    import sys
    import shutil

    if not self.could_not_get_terminal_window:
        self.tw_width, self.tw_height = shutil.get_terminal_size((0, 0))
        if self.tw_width + self.tw_height == 0:
            self.could_not_get_terminal_window = True
        else:
            self.adjust_printlength()  # this line cannot be tested :'(

    capped_iteration = iteration if iteration <= self.total else self.total
    # Print New Line on Complete
    if iteration >= self.total:
        self.finished = True
        percent = ("{0:." + str(self.decimals) + "f}").format(100)
        filledLength = int(round(self.length))
        bar = self.fill * filledLength
        sys.stdout.flush()
        if self.display_only_percent:
            sys.stdout.write('\r%s' % (percent))
        else:
            sys.stdout.write('\r%s |%s| %s%% %s' %
                             (self.prefix, bar, percent, self.suffix))
        print()
    elif self.over_time_limit():
        self.now = self.get_now()
        percent = ("{0:." + str(self.decimals) + "f}").format(100 *
                                                              (capped_iteration / float(self.total)))
        filledLength = int(self.length * capped_iteration // self.total)
        bar = self.fill * filledLength + '-' * (self.length - filledLength)
        sys.stdout.flush()
        if self.display_only_percent:
            sys.stdout.write('\r%s' % (percent))
        else:
            sys.stdout.write('\r%s |%s| %s%% %s' %
                             (self.prefix, bar, percent, self.suffix))
    else:
        return

MgImage

MgImage(filename)

Class for handling images in the Musical Gestures Toolbox.

Initializes the MgImage object.

Parameters:

Name Type Description Default
filename str

The path to the image file to load.

required
Source code in musicalgestures/_utils.py
219
220
221
222
223
224
225
226
227
228
229
def __init__(self, filename: str):
    """
    Initializes the MgImage object.

    Args:
        filename (str): The path to the image file to load.
    """
    self.filename = filename
    import os
    self.of = os.path.splitext(self.filename)[0]
    self.fex = os.path.splitext(self.filename)[1]

save

save(target_name)

Save (copy) the image to target_name and return a new MgImage pointing to it.

Source code in musicalgestures/_utils.py
235
236
237
238
239
240
241
242
def save(self, target_name: str) -> "MgImage":
    """Save (copy) the image to ``target_name`` and return a new MgImage pointing to it."""
    import os
    import shutil
    target_name = os.path.splitext(target_name)[0] + (self.fex or '.png')
    if os.path.abspath(target_name) != os.path.abspath(self.filename):
        shutil.copyfile(self.filename, target_name)
    return MgImage(target_name)

to_html

to_html()

Return an HTML snippet embedding the image (base64-encoded).

NB: This is intentionally not exposed as _repr_html_, so an MgImage is not auto-rendered as the last expression of a Jupyter cell. Use show() to display the image.

Source code in musicalgestures/_utils.py
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
def to_html(self) -> str:
    """
    Return an HTML snippet embedding the image (base64-encoded).

    NB: This is intentionally **not** exposed as ``_repr_html_``, so an MgImage
    is not auto-rendered as the last expression of a Jupyter cell. Use ``show()``
    to display the image.
    """
    import base64
    import os
    if not os.path.exists(self.filename):
        return f"<i>MgImage('{self.filename}') – file not found</i>"
    ext = self.fex.lower().lstrip('.')
    mime = {'jpg': 'jpeg', 'jpeg': 'jpeg', 'png': 'png', 'gif': 'gif'}.get(ext, 'png')
    with open(self.filename, 'rb') as f:
        b64 = base64.b64encode(f.read()).decode('ascii')
    return (
        f'<div style="display:inline-block;text-align:center;">'
        f'<img src="data:image/{mime};base64,{b64}" '
        f'style="max-width:600px;max-height:400px;" />'
        f'<br><small><code>{self.filename}</code></small></div>'
    )

MgFigure

MgFigure(figure=None, figure_type=None, data=None, layers=None, image=None)

Class for working with figures and plots within the Musical Gestures Toolbox.

Initializes the MgFigure object.

Parameters:

Name Type Description Default
figure figure

The internal figure. Defaults to None.

None
figure_type str

A keyword describing the type of the figure, such as "audio.spectrogram", "audio.tempogram", "audio.descriptors", "layers", etc. Defaults to None.

None
data dictionary

The dictionary containing all the necessary variables, lists and (typically) NumPy arrays necessary to rebuild each subplot in the figure. Defaults to None.

None
layers list

This is only relevant if the MgFigure instance is of "layers" type, which indicates that it is a composit of several MgFigures and/or MgImages. In this case the layers list should contain all the child instances (MgFigures, MgImages, or MgLists of these) which are included in this MgFigure and are displayed as subplots. Defaults to None.

None
image str

Path to the image file (the rendered figure). Defaults to None.

None
Source code in musicalgestures/_utils.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
def __init__(self, figure=None, figure_type: str = None, data: dict = None,
             layers: list = None, image=None):
    """
    Initializes the MgFigure object.

    Args:
        figure (matplotlib.pyplot.figure, optional): The internal figure. Defaults to None.
        figure_type (str, optional): A keyword describing the type of the figure, such as "audio.spectrogram", "audio.tempogram", "audio.descriptors", "layers", etc. Defaults to None.
        data (dictionary, optional): The dictionary containing all the necessary variables, lists and (typically) NumPy arrays necessary to rebuild each subplot in the figure. Defaults to None.
        layers (list, optional): This is only relevant if the MgFigure instance is of "layers" type, which indicates that it is a composit of several MgFigures and/or MgImages. In this case the layers list should contain all the child instances (MgFigures, MgImages, or MgLists of these) which are included in this MgFigure and are displayed as subplots. Defaults to None.
        image (str, optional): Path to the image file (the rendered figure). Defaults to None.
    """
    self.figure = figure
    self.figure_type = figure_type
    self.data = data
    self.layers = layers
    self.image = image

save

save(target_name)

Save the rendered figure to target_name.

Copies the rendered PNG if one exists, otherwise re-saves the internal matplotlib figure. Returns an MgImage pointing to the saved file (or None if nothing to save).

Source code in musicalgestures/_utils.py
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
def save(self, target_name: str):
    """Save the rendered figure to ``target_name``.

    Copies the rendered PNG if one exists, otherwise re-saves the internal matplotlib
    figure. Returns an MgImage pointing to the saved file (or None if nothing to save).
    """
    import os
    import shutil
    target_name = os.path.splitext(target_name)[0] + '.png'
    if isinstance(self.image, str) and os.path.exists(self.image):
        if os.path.abspath(target_name) != os.path.abspath(self.image):
            shutil.copyfile(self.image, target_name)
        return MgImage(target_name)
    if self.figure is not None:
        self.figure.savefig(target_name)
        return MgImage(target_name)
    return None

show

show(**kwargs)

Display the rendered figure.

In a Jupyter notebook the saved image is shown inline; otherwise it is opened in a viewer window. Additional keyword arguments are forwarded to the underlying MgImage.show(). If no rendered image is available, the internal matplotlib figure is returned instead.

Source code in musicalgestures/_utils.py
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
def show(self, **kwargs):
    """
    Display the rendered figure.

    In a Jupyter notebook the saved image is shown inline; otherwise it is opened
    in a viewer window. Additional keyword arguments are forwarded to the underlying
    MgImage.show(). If no rendered image is available, the internal matplotlib figure
    is returned instead.
    """
    import os
    if isinstance(self.image, (list, tuple)):
        for img in self.image:
            MgImage(img).show(**kwargs)
        return self
    if self.image and isinstance(self.image, str) and os.path.exists(self.image):
        return MgImage(self.image).show(**kwargs)
    return self.figure

to_html

to_html()

Return an HTML snippet embedding the rendered figure (base64-encoded).

NB: This is intentionally not exposed as _repr_html_, so an MgFigure is not auto-rendered as the last expression of a Jupyter cell. Use show() to display the figure.

Source code in musicalgestures/_utils.py
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
def to_html(self) -> str:
    """
    Return an HTML snippet embedding the rendered figure (base64-encoded).

    NB: This is intentionally **not** exposed as ``_repr_html_``, so an MgFigure
    is not auto-rendered as the last expression of a Jupyter cell. Use ``show()``
    to display the figure.
    """
    import base64
    import io
    import os
    if self.image and isinstance(self.image, str) and os.path.exists(self.image):
        with open(self.image, 'rb') as f:
            b64 = base64.b64encode(f.read()).decode('ascii')
        return (
            f'<div style="display:inline-block;text-align:center;">'
            f'<img src="data:image/png;base64,{b64}" '
            f'style="max-width:800px;max-height:600px;" />'
            f'<br><small><code>MgFigure(type={self.figure_type!r})</code></small></div>'
        )
    elif self.figure is not None:
        buf = io.BytesIO()
        self.figure.savefig(buf, format='png', bbox_inches='tight')
        buf.seek(0)
        b64 = base64.b64encode(buf.read()).decode('ascii')
        return (
            f'<div style="display:inline-block;text-align:center;">'
            f'<img src="data:image/png;base64,{b64}" '
            f'style="max-width:800px;max-height:600px;" />'
            f'<br><small><code>MgFigure(type={self.figure_type!r})</code></small></div>'
        )
    return f"<i>MgFigure(type={self.figure_type!r}) – no image available</i>"

show_progress

show_progress(enabled)

Enable or disable the MGT progress bars globally.

Disabling the progress bars is useful when running batch processing jobs or when the output is captured by a logging framework where the repeated \r updates would clutter the log.

Parameters:

Name Type Description Default
enabled bool

Pass True to show progress bars (default behaviour) or False to suppress them.

required

Examples:

>>> import musicalgestures as mg
>>> mg.show_progress(False)  # suppress all progress bars
>>> # … batch processing …
>>> mg.show_progress(True)   # re-enable for interactive use
Source code in musicalgestures/_utils.py
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
def show_progress(enabled: bool) -> None:
    """Enable or disable the MGT progress bars globally.

    Disabling the progress bars is useful when running batch processing jobs or
    when the output is captured by a logging framework where the repeated
    ``\\r`` updates would clutter the log.

    Args:
        enabled (bool): Pass ``True`` to show progress bars (default behaviour)
            or ``False`` to suppress them.

    Examples:
        >>> import musicalgestures as mg
        >>> mg.show_progress(False)  # suppress all progress bars
        >>> # … batch processing …
        >>> mg.show_progress(True)   # re-enable for interactive use
    """
    global _SHOW_PROGRESS
    _SHOW_PROGRESS = bool(enabled)

roundup

roundup(num, modulo_num)

Rounds up a number to the next integer multiple of another.

Parameters:

Name Type Description Default
num int

The number to round up.

required
modulo_num int

The number whose next integer multiple we want.

required

Returns:

Name Type Description
int int

The rounded-up number.

Source code in musicalgestures/_utils.py
363
364
365
366
367
368
369
370
371
372
373
374
375
def roundup(num: int, modulo_num: int) -> int:
    """
    Rounds up a number to the next integer multiple of another.

    Args:
        num (int): The number to round up.
        modulo_num (int): The number whose next integer multiple we want.

    Returns:
        int: The rounded-up number.
    """
    num, modulo_num = int(num), int(modulo_num)
    return num - (num % modulo_num) + modulo_num*((num % modulo_num) != 0)

clamp

clamp(num, min_value, max_value)

Clamps a number between a minimum and maximum value.

Parameters:

Name Type Description Default
num float

The number to clamp.

required
min_value float

The minimum allowed value.

required
max_value float

The maximum allowed value.

required

Returns:

Name Type Description
float float

The clamped number.

Source code in musicalgestures/_utils.py
378
379
380
381
382
383
384
385
386
387
388
389
390
def clamp(num: float, min_value: float, max_value: float) -> float:
    """
    Clamps a number between a minimum and maximum value.

    Args:
        num (float): The number to clamp.
        min_value (float): The minimum allowed value.
        max_value (float): The maximum allowed value.

    Returns:
        float: The clamped number.
    """
    return max(min(num, max_value), min_value)

scale_num

scale_num(val, in_low, in_high, out_low, out_high)

Scales a number linearly.

Parameters:

Name Type Description Default
val float

The value to be scaled.

required
in_low float

Minimum of input range.

required
in_high float

Maximum of input range.

required
out_low float

Minimum of output range.

required
out_high float

Maximum of output range.

required

Returns:

Name Type Description
float float

The scaled number.

Source code in musicalgestures/_utils.py
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def scale_num(val: float, in_low: float, in_high: float, out_low: float, out_high: float) -> float:
    """
    Scales a number linearly.

    Args:
        val (float): The value to be scaled.
        in_low (float): Minimum of input range.
        in_high (float): Maximum of input range.
        out_low (float): Minimum of output range.
        out_high (float): Maximum of output range.

    Returns:
        float: The scaled number.
    """

    return ((val - in_low) * (out_high - out_low)) / (in_high - in_low) + out_low

scale_array

scale_array(array, out_low, out_high)

Scales an array linearly.

Parameters:

Name Type Description Default
array arraylike

The array to be scaled.

required
out_low float

Minimum of output range.

required
out_high float

Maximum of output range.

required

Returns:

Name Type Description
arraylike ndarray

The scaled array.

Source code in musicalgestures/_utils.py
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
def scale_array(array: np.ndarray, out_low: float, out_high: float) -> np.ndarray:
    """
    Scales an array linearly.

    Args:
        array (arraylike): The array to be scaled.
        out_low (float): Minimum of output range.
        out_high (float): Maximum of output range.

    Returns:
        arraylike: The scaled array.
    """

    import numpy as np
    minimum, maximum = np.min(array), np.max(array)
    m = (out_high - out_low) / (maximum - minimum)
    b = out_low - m * minimum
    return m * array + b

generate_outfilename

generate_outfilename(requested_name)

Returns a unique filepath to avoid overwriting existing files. Increments requested filename if necessary by appending an integer, like "_0" or "_1", etc to the file name.

Parameters:

Name Type Description Default
requested_name str

Requested file name as path string.

required

Returns:

Name Type Description
str str

If file at requested_name is not present, then requested_name, else an incremented filename.

Source code in musicalgestures/_utils.py
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
def generate_outfilename(requested_name: str) -> str:
    """Returns a unique filepath to avoid overwriting existing files. Increments requested 
    filename if necessary by appending an integer, like "_0" or "_1", etc to the file name.

    Args:
        requested_name (str): Requested file name as path string.

    Returns:
        str: If file at requested_name is not present, then requested_name, else an incremented filename.
    """
    import os
    requested_name = os.path.abspath(requested_name).replace('\\', '/')
    req_of, req_fex = os.path.splitext(requested_name)
    req_of = req_of.replace('\\', '/')
    req_folder = os.path.dirname(requested_name).replace('\\', '/')
    req_of_base = os.path.basename(req_of)
    req_file_base = os.path.basename(requested_name)
    out_increment = 0
    files_in_folder = os.listdir(req_folder)
    # if the target folder is empty, return the requested path
    if len(files_in_folder) == 0:
        return requested_name
    # filter files with same ext
    files_w_same_ext = list(filter(lambda x: os.path.splitext(x)[
                            1] == req_fex, files_in_folder))
    # if there are no files with the same ext
    if len(files_w_same_ext) == 0:
        return requested_name
    # filter for files with same start and ext
    files_w_same_start_ext = list(
        filter(lambda x: x.startswith(req_of_base), files_w_same_ext))
    # if there are no files with the same start and ext
    if len(files_w_same_start_ext) == 0:
        return requested_name
    # check if requested file is already present
    present = None
    try:
        ind = files_w_same_start_ext.index(req_file_base)
        present = True
    except ValueError:
        present = False
    # if requested file is not present
    if not present:
        return requested_name
    # if the original filename is already taken, check if there are incremented filenames
    files_w_increment = list(filter(lambda x: x.startswith(
        req_of_base+"_"), files_w_same_start_ext))
    # if there are no files with increments
    if len(files_w_increment) == 0:
        return f'{req_of}_0{req_fex}'
    # parse increments, discard the ones that are invalid, increment highest
    for file in files_w_increment:
        _of = os.path.splitext(file)[0]
        _only_incr = _of[len(req_of_base)+1:]
        try:
            found_incr = int(_only_incr)
            found_incr = max(0, found_incr)  # clip at 0
            out_increment = max(out_increment, found_incr+1)
        except ValueError:  # if cannot be converted to int
            pass
    # return incremented filename
    return f'{req_of}_{out_increment}{req_fex}'

resolve_filename

resolve_filename(stem, suffix, target_name=None, overwrite=True)

Resolve an output filename for a rendered result.

Centralises the target_name/overwrite logic that most methods repeat: use stem + suffix when no name is given, otherwise honour the provided target_name but enforce the extension from suffix; when overwrite is False, auto-increment the name so nothing is clobbered.

Parameters:

Name Type Description Default
stem str

Input filename stem (e.g. self.of), used when target_name is None.

required
suffix str

Suffix incl. extension to append to stem (e.g. '_grid.png'); its extension is also the one enforced on a provided target_name.

required
target_name str

Explicit output path (its extension is normalised to the suffix extension). Defaults to None.

None
overwrite bool

If False, auto-increment to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The resolved output path.

Source code in musicalgestures/_utils.py
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
def resolve_filename(stem: str, suffix: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """Resolve an output filename for a rendered result.

    Centralises the ``target_name``/``overwrite`` logic that most methods repeat: use
    ``stem + suffix`` when no name is given, otherwise honour the provided ``target_name`` but
    enforce the extension from ``suffix``; when ``overwrite`` is False, auto-increment the name so
    nothing is clobbered.

    Args:
        stem (str): Input filename stem (e.g. ``self.of``), used when ``target_name`` is None.
        suffix (str): Suffix incl. extension to append to ``stem`` (e.g. ``'_grid.png'``); its
            extension is also the one enforced on a provided ``target_name``.
        target_name (str, optional): Explicit output path (its extension is normalised to the
            ``suffix`` extension). Defaults to None.
        overwrite (bool, optional): If False, auto-increment to avoid overwriting. Defaults to True.

    Returns:
        str: The resolved output path.
    """
    import os
    ext = os.path.splitext(suffix)[1]
    if target_name is None:
        target_name = stem + suffix
    else:
        target_name = os.path.splitext(target_name)[0] + ext
    if not overwrite:
        target_name = generate_outfilename(target_name)
    return target_name

get_frame_planecount

get_frame_planecount(frame)

Gets the planecount (color channel count) of a video frame.

Parameters:

Name Type Description Default
frame numpy array

A frame extracted by cv2.VideoCapture().read().

required

Returns:

Name Type Description
int int

The planecount of the input frame, 3 or 1.

Source code in musicalgestures/_utils.py
525
526
527
528
529
530
531
532
533
534
535
536
537
def get_frame_planecount(frame: np.ndarray) -> int:
    """
    Gets the planecount (color channel count) of a video frame.

    Args:
        frame (numpy array): A frame extracted by `cv2.VideoCapture().read()`.

    Returns:
        int: The planecount of the input frame, 3 or 1.
    """

    import numpy as np
    return 3 if len(np.array(frame).shape) == 3 else 1

frame2ms

frame2ms(frame, fps)

Converts frames to milliseconds.

Parameters:

Name Type Description Default
frame int

The index of the frame to be converted to milliseconds.

required
fps int

Frames per second.

required

Returns:

Name Type Description
int int

The rounded millisecond value of the input frame index.

Source code in musicalgestures/_utils.py
540
541
542
543
544
545
546
547
548
549
550
551
552
def frame2ms(frame: int, fps: int) -> int:
    """
    Converts frames to milliseconds.

    Args:
        frame (int): The index of the frame to be converted to milliseconds.
        fps (int): Frames per second.

    Returns:
        int: The rounded millisecond value of the input frame index.
    """

    return round(frame / fps * 1000)

pass_if_containers_match

pass_if_containers_match(file_1, file_2)

Checks if file extensions match between two files. If they do it passes, is they don't it raises WrongContainer exception.

Parameters:

Name Type Description Default
file_1 str

First file in comparison.

required
file_2 str

Second file in comparison.

required

Raises:

Type Description
WrongContainer

If file extensions (containers) mismatch.

Source code in musicalgestures/_utils.py
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
def pass_if_containers_match(file_1: str, file_2: str) -> None:
    """Checks if file extensions match between two files. If they do it passes, is they don't it raises WrongContainer exception.

    Args:
        file_1 (str): First file in comparison.
        file_2 (str): Second file in comparison.

    Raises:
        WrongContainer: If file extensions (containers) mismatch.
    """
    import os
    fex_1 = os.path.splitext(file_1)[1].lower()
    fex_2 = os.path.splitext(file_2)[1]. lower()
    if fex_1 != fex_2:
        raise WrongContainer(
            f"Container mismatch: {fex_1} vs {fex_2}; between {file_1} and {file_2}.")

pass_if_container_is

pass_if_container_is(container, file)

Checks if a file's extension matches a desired one. Passes if so, raises WrongContainer if not.

Parameters:

Name Type Description Default
container str

The container to match.

required
file str

Path to the file to inspect.

required

Raises:

Type Description
WrongContainer

If the file extension (container) matches the desired one.

Source code in musicalgestures/_utils.py
578
579
580
581
582
583
584
585
586
587
588
589
590
591
def pass_if_container_is(container: str, file: str) -> None:
    """Checks if a file's extension matches a desired one. Passes if so, raises WrongContainer if not.

    Args:
        container (str): The container to match.
        file (str): Path to the file to inspect.

    Raises:
        WrongContainer: If the file extension (container) matches the desired one.
    """
    import os
    if os.path.splitext(file)[1].lower() != container.lower():
        raise WrongContainer(
            f"Container should be {container.lower()}, but it is {os.path.splitext(file)[1].lower()} in file {file}.")

ffmpeg_has_encoder

ffmpeg_has_encoder(name)

Returns True if the installed FFmpeg has the named encoder (e.g. 'libtheora').

Useful for guarding format conversions whose codec may be missing from a given FFmpeg build (notably libtheora/libvorbis for .ogg on some macOS/Windows builds).

Source code in musicalgestures/_utils.py
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
def ffmpeg_has_encoder(name: str) -> bool:
    """
    Returns True if the installed FFmpeg has the named encoder (e.g. 'libtheora').

    Useful for guarding format conversions whose codec may be missing from a given
    FFmpeg build (notably libtheora/libvorbis for .ogg on some macOS/Windows builds).
    """
    import subprocess
    if name in _ENCODER_CACHE:
        return _ENCODER_CACHE[name]
    try:
        out = subprocess.run(['ffmpeg', '-hide_banner', '-encoders'],
                             stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
                             universal_newlines=True, timeout=15).stdout
        available = any(line.split()[1] == name for line in out.splitlines()
                        if len(line.split()) > 1)
    except Exception:
        available = False
    _ENCODER_CACHE[name] = available
    return available

convert

convert(filename, target_name, overwrite=True)

Converts a video to another format/container using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file to convert.

required
target_name str

Target filename as path.

required
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output file.

Source code in musicalgestures/_utils.py
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
def convert(filename: str, target_name: str, overwrite: bool = True) -> str:
    """
    Converts a video to another format/container using ffmpeg.

    Args:
        filename (str): Path to the input video file to convert.
        target_name (str): Target filename as path.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output file.
    """

    import os
    of, fex = os.path.splitext(filename)
    target_of, target_fex = os.path.splitext(target_name)
    if fex.lower() == target_fex.lower():
        print(f'{filename} is already in {fex} container.')
        return filename
    if not overwrite:
        target_name = generate_outfilename(target_name)
    # OGG video requires explicit Theora/Vorbis codecs; FFmpeg won't auto-select them
    if target_fex.lower() == '.ogg':
        if not ffmpeg_has_encoder('libtheora'):
            raise FFmpegError(
                "Converting to .ogg requires the 'libtheora' encoder, which this FFmpeg "
                "build does not include. Install an FFmpeg with libtheora/libvorbis, or "
                "use a different output format.")
        cmds = ['ffmpeg', '-y', '-i', filename,
                '-c:v', 'libtheora', '-c:a', 'libvorbis', '-q:v', '5', target_name]
    else:
        cmds = ['ffmpeg', '-y', '-i', filename,
                '-q:v', '3', target_name]
    ffmpeg_cmd(cmds, get_length(filename),
               pb_prefix=f'Converting to {target_fex}:')
    return target_name

convert_to_avi

convert_to_avi(filename, target_name=None, overwrite=True)

Converts a video to one with .avi extension using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file to convert.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output '.avi' file.

Source code in musicalgestures/_utils.py
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
def convert_to_avi(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Converts a video to one with .avi extension using ffmpeg.

    Args:
        filename (str): Path to the input video file to convert.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output '.avi' file.
    """

    import os
    of, fex = os.path.splitext(filename)
    if fex.lower() == '.avi':
        print(f'{filename} is already in avi container.')
        return filename
    if not target_name:
        target_name = of + '.avi'
    if not overwrite:
        target_name = generate_outfilename(target_name)
    pass_if_container_is(".avi", target_name)
    cmds = ['ffmpeg', '-y', '-i', filename, "-c:v", "mjpeg",
            "-q:v", "3", "-c:a", "copy", target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Converting to avi:')
    return target_name

convert_to_mp4

convert_to_mp4(filename, target_name=None, overwrite=True)

Converts a video to one with .mp4 extension using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file to convert.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output '.mp4' file.

Source code in musicalgestures/_utils.py
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
def convert_to_mp4(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Converts a video to one with .mp4 extension using ffmpeg.

    Args:
        filename (str): Path to the input video file to convert.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output '.mp4' file.
    """

    import os
    of, fex = os.path.splitext(filename)
    if fex.lower() == '.mp4':
        print(f'{filename} is already in mp4 container.')
        return filename
    if not target_name:
        target_name = of + '.mp4'
    if not overwrite:
        target_name = generate_outfilename(target_name)
    pass_if_container_is(".mp4", target_name)
    cmds = ['ffmpeg', '-y', '-i', filename,
            "-q:v", "3", target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Converting to mp4:')
    return target_name

convert_to_webm

convert_to_webm(filename, target_name=None, overwrite=True)

Converts a video to one with .webm extension using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file to convert.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output '.webm' file.

Source code in musicalgestures/_utils.py
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
def convert_to_webm(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Converts a video to one with .webm extension using ffmpeg.

    Args:
        filename (str): Path to the input video file to convert.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output '.webm' file.
    """

    import os
    of, fex = os.path.splitext(filename)
    if fex.lower() == '.webm':
        print(f'{filename} is already in webm container.')
        return filename
    if not target_name:
        target_name = of + '.webm'
    if not overwrite:
        target_name = generate_outfilename(target_name)
    pass_if_container_is(".webm", target_name)
    cmds = ['ffmpeg', '-y', '-i', filename,
            "-q:v", "3", target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Converting to webm:')
    return target_name

cast_into_avi

cast_into_avi(filename, target_name=None, overwrite=True)

Experimental Casts a video into and .avi container using ffmpeg. Much faster than convert_to_avi, but does not always work well with cv2 or built-in video players.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output '.avi' file.

Source code in musicalgestures/_utils.py
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
def cast_into_avi(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    *Experimental*
    Casts a video into and .avi container using ffmpeg. Much faster than `convert_to_avi`,
    but does not always work well with cv2 or built-in video players.

    Args:
        filename (str): Path to the input video file.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output '.avi' file.
    """

    import os
    of = os.path.splitext(filename)[0]
    if not target_name:
        target_name = of + '.avi'
    if not overwrite:
        target_name = generate_outfilename(target_name)
    pass_if_container_is(".avi", target_name)
    cmds = ['ffmpeg', '-y', '-i', filename, "-codec", "copy", target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Casting to avi')
    return target_name

extract_frame

extract_frame(filename, frame=None, time=None, target_name=None, overwrite=False)

Extracts a single frame from a video using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
frame int

The frame number to extract.

None
time Union[str, float]

The time in HH:MM:ss.ms where to extract the frame from. If float, it is interpreted as seconds from the start of the video.

None
target_name str

The name for the output file. If None, the name will be FRAME.. Defaults to None.

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

False
Source code in musicalgestures/_utils.py
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
def extract_frame(
    filename: str,
    frame: int=None,
    time: Union[str, float]=None,
    target_name: str=None,
    overwrite: bool=False,
    )-> str:
    """
    Extracts a single frame from a video using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        frame (int): The frame number to extract.
        time (Union[str, float]): The time in HH:MM:ss.ms where to extract the frame from. If float, it is interpreted as seconds from the start of the video.
        target_name (str, optional): The name for the output file. If None, the name will be <input name>FRAME<frame number>.<file extension>. Defaults to None.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.
    """

    import os
    import datetime
    if frame is not None and time is not None:
        raise ValueError("frame and time cannot be both not None.")
    if frame is None and time is None:
        raise ValueError("frame and time cannot be both None.")

    name, ext = os.path.splitext(filename)
    if not target_name:
        if frame is not None:
            target_name = f"{name}_frame_{str(frame)}.png"
        elif time is not None:
            time = time if isinstance(time, str) else datetime.datetime.fromtimestamp(time-3600).strftime('%H:%M:%S.%f')
            target_name = f"{name}_time_{time}.png"
    if not overwrite:
        target_name = generate_outfilename(target_name)

    if frame is not None:
        cmds = ['ffmpeg',
                '-y' if overwrite else "-n",
                '-i', filename,
                "-vf", rf"select='eq(n\,{frame})'",
                "-vsync", "0",
                # "-vframes", "1",
                target_name]
    elif time is not None:
        cmds = ['ffmpeg',
                '-y' if overwrite else "-n",
                '-i', filename,
                "-vf", rf"select='eq(t\,{time})'",
                "-vsync", "0",
                # "-vframes", "1",
                target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Extracting frame:')

    return target_name

extract_subclip

extract_subclip(filename, t1, t2, target_name=None, overwrite=True)

Extracts a section of the video using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
t1 float

The start of the section to extract in seconds.

required
t2 float

The end of the section to extract in seconds.

required
target_name str

The name for the output file. If None, the name will be SUB_.. Defaults to None.

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the extracted section as a video.

Source code in musicalgestures/_utils.py
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
def extract_subclip(filename: str, t1: float, t2: float, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Extracts a section of the video using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        t1 (float): The start of the section to extract in seconds.
        t2 (float): The end of the section to extract in seconds.
        target_name (str, optional): The name for the output file. If None, the name will be <input name>SUB<start time in ms>_<end time in ms>.<file extension>. Defaults to None.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the extracted section as a video.
    """

    import os
    import numpy as np
    name, ext = os.path.splitext(filename)
    length = get_length(filename)
    start, end = np.clip(t1, 0, length), np.clip(t2, 0, length)
    if start > end:
        # end = length
        start, end = end, start

    if not target_name:
        T1, T2 = [int(1000*t) for t in [start, end]]
        target_name = "%sSUB%d_%d.%s" % (name, T1, T2, ext)

    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    # avoiding ffmpeg glitch if format is not avi:
    if os.path.splitext(filename)[1] != '.avi':
        cmd = ['ffmpeg', "-y",
               "-ss", "%0.2f" % start,
               "-i", filename,
               "-t", "%0.2f" % (end-start),
               "-max_muxing_queue_size", "9999",
               "-map", "0", target_name]
    else:
        cmd = ['ffmpeg', "-y",
               "-ss", "%0.2f" % start,
               "-i", filename,
               "-t", "%0.2f" % (end-start),
               "-max_muxing_queue_size", "9999",
               "-map", "0", "-codec", "copy", target_name]

    ffmpeg_cmd(cmd, length, pb_prefix='Trimming:')
    return target_name

rotate_video

rotate_video(filename, angle, target_name=None, overwrite=True)

Rotates a video by an angle using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
angle float

The angle (in degrees) specifying the amount of rotation. Positive values rotate clockwise.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename with the suffix "_rot" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the rotated video file.

Source code in musicalgestures/_utils.py
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
def rotate_video(filename: str, angle: float, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Rotates a video by an `angle` using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        angle (float): The angle (in degrees) specifying the amount of rotation. Positive values rotate clockwise.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename with the suffix "_rot" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the rotated video file.
    """

    import os
    import math
    import numpy as np
    of, fex = os.path.splitext(filename)

    if not target_name:
        target_name = of + '_rot' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    if np.abs(angle) == 90 or np.abs(angle) == 180:
        # Rotate video without encoding for faster computation
        cmds = ['ffmpeg', '-y', '-i', filename, 
                '-metadata:s:v:0', f'rotate={angle}', '-codec', 'copy', target_name]
    else:
        # Rotate video with encoding
        cmds = ['ffmpeg', '-y', '-i', filename, "-vf", 
                f"rotate={math.radians(angle)}", "-q:v", "3", "-c:a", "copy", target_name]
    ffmpeg_cmd(cmds, get_length(filename),
               pb_prefix=f"Rotating video by {angle} degrees:")
    return target_name

convert_to_grayscale

convert_to_grayscale(filename, target_name=None, overwrite=True)

Converts a video to grayscale using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
target_name str

Target filename as path. Defaults to None (which assumes that the input filename with the suffix "_gray" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the grayscale video file.

Source code in musicalgestures/_utils.py
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
def convert_to_grayscale(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Converts a video to grayscale using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        target_name (str, optional): Target filename as path. Defaults to None (which assumes that the input filename with the suffix "_gray" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the grayscale video file.
    """

    import os
    of, fex = os.path.splitext(filename)

    if not target_name:
        target_name = of + '_gray' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    cmds = ['ffmpeg', '-y', '-i', filename, '-vf',
            'hue=s=0', "-q:v", "3", "-c:a", "copy", target_name]
    ffmpeg_cmd(cmds, get_length(filename), pb_prefix='Converting to grayscale:')
    return target_name

framediff_ffmpeg

framediff_ffmpeg(filename, target_name=None, color=True, overwrite=True)

Renders a frame difference video from the input using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
target_name str

The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_framediff" should be used).

None
color bool

If False, the output will be grayscale. Defaults to True.

True
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the output video.

Source code in musicalgestures/_utils.py
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
def framediff_ffmpeg(filename: str, target_name: str | None = None, color: bool = True, overwrite: bool = True) -> str:
    """
    Renders a frame difference video from the input using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_framediff" should be used).
        color (bool, optional): If False, the output will be grayscale. Defaults to True.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output video.
    """

    import os
    of, fex = os.path.splitext(filename)

    if target_name is None:
        target_name = of + '_framediff' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)
    pass_if_containers_match(filename, target_name)
    if color == True:
        pixformat = 'gbrp'
    else:
        pixformat = 'gray'
    cmd = ['ffmpeg', '-y', '-i', filename, '-filter_complex',
           f'format={pixformat},tblend=all_mode=difference', '-q:v', '3', "-c:a", "copy", target_name]
    ffmpeg_cmd(cmd, get_length(filename),
               pb_prefix='Rendering frame difference video:')
    return target_name

threshold_ffmpeg

threshold_ffmpeg(filename, threshold=0.1, target_name=None, binary=False, overwrite=True)

Renders a pixel-thresholded video from the input using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
threshold float

The normalized pixel value to use as the threshold. Pixels below the threshold will turn black. Defaults to 0.1.

0.1
target_name str

The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_thresh" should be used).

None
binary bool

If True, the pixels above the threshold will turn white. Defaults to False.

False
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the output video.

Source code in musicalgestures/_utils.py
 979
 980
 981
 982
 983
 984
 985
 986
 987
 988
 989
 990
 991
 992
 993
 994
 995
 996
 997
 998
 999
1000
1001
1002
1003
1004
1005
1006
1007
1008
1009
1010
1011
1012
1013
1014
1015
1016
1017
1018
1019
1020
def threshold_ffmpeg(filename: str, threshold: float = 0.1, target_name: str | None = None, binary: bool = False, overwrite: bool = True) -> str:
    """
    Renders a pixel-thresholded video from the input using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        threshold (float, optional): The normalized pixel value to use as the threshold. Pixels below the threshold will turn black. Defaults to 0.1.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_thresh" should be used).
        binary (bool, optional): If True, the pixels above the threshold will turn white. Defaults to False.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output video.
    """

    import os
    import matplotlib
    of, fex = os.path.splitext(filename)

    if target_name is None:
        target_name = of + '_thresh' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    width, height = get_widthheight(filename)

    thresh_color = matplotlib.colors.to_hex([threshold, threshold, threshold])
    thresh_color = '0x' + thresh_color[1:]

    if binary == False:
        cmd = ['ffmpeg', '-y', '-i', filename, '-f', 'lavfi', '-i', f'color={thresh_color},scale={width}:{height}', '-f', 'lavfi',
               '-i', f'color=black,scale={width}:{height}', '-i', filename, '-lavfi', 'format=gbrp,threshold', '-q:v', '3', "-c:a", "copy", target_name]
    else:
        cmd = ['ffmpeg', '-y', '-i', filename, '-f', 'lavfi', '-i', f'color={thresh_color},scale={width}:{height}', '-f', 'lavfi',
               '-i', f'color=black,scale={width}:{height}', '-f', 'lavfi', '-i', f'color=white,scale={width}:{height}', '-lavfi', 'format=gray,threshold', '-q:v', '3', "-c:a", "copy", target_name]

    ffmpeg_cmd(cmd, get_length(filename),
               pb_prefix='Rendering threshold video:')

    return target_name

motionvideo_ffmpeg

motionvideo_ffmpeg(filename, color=True, filtertype='regular', threshold=0.05, blur='none', use_median=False, kernel_size=5, invert=False, target_name=None, overwrite=True)

Renders a motion video using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
color bool

If False the input is converted to grayscale at the start of the process. This can significantly reduce render time. Defaults to True.

True
filtertype str

'Regular' turns all values below threshold to 0. 'Binary' turns all values below threshold to 0, above threshold to 1. 'Blob' removes individual pixels with erosion method. Defaults to 'Regular'.

'regular'
threshold float

Eliminates pixel values less than given threshold. Ranges from 0 to 1. Defaults to 0.05.

0.05
blur str

'Average' to apply a 10px * 10px blurring filter, 'None' otherwise. Defaults to 'None'.

'none'
use_median bool

If True the algorithm applies a median filter on the thresholded frame-difference stream. Defaults to False.

False
kernel_size int

Size of the median filter (if use_median=True) or the erosion filter (if filtertype='blob'). Defaults to 5.

5
invert bool

If True, inverts colors of the motion video. Defaults to False.

False
target_name str

Defaults to None (which assumes that the input filename with the suffix "_motion" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the output video.

Source code in musicalgestures/_utils.py
1023
1024
1025
1026
1027
1028
1029
1030
1031
1032
1033
1034
1035
1036
1037
1038
1039
1040
1041
1042
1043
1044
1045
1046
1047
1048
1049
1050
1051
1052
1053
1054
1055
1056
1057
1058
1059
1060
1061
1062
1063
1064
1065
1066
1067
1068
1069
1070
1071
1072
1073
1074
1075
1076
def motionvideo_ffmpeg(
        filename: str,
        color: bool = True,
        filtertype: str = 'regular',
        threshold: float = 0.05,
        blur: str = 'none',
        use_median: bool = False,
        kernel_size: int = 5,
        invert: bool = False,
        target_name: str | None = None,
        overwrite: bool = True) -> str:
    """
    Renders a motion video using ffmpeg. 

    Args:
        filename (str): Path to the input video file.
        color (bool, optional): If False the input is converted to grayscale at the start of the process. This can significantly reduce render time. Defaults to True.
        filtertype (str, optional): 'Regular' turns all values below `threshold` to 0. 'Binary' turns all values below `threshold` to 0, above `threshold` to 1. 'Blob' removes individual pixels with erosion method. Defaults to 'Regular'.
        threshold (float, optional): Eliminates pixel values less than given threshold. Ranges from 0 to 1. Defaults to 0.05.
        blur (str, optional): 'Average' to apply a 10px * 10px blurring filter, 'None' otherwise. Defaults to 'None'.
        use_median (bool, optional): If True the algorithm applies a median filter on the thresholded frame-difference stream. Defaults to False.
        kernel_size (int, optional): Size of the median filter (if `use_median=True`) or the erosion filter (if `filtertype='blob'`). Defaults to 5.
        invert (bool, optional): If True, inverts colors of the motion video. Defaults to False.
        target_name (str, optional): Defaults to None (which assumes that the input filename with the suffix "_motion" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output video.
    """

    import os
    from musicalgestures._filter import filter_frame_ffmpeg

    of, fex = os.path.splitext(filename)

    cmd = ['ffmpeg', '-y', '-i', filename]
    # cmd_filter = ''

    if target_name is None:
        target_name = of + '_motion' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    cmd, cmd_filter = filter_frame_ffmpeg(filename, cmd, color, blur, filtertype, threshold, kernel_size, use_median, invert=invert)
    # remove last comma after previous filter
    cmd_filter = cmd_filter[:-1]

    pass_if_containers_match(filename, target_name)
    cmd_end = ['-q:v', '3', "-c:a", "copy", target_name]
    cmd += ['-filter_complex', cmd_filter] + cmd_end

    ffmpeg_cmd(cmd, get_length(filename), pb_prefix='Rendering motion video:')

    return target_name

motiongrams_ffmpeg

motiongrams_ffmpeg(filename, color=True, filtertype='regular', threshold=0.05, blur='none', use_median=False, kernel_size=5, invert=False, target_name_x=None, target_name_y=None, overwrite=True)

Renders horizontal and vertical motiongrams using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
color bool

If False the input is converted to grayscale at the start of the process. This can significantly reduce render time. Defaults to True.

True
filtertype str

'Regular' turns all values below threshold to 0. 'Binary' turns all values below threshold to 0, above threshold to 1. 'Blob' removes individual pixels with erosion method. Defaults to 'Regular'.

'regular'
threshold float

Eliminates pixel values less than given threshold. Ranges from 0 to 1. Defaults to 0.05.

0.05
blur str

'Average' to apply a 10px * 10px blurring filter, 'None' otherwise. Defaults to 'None'.

'none'
use_median bool

If True the algorithm applies a median filter on the thresholded frame-difference stream. Defaults to False.

False
kernel_size int

Size of the median filter (if use_median=True) or the erosion filter (if filtertype='blob'). Defaults to 5.

5
invert bool

If True, inverts colors of the motiongrams. Defaults to False.

False
target_name_x str

Target output name for the motiongram on the X axis. Defaults to None (which assumes that the input filename with the suffix "_mgx_ffmpeg" should be used).

None
target_name_y str

Target output name for the motiongram on the Y axis. Defaults to None (which assumes that the input filename with the suffix "_mgy_ffmpeg" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filenames to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str tuple

Path to the output horizontal motiongram (_mgx).

str tuple

Path to the output vertical motiongram (_mgy).

Source code in musicalgestures/_utils.py
1079
1080
1081
1082
1083
1084
1085
1086
1087
1088
1089
1090
1091
1092
1093
1094
1095
1096
1097
1098
1099
1100
1101
1102
1103
1104
1105
1106
1107
1108
1109
1110
1111
1112
1113
1114
1115
1116
1117
1118
1119
1120
1121
1122
1123
1124
1125
1126
1127
1128
1129
1130
1131
1132
1133
1134
1135
1136
1137
1138
1139
1140
1141
1142
1143
1144
1145
1146
1147
1148
1149
1150
1151
1152
def motiongrams_ffmpeg(
        filename: str,
        color: bool = True,
        filtertype: str = 'regular',
        threshold: float = 0.05,
        blur: str = 'none',
        use_median: bool = False,
        kernel_size: int = 5,
        invert: bool = False,
        target_name_x: str | None = None,
        target_name_y: str | None = None,
        overwrite: bool = True) -> tuple:
    """
    Renders horizontal and vertical motiongrams using ffmpeg. 

    Args:
        filename (str): Path to the input video file.
        color (bool, optional): If False the input is converted to grayscale at the start of the process. This can significantly reduce render time. Defaults to True.
        filtertype (str, optional): 'Regular' turns all values below `threshold` to 0. 'Binary' turns all values below `threshold` to 0, above `threshold` to 1. 'Blob' removes individual pixels with erosion method. Defaults to 'Regular'.
        threshold (float, optional): Eliminates pixel values less than given threshold. Ranges from 0 to 1. Defaults to 0.05.
        blur (str, optional): 'Average' to apply a 10px * 10px blurring filter, 'None' otherwise. Defaults to 'None'.
        use_median (bool, optional): If True the algorithm applies a median filter on the thresholded frame-difference stream. Defaults to False.
        kernel_size (int, optional): Size of the median filter (if `use_median=True`) or the erosion filter (if `filtertype='blob'`). Defaults to 5.
        invert (bool, optional): If True, inverts colors of the motiongrams. Defaults to False.
        target_name_x (str, optional): Target output name for the motiongram on the X axis. Defaults to None (which assumes that the input filename with the suffix "_mgx_ffmpeg" should be used).
        target_name_y (str, optional): Target output name for the motiongram on the Y axis. Defaults to None (which assumes that the input filename with the suffix "_mgy_ffmpeg" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filenames to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output horizontal motiongram (_mgx).
        str: Path to the output vertical motiongram (_mgy).
    """

    import os
    from musicalgestures._filter import filter_frame_ffmpeg

    of, fex = os.path.splitext(filename)

    if target_name_x is None:
        target_name_x = of+'_mgx_ffmpeg.png'
    if target_name_y is None:
        target_name_y = of+'_mgy_ffmpeg.png'
    if not overwrite:
        target_name_x = generate_outfilename(target_name_x)
        target_name_y = generate_outfilename(target_name_y)

    pass_if_container_is(".png", target_name_x)
    pass_if_container_is(".png", target_name_y)

    cmd = ['ffmpeg', '-y', '-i', filename]

    width, height = get_widthheight(filename)
    framecount = get_framecount(filename)

    cmd_end_y = ['-aspect', f'{framecount}:{height}', '-frames', '1', target_name_y]
    cmd_end_x = ['-aspect', f'{width}:{framecount}', '-frames', '1', target_name_x]

    cmd, cmd_filter = filter_frame_ffmpeg(filename, cmd, color, blur, filtertype, threshold, kernel_size, use_median, invert=invert)
    cmd_filter += 'atadenoise=s=129,' # apply adaptive temporal averaging denoiser every 129 frames

    cmd_filter_y = cmd_filter + \
        f'scale=1:{height},tile={framecount}x1,normalize=independence=0'
    # f'scale=1:{height}:sws_flags=area,normalize,tile={framecount}x1'
    cmd_filter_x = cmd_filter + \
        f'scale={width}:1,tile=1x{framecount},normalize=independence=0'
    # f'scale={width}:1:sws_flags=area,normalize,tile=1x{framecount}'

    cmd_y = cmd + ['-filter_complex', cmd_filter_y] + cmd_end_y
    cmd_x = cmd + ['-filter_complex', cmd_filter_x] + cmd_end_x

    ffmpeg_cmd(cmd_x, get_length(filename), pb_prefix='Rendering horizontal motiongram:', stream=False)
    ffmpeg_cmd(cmd_y, get_length(filename), pb_prefix='Rendering vertical motiongram:', stream=False)

    return target_name_x, target_name_y

crop_ffmpeg

crop_ffmpeg(filename, w, h, x, y, target_name=None, overwrite=True)

Crops a video using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
w int

The desired width.

required
h int

The desired height.

required
x int

The horizontal coordinate of the top left pixel of the cropping rectangle.

required
y int

The vertical coordinate of the top left pixel of the cropping rectangle.

required
target_name str

The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_crop" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filenames to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the output video.

Source code in musicalgestures/_utils.py
1155
1156
1157
1158
1159
1160
1161
1162
1163
1164
1165
1166
1167
1168
1169
1170
1171
1172
1173
1174
1175
1176
1177
1178
1179
1180
1181
1182
1183
1184
1185
1186
1187
1188
def crop_ffmpeg(filename: str, w: int, h: int, x: int, y: int, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Crops a video using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        w (int): The desired width.
        h (int): The desired height.
        x (int): The horizontal coordinate of the top left pixel of the cropping rectangle.
        y (int): The vertical coordinate of the top left pixel of the cropping rectangle.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_crop" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filenames to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output video.
    """

    import os

    of, fex = os.path.splitext(filename)

    if target_name is None:
        target_name = of + '_crop' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    cmd = ['ffmpeg', '-y', '-i', filename, '-vf',
           f'crop={w}:{h}:{x}:{y}', '-q:v', '3', "-c:a", "copy", target_name]

    ffmpeg_cmd(cmd, get_length(filename), pb_prefix='Rendering cropped video:')

    return target_name

extract_wav

extract_wav(filename, target_name=None, overwrite=True)

Extracts audio from video into a .wav file via ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the video file from which the audio track shall be extracted.

required
target_name str

The name of the output video. Defaults to None (which assumes that the input filename should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output audio file.

Source code in musicalgestures/_utils.py
1191
1192
1193
1194
1195
1196
1197
1198
1199
1200
1201
1202
1203
1204
1205
1206
1207
1208
1209
1210
1211
1212
1213
1214
1215
1216
1217
1218
1219
1220
def extract_wav(filename: str, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Extracts audio from video into a .wav file via ffmpeg.

    Args:
        filename (str): Path to the video file from which the audio track shall be extracted.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output audio file.
    """

    import os
    of, fex = os.path.splitext(filename)

    if target_name is None:
        target_name = of + '.wav'
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_container_is(".wav", target_name)

    if fex in ['.wav', '.WAV']:
        print(f'{filename} is already in .wav container.')
        return filename

    cmds = ' '.join(['ffmpeg', '-loglevel', 'quiet', '-y', '-i', wrap_str(filename), "-acodec", "pcm_s16le", wrap_str(target_name)])
    os.system(cmds)
    return target_name

ffprobe

ffprobe(filename)

Returns info about video/audio file using FFprobe.

The result is cached per file (keyed by path + modification time + size), so repeated probes of an unchanged file don't spawn a new subprocess each time.

Parameters:

Name Type Description Default
filename str

Path to the video file to measure.

required

Returns:

Name Type Description
str str

decoded FFprobe output (stdout) as one string.

Source code in musicalgestures/_utils.py
1241
1242
1243
1244
1245
1246
1247
1248
1249
1250
1251
1252
1253
1254
1255
1256
1257
1258
1259
1260
1261
1262
1263
1264
1265
1266
1267
1268
1269
1270
1271
1272
1273
1274
1275
1276
1277
1278
1279
1280
1281
1282
1283
def ffprobe(filename: str) -> str:
    """
    Returns info about video/audio file using FFprobe.

    The result is cached per file (keyed by path + modification time + size), so
    repeated probes of an unchanged file don't spawn a new subprocess each time.

    Args:
        filename (str): Path to the video file to measure.

    Returns:
        str: decoded FFprobe output (stdout) as one string.
    """
    import subprocess
    import os

    cache_key = None
    try:
        st = os.stat(filename)
        cache_key = (filename, st.st_mtime, st.st_size)
        cached = _FFPROBE_CACHE.get(cache_key)
        if cached is not None:
            return cached
    except OSError:
        pass  # file not found yet — fall through to ffprobe, which raises a clear error

    command = ['ffprobe', filename]
    process = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
    try:
        out, err = process.communicate(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill()
        out, err = process.communicate()

    if err:
        raise FFprobeError(err)
    else:
        if out.splitlines()[-1].find("No such file or directory") != -1:
            raise FileNotFoundError(out.splitlines()[-1])
        if cache_key is not None:
            _FFPROBE_CACHE[cache_key] = out
        return out

get_widthheight

get_widthheight(filename)

Gets the width and height of a video using FFprobe.

Parameters:

Name Type Description Default
filename str

Path to the video file to measure.

required

Returns:

Name Type Description
int int

The width of the input video file.

int int

The height of the input video file.

Source code in musicalgestures/_utils.py
1285
1286
1287
1288
1289
1290
1291
1292
1293
1294
1295
1296
1297
1298
1299
1300
1301
1302
1303
1304
1305
1306
1307
1308
1309
1310
1311
1312
1313
1314
1315
1316
1317
1318
1319
1320
1321
1322
1323
def get_widthheight(filename: str) -> Tuple[int, int]:
    """
    Gets the width and height of a video using FFprobe.

    Args:
        filename (str): Path to the video file to measure.

    Returns:
        int: The width of the input video file.
        int: The height of the input video file.
    """
    out = ffprobe(filename)
    out_array = out.splitlines()
    video_stream = None
    at_line = -1
    while video_stream is None:
        video_stream = out_array[at_line] if out_array[at_line].find("Video:") != -1 else None

        if out_array[at_line].find("displaymatrix:") != -1:
            import re
            rotation = [d for d in re.findall(r"\d+\.\d+", out_array[at_line])]

        at_line -= 1
        if at_line < -len(out_array):
            raise NoStreamError("No video stream found. (Is this a video file?)")

    try:
        if int(float(rotation[0])) == 90:
            # If the video has been rotated for 90°, we need to invert width and height
            width = int(video_stream.split('x')[-1].split(',')[0].split(' ')[0])
            height = int(video_stream.split('x')[-2].split(' ')[-1])
        else:
            width = int(video_stream.split('x')[-2].split(' ')[-1])
            height = int(video_stream.split('x')[-1].split(',')[0].split(' ')[0])
    except:
        width = int(video_stream.split('x')[-2].split(' ')[-1])
        height = int(video_stream.split('x')[-1].split(',')[0].split(' ')[0])

    return width, height

has_audio

has_audio(filename)

Checks if video has audio track using FFprobe.

Parameters:

Name Type Description Default
filename str

Path to the video file to check.

required

Returns:

Name Type Description
bool bool

True if filename has an audio track, False otherwise.

Source code in musicalgestures/_utils.py
1326
1327
1328
1329
1330
1331
1332
1333
1334
1335
1336
1337
1338
1339
1340
1341
1342
1343
1344
1345
1346
1347
1348
1349
def has_audio(filename: str) -> bool:
    """
    Checks if video has audio track using FFprobe.

    Args:
        filename (str): Path to the video file to check.

    Returns:
        bool: True if `filename` has an audio track, False otherwise.
    """
    out = ffprobe(filename)
    out_array = out.splitlines()
    audio_stream = None
    at_line = -1
    while audio_stream is None:
        audio_stream = out_array[at_line] if out_array[at_line].find(
            "Audio:") != -1 else None
        at_line -= 1
        if at_line < -len(out_array):
            break
    if audio_stream is None:
        return False
    else:
        return True

get_rotation

get_rotation(filename)

Returns the display rotation (degrees) stored in a video's metadata.

Phone/portrait videos often store landscape pixels plus a rotation flag (display matrix). FFmpeg's frame pipe applies this automatically while OpenCV's VideoCapture does not, which can leave some processes rotated. This reads the flag so the orientation can be normalised.

Returns:

Name Type Description
int int

rotation in degrees (e.g. 0, 90, 180, 270), or 0 if none/unknown.

Source code in musicalgestures/_utils.py
1352
1353
1354
1355
1356
1357
1358
1359
1360
1361
1362
1363
1364
1365
1366
1367
1368
1369
1370
1371
1372
1373
1374
1375
1376
1377
1378
1379
def get_rotation(filename: str) -> int:
    """
    Returns the display rotation (degrees) stored in a video's metadata.

    Phone/portrait videos often store landscape pixels plus a rotation flag (display
    matrix). FFmpeg's frame pipe applies this automatically while OpenCV's VideoCapture
    does not, which can leave some processes rotated. This reads the flag so the
    orientation can be normalised.

    Returns:
        int: rotation in degrees (e.g. 0, 90, 180, 270), or 0 if none/unknown.
    """
    import subprocess
    # Preferred: structured side-data rotation (one value per packet; take the first)
    for entries in ("side_data=rotation", "stream_side_data=rotation", "stream_tags=rotate"):
        try:
            out = subprocess.run(
                ["ffprobe", "-v", "error", "-select_streams", "v:0",
                 "-show_entries", entries, "-of", "default=nk=1:nw=1", filename],
                stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                universal_newlines=True, timeout=10).stdout
            for line in out.splitlines():
                line = line.strip()
                if line and line.lstrip("-").isdigit():
                    return int(line) % 360
        except Exception:
            continue
    return 0

normalize_rotation

normalize_rotation(filename, overwrite=True)

If a video carries a display-rotation flag (e.g. a phone portrait recording with landscape pixels), re-encode it so the rotation is baked into the pixels and the flag removed. This makes every downstream reader (FFmpeg pipe, OpenCV, filters) agree on the orientation, preventing some processes from coming out rotated.

Parameters:

Name Type Description Default
filename str

Path to the video file.

required
overwrite bool

Overwrite the "_oriented" output if it exists. Defaults to True.

True

Returns:

Name Type Description
str str

Path to an upright video — the original if it had no rotation, otherwise a new "_oriented" copy.

Source code in musicalgestures/_utils.py
1382
1383
1384
1385
1386
1387
1388
1389
1390
1391
1392
1393
1394
1395
1396
1397
1398
1399
1400
1401
1402
1403
1404
1405
1406
1407
1408
1409
1410
1411
1412
1413
1414
def normalize_rotation(filename: str, overwrite: bool = True) -> str:
    """
    If a video carries a display-rotation flag (e.g. a phone portrait recording with
    landscape pixels), re-encode it so the rotation is baked into the pixels and the
    flag removed. This makes every downstream reader (FFmpeg pipe, OpenCV, filters)
    agree on the orientation, preventing some processes from coming out rotated.

    Args:
        filename (str): Path to the video file.
        overwrite (bool): Overwrite the "_oriented" output if it exists. Defaults to True.

    Returns:
        str: Path to an upright video — the original if it had no rotation, otherwise a
            new "_oriented" copy.
    """
    import os
    rotation = get_rotation(filename)
    if rotation % 360 == 0:
        return filename

    of, fex = os.path.splitext(filename)
    target = of + '_oriented' + fex
    if not overwrite:
        target = generate_outfilename(target)

    print(f"Detected {rotation}° rotation metadata — baking orientation into the pixels "
          "so all processes keep the original orientation.")
    cmd = ['ffmpeg', '-y', '-i', filename, '-c:v', 'libx264', '-pix_fmt', 'yuv420p']
    if has_audio(filename):
        cmd += ['-c:a', 'copy']
    cmd += [target]
    ffmpeg_cmd(cmd, get_length(filename), pb_prefix='Normalising orientation:')
    return target

get_length

get_length(filename)

Gets the length (in seconds) of a video using FFprobe.

Parameters:

Name Type Description Default
filename str

Path to the video file to measure.

required

Returns:

Name Type Description
float float

The length of the input video file in seconds.

Source code in musicalgestures/_utils.py
1417
1418
1419
1420
1421
1422
1423
1424
1425
1426
1427
1428
1429
1430
1431
1432
1433
1434
1435
1436
1437
1438
1439
1440
1441
1442
def get_length(filename: str) -> float:
    """
    Gets the length (in seconds) of a video using FFprobe.

    Args:
        filename (str): Path to the video file to measure.

    Returns:
        float: The length of the input video file in seconds.
    """
    out = ffprobe(filename)
    out_array = out.splitlines()
    duration = None
    at_line = -1
    while duration is None:
        duration = out_array[at_line] if out_array[at_line].find(
            "Duration:") != -1 else None
        at_line -= 1
        if at_line < -len(out_array):
            raise NoDurationError(
                "Could not get duration.")
    duration_array = duration.split(' ')
    time_string_index = duration_array.index("Duration:") + 1
    time_string = duration_array[time_string_index][:-1]
    elems = [float(elem) for elem in time_string.split(':')]
    return elems[0]*3600 + elems[1]*60 + elems[2]

get_samplerate

get_samplerate(filename)

Gets the sampling rate (in Hz) of a file's audio track using FFprobe.

This exists because librosa cannot be asked. Until version 1.0 librosa fell back to audioread (and so to ffmpeg) for containers libsndfile cannot open; 1.0 removed that fallback, so librosa.get_samplerate on a video raises soundfile.LibsndfileError: Format not recognised. FFprobe reads the container header without decoding it, which is what a constructor should be doing anyway.

Parameters:

Name Type Description Default
filename str

Path to the video or audio file to measure.

required

Returns:

Name Type Description
int int

The sampling rate of the file's audio track, in Hz.

Raises:

Type Description
NoStreamError

If the file has no audio track, or ffprobe reports no rate for it.

Source code in musicalgestures/_utils.py
1445
1446
1447
1448
1449
1450
1451
1452
1453
1454
1455
1456
1457
1458
1459
1460
1461
1462
1463
1464
1465
1466
1467
1468
1469
1470
1471
1472
1473
1474
1475
1476
def get_samplerate(filename: str) -> int:
    """
    Gets the sampling rate (in Hz) of a file's audio track using FFprobe.

    This exists because librosa cannot be asked. Until version 1.0 librosa fell back
    to audioread (and so to ffmpeg) for containers libsndfile cannot open; 1.0 removed
    that fallback, so `librosa.get_samplerate` on a video raises
    `soundfile.LibsndfileError: Format not recognised`. FFprobe reads the container
    header without decoding it, which is what a constructor should be doing anyway.

    Args:
        filename (str): Path to the video or audio file to measure.

    Returns:
        int: The sampling rate of the file's audio track, in Hz.

    Raises:
        NoStreamError: If the file has no audio track, or ffprobe reports no rate for it.
    """
    import re

    out = ffprobe(filename)
    for line in reversed(out.splitlines()):
        if "Audio:" not in line:
            continue
        # ffprobe writes the rate on the audio stream line, e.g.
        #   Stream #0:1: Audio: mp3 (U[0][0][0] / 0x0055), 44100 Hz, stereo, fltp, 128 kb/s
        match = re.search(r"(\d+)\s*Hz", line)
        if match:
            return int(match.group(1))
        raise NoStreamError(f"Could not read a sampling rate for {filename}.")
    raise NoStreamError(f"{filename} has no audio stream.")

audio_source

audio_source(filename)

Returns a path that libsndfile can open, extracting the audio track if it cannot.

librosa reads audio through soundfile, which is libsndfile, which knows audio containers and not video ones. Passing it a .avi raises Format not recognised. This is the same try-then-convert shape MgAudioProcessor already uses in _colored.py, lifted out so every librosa call site can share it -- and cached, because a waveform, a spectrogram and a descriptor pass over one video would otherwise each re-extract the same track.

The extracted file sits beside the source as <name>.wav, which is where extract_wav and convert already put such things.

Parameters:

Name Type Description Default
filename str

Path to the audio or video file.

required

Returns:

Name Type Description
str str

filename itself when libsndfile can open it, else the extracted .wav.

Source code in musicalgestures/_utils.py
1484
1485
1486
1487
1488
1489
1490
1491
1492
1493
1494
1495
1496
1497
1498
1499
1500
1501
1502
1503
1504
1505
1506
1507
1508
1509
1510
1511
1512
1513
1514
1515
1516
1517
1518
1519
1520
1521
1522
1523
1524
1525
1526
1527
1528
1529
def audio_source(filename: str) -> str:
    """
    Returns a path that libsndfile can open, extracting the audio track if it cannot.

    librosa reads audio through soundfile, which is libsndfile, which knows audio
    containers and not video ones. Passing it a .avi raises `Format not recognised`.
    This is the same try-then-convert shape `MgAudioProcessor` already uses in
    `_colored.py`, lifted out so every librosa call site can share it -- and cached,
    because a waveform, a spectrogram and a descriptor pass over one video would
    otherwise each re-extract the same track.

    The extracted file sits beside the source as `<name>.wav`, which is where
    `extract_wav` and `convert` already put such things.

    Args:
        filename (str): Path to the audio or video file.

    Returns:
        str: `filename` itself when libsndfile can open it, else the extracted .wav.
    """
    import os

    import soundfile as sf

    try:
        sf.info(filename)
        return filename
    except RuntimeError:
        # LibsndfileError subclasses RuntimeError, and so did the error soundfile
        # raised before that class existed, so this catches both vintages.
        pass

    try:
        key = (os.path.realpath(filename),) + tuple(
            getattr(os.stat(filename), a) for a in ("st_mtime", "st_size"))
    except OSError:
        key = None
    if key is not None and key in _AUDIO_SOURCE_CACHE:
        cached = _AUDIO_SOURCE_CACHE[key]
        if os.path.isfile(cached):
            return cached

    target = extract_wav(filename, target_name=filename + ".wav")
    if key is not None:
        _AUDIO_SOURCE_CACHE[key] = target
    return target

get_framecount

get_framecount(filename, fast=True)

Returns the number of frames in a video using FFprobe.

Parameters:

Name Type Description Default
filename str

Path to the video file to measure.

required
fast bool

If True (default), count demuxed video packets (-count_packets). This is fast (no decoding) and — unlike the container's nb_frames metadata, which is unreliable (e.g. off by one on many AVIs, or absent on WebM) — matches the true decoded frame count for normal video streams. If False, fully decode and count frames (-count_frames): the ground truth, but slower. Defaults to True.

True

Returns:

Name Type Description
int int

The number of frames in the input video file.

Source code in musicalgestures/_utils.py
1532
1533
1534
1535
1536
1537
1538
1539
1540
1541
1542
1543
1544
1545
1546
1547
1548
1549
1550
1551
1552
1553
1554
1555
1556
1557
1558
1559
1560
1561
1562
1563
1564
1565
1566
1567
1568
1569
1570
1571
1572
1573
1574
1575
1576
1577
1578
1579
1580
1581
1582
1583
1584
1585
1586
1587
def get_framecount(filename: str, fast: bool = True) -> int:
    """
    Returns the number of frames in a video using FFprobe.

    Args:
        filename (str): Path to the video file to measure.
        fast (bool, optional): If True (default), count demuxed video packets
            (``-count_packets``). This is fast (no decoding) and — unlike the container's
            ``nb_frames`` metadata, which is unreliable (e.g. off by one on many AVIs, or absent
            on WebM) — matches the true decoded frame count for normal video streams. If False,
            fully decode and count frames (``-count_frames``): the ground truth, but slower.
            Defaults to True.

    Returns:
        int: The number of frames in the input video file.
    """
    import subprocess
    # Counting packets is the "fast" path: it does not decode, yet (one packet ≈ one frame for
    # video) it agrees with the decoded count, whereas the container's nb_frames metadata does not.
    command_count_packets = 'ffprobe -v error -select_streams v:0 -count_packets -show_entries stream=nb_read_packets -of default=nokey=1:noprint_wrappers=1'.split(
        ' ')
    command_count_packets.append(filename)
    command_count = 'ffprobe -v error -count_frames -select_streams v:0 -show_entries stream=nb_read_frames -of default=nokey=1:noprint_wrappers=1'.split(
        ' ')
    command_count.append(filename)
    command = command_count_packets if fast else command_count

    process = subprocess.Popen(
        command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
    try:
        out, err = process.communicate(timeout=10)
    except subprocess.TimeoutExpired:
        process.kill()
        out, err = process.communicate()

    if err:
        raise FFprobeError(err)

    elif out:
        if out.splitlines()[-1].find("No such file or directory") != -1:
            raise FileNotFoundError(out.splitlines()[-1])
        elif out.startswith("N/A"):
            if fast:
                return get_framecount(filename, fast=False)
            else:
                raise FFprobeError(
                    "Could not count frames. (Is this a video file?) If you are working with audio file use MgAudio instead.")
        else:
            return int(out)

    else:
        if fast:
            return get_framecount(filename, fast=False)
        else:
            raise FFprobeError(
                "Could not count frames. (Is this a video file?). If you are working with audio file use MgAudio instead.")

get_fps

get_fps(filename)

Gets the FPS (frames per second) value of a video using FFprobe.

Parameters:

Name Type Description Default
filename str

Path to the video file to measure.

required

Returns:

Name Type Description
float float

The FPS value of the input video file.

Source code in musicalgestures/_utils.py
1590
1591
1592
1593
1594
1595
1596
1597
1598
1599
1600
1601
1602
1603
1604
1605
1606
1607
1608
1609
1610
1611
1612
1613
1614
1615
1616
1617
1618
1619
1620
def get_fps(filename: str) -> float:
    """
    Gets the FPS (frames per second) value of a video using FFprobe.

    Args:
        filename (str): Path to the video file to measure.

    Returns:
        float: The FPS value of the input video file.
    """
    out = ffprobe(filename)
    out_array = out.splitlines()
    video_stream = None
    at_line = -1
    while video_stream is None:
        video_stream = out_array[at_line] if out_array[at_line].find(
            "Video:") != -1 else None
        at_line -= 1
        if at_line < -len(out_array):
            raise NoStreamError(
                "No video stream found. (Is this a video file?)")
    video_stream_array = video_stream.split(',')
    fps = None
    at_chunk = -1
    while fps is None:
        fps = float(video_stream_array[at_chunk].split(
            ' ')[-2]) if video_stream_array[at_chunk].split(' ')[-1] == 'fps' else None
        at_chunk -= 1
        if at_chunk < -len(video_stream_array):
            raise FFprobeError("Could not fetch FPS.")
    return fps

get_first_frame_as_image

get_first_frame_as_image(filename, target_name=None, pict_format='.png', overwrite=True)

Extracts the first frame of a video and saves it as an image using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
target_name str

The name for the output image. Defaults to None (which assumes that the input filename should be used).

None
pict_format str

The format to use for the output image. Defaults to '.png'.

'.png'
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

Path to the output image file.

Source code in musicalgestures/_utils.py
1623
1624
1625
1626
1627
1628
1629
1630
1631
1632
1633
1634
1635
1636
1637
1638
1639
1640
1641
1642
1643
1644
1645
1646
1647
1648
1649
1650
1651
1652
def get_first_frame_as_image(filename: str, target_name: str | None = None, pict_format: str = '.png', overwrite: bool = True) -> str:
    """
    Extracts the first frame of a video and saves it as an image using ffmpeg.

    Args:
        filename (str): Path to the input video file.
        target_name (str, optional): The name for the output image. Defaults to None (which assumes that the input filename should be used).
        pict_format (str, optional): The format to use for the output image. Defaults to '.png'.
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output image file.
    """

    import os
    of = os.path.splitext(filename)[0]

    if target_name is None:
        target_name = of + pict_format
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_container_is(pict_format, target_name)

    cmd = ' '.join(['ffmpeg', '-y', '-i', wrap_str(filename),
                    '-frames', '1', wrap_str(target_name)])

    os.system(cmd)

    return target_name

get_box_video_ratio

get_box_video_ratio(filename, box_width=800, box_height=600)

Gets the box-to-video ratio between an arbitrarily defind box and the video dimensions. Useful to fit windows into a certain area.

Parameters:

Name Type Description Default
filename str

Path to the input video file.

required
box_width int

The width of the box to fit the video into.

800
box_height int

The height of the box to fit the video into.

600

Returns:

Name Type Description
int float

The smallest ratio (ie. the one to use for scaling the video window to fit into the box).

Source code in musicalgestures/_utils.py
1655
1656
1657
1658
1659
1660
1661
1662
1663
1664
1665
1666
1667
1668
1669
1670
1671
1672
1673
1674
1675
1676
1677
1678
def get_box_video_ratio(filename: str, box_width: int = 800, box_height: int = 600) -> float:
    """
    Gets the box-to-video ratio between an arbitrarily defind box and the video dimensions. Useful to fit windows into a certain area.

    Args:
        filename (str): Path to the input video file.
        box_width (int, optional): The width of the box to fit the video into.
        box_height (int, optional): The height of the box to fit the video into.

    Returns:
        int: The smallest ratio (ie. the one to use for scaling the video window to fit into the box).
    """

    video_width, video_height = get_widthheight(filename)

    ratio_x, ratio_y = clamp(box_width / video_width,
                             0, 1), clamp(box_height / video_height, 0, 1)

    smallest_ratio = sorted([ratio_x, ratio_y])[0]

    if smallest_ratio < 1:
        smallest_ratio *= 0.9

    return smallest_ratio

quality_metrics

quality_metrics(original, processed, metric=None)

Compute video quality metrics between two video files for comparing the quality of video codecs or measuring the efficacy of encoding configuration. Possible to compute three major video quality metrics used for objective evaluation, namely:

  • PSNR: It is the most commonly used video quality metric. But it has the lowest predictive value, so the results are inconsistent. Used by major platforms like Netflix and Facebook to compare different codecs and for similar use cases. Overall usage is declining.
  • SSIM: Mostly used by technical experts like codec researchers and compression engineers. Usage is declining steadily. However, it has a higher predictive value than PSNR.
  • VMAF: Introduced first by Netflix but then converted into an open-source asset. VMAF is easily accessible and widely used. Designed specifically for evaluating the video quality of streams encoded for multiple-resolution rungs.

Parameters:

Name Type Description Default
original str

Path to the original/reference video file.

required
processed str

Path to the processed/distorted video file.

required
metric str

Type of quality metric to compute ('vmaf', 'ssim', or 'psnr'). Defaults to None (which computes all the metrics).

None
Source code in musicalgestures/_utils.py
1681
1682
1683
1684
1685
1686
1687
1688
1689
1690
1691
1692
1693
1694
1695
1696
1697
1698
1699
1700
1701
1702
1703
1704
1705
1706
1707
1708
1709
1710
1711
1712
1713
1714
1715
1716
1717
1718
1719
1720
1721
1722
1723
1724
1725
1726
1727
def quality_metrics(original: str, processed: str, metric: str | None = None) -> None:
    """
    Compute video quality metrics between two video files for comparing the quality of video codecs or measuring the efficacy of encoding configuration.
    Possible to compute three major video quality metrics used for objective evaluation, namely:

    - PSNR: It is the most commonly used video quality metric. But it has the lowest predictive value, so the results are inconsistent. 
      Used by major platforms like Netflix and Facebook to compare different codecs and for similar use cases. Overall usage is declining.
    - SSIM: Mostly used by technical experts like codec researchers and compression engineers. 
      Usage is declining steadily. However, it has a higher predictive value than PSNR.
    - VMAF: Introduced first by Netflix but then converted into an open-source asset. VMAF is easily accessible and widely used. 
      Designed specifically for evaluating the video quality of streams encoded for multiple-resolution rungs.

    Args:
        original (str): Path to the original/reference video file.
        processed (str): Path to the processed/distorted video file.
        metric (str, optional): Type of quality metric to compute ('vmaf', 'ssim', or 'psnr'). Defaults to None (which computes all the metrics).

    """
    import subprocess, re

    if metric is None: # compute all the video quality metrics (VMAF, SSIM and PSNR)
        cmd = ['ffmpeg', '-i', processed, '-i', original, '-lavfi', 'libvmaf', '-lavfi', "[0][1]ssim;[0][1]psnr", '-f', 'null', '-']
    elif metric == 'vmaf'.lower(): # compute the VMAF score
        cmd = ['ffmpeg', '-i', processed, '-i', original, '-lavfi', 'libvmaf', '-f', 'null', '-']
    elif metric == 'ssim'.lower(): # compute the SSIM score
        cmd = ['ffmpeg', '-i', processed, '-i', original, '-lavfi', 'ssim', '-f', 'null', '-']
    elif metric == 'psnr'.lower(): # compute the PSNR score
        cmd = ['ffmpeg', '-i', processed, '-i', original, '-lavfi', 'psnr', '-f', 'null', '-']
    else:
        print(f"The metric {metric} is not available in the toolbox. Please refer to the following metrics 'vmaf', 'ssim' or 'psnr'.")
        return

    # Run the FFmpeg command using subprocess
    process = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
    out, _ = process.communicate() # read data from stdout and stderr, until end-of-file is reached
    splitted = out.split('\n') # split output

    # Check if string is present in each string in the list using enumerate
    for index, item in enumerate(splitted):
        if isinstance(item, str) and re.search('VMAF', item):
            print(splitted[index].split("] ")[1])
        elif isinstance(item, str) and re.search('SSIM', item):
            print(splitted[index].split("] ")[1])
        elif isinstance(item, str) and re.search('PSNR', item):
            print(splitted[index].split("] ")[1])
        elif isinstance(item, str) and re.search('Conversion failed', item):
            print(splitted[index])

audio_dilate

audio_dilate(filename, dilation_ratio=1, target_name=None, overwrite=True)

Time-stretches or -shrinks (dilates) an audio file using ffmpeg.

Parameters:

Name Type Description Default
filename str

Path to the audio file to dilate.

required
dilation_ratio float

The source file's length divided by the resulting file's length. Defaults to 1.

1
target_name str

The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_dilated" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

True

Returns:

Name Type Description
str str

The path to the output audio file.

Source code in musicalgestures/_utils.py
1730
1731
1732
1733
1734
1735
1736
1737
1738
1739
1740
1741
1742
1743
1744
1745
1746
1747
1748
1749
1750
1751
1752
1753
1754
1755
1756
1757
def audio_dilate(filename: str, dilation_ratio: float = 1, target_name: str | None = None, overwrite: bool = True) -> str:
    """
    Time-stretches or -shrinks (dilates) an audio file using ffmpeg.

    Args:
        filename (str): Path to the audio file to dilate.
        dilation_ratio (float, optional): The source file's length divided by the resulting file's length. Defaults to 1.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_dilated" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: The path to the output audio file.
    """

    import os
    of, fex = os.path.splitext(filename)

    if target_name is None:
        target_name = of + '_dilated' + fex
    if not overwrite:
        target_name = generate_outfilename(target_name)

    pass_if_containers_match(filename, target_name)

    cmds = ' '.join(['ffmpeg', '-loglevel', 'quiet', '-y', '-i', wrap_str(filename), '-codec:a', 'pcm_s16le',
                     '-filter:a', 'atempo=' + str(dilation_ratio), wrap_str(target_name)])
    os.system(cmds)
    return target_name

embed_audio_in_video

embed_audio_in_video(source_audio, destination_video, dilation_ratio=1)

Embeds an audio file as the audio channel of a video file using ffmpeg.

Parameters:

Name Type Description Default
source_audio str

Path to the audio file to embed.

required
destination_video str

Path to the video file to embed the audio file in.

required
dilation_ratio float

The source file's length divided by the resulting file's length. Defaults to 1.

1
Source code in musicalgestures/_utils.py
1760
1761
1762
1763
1764
1765
1766
1767
1768
1769
1770
1771
1772
1773
1774
1775
1776
1777
1778
1779
1780
1781
1782
1783
1784
1785
1786
1787
1788
1789
1790
1791
1792
1793
1794
1795
1796
1797
def embed_audio_in_video(source_audio: str, destination_video: str, dilation_ratio: float = 1) -> None:
    """
    Embeds an audio file as the audio channel of a video file using ffmpeg.

    Args:
        source_audio (str): Path to the audio file to embed.
        destination_video (str): Path to the video file to embed the audio file in.
        dilation_ratio (float, optional): The source file's length divided by the resulting file's length. Defaults to 1.
    """

    import os
    of, fex = os.path.splitext(destination_video)

    # dilate audio file if necessary (ie. when skipping)
    if dilation_ratio != 1:
        audio_to_embed = audio_dilate(source_audio, dilation_ratio)  # creates '_dilated.wav'
        dilated = True
    else:
        audio_to_embed = source_audio
        dilated = False

    # embed audio in video
    outname = of + '_w_audio' + fex

    cmds = ' '.join(['ffmpeg', '-hide_banner', '-loglevel', 'quiet', '-y', 
                     '-i', wrap_str(destination_video), '-i', wrap_str(audio_to_embed), 
                     '-c:v', 'copy', '-map', '0:v:0', '-map', '1:a:0', 
                     '-shortest', wrap_str(outname)])

    os.system(cmds)  # creates '_w_audio.avi'

    # cleanup:
    # if we needed to create an additional (dilated) audio file, delete it
    if dilated:
        os.remove(audio_to_embed)
    # replace (silent) destination_video with the one with the embedded audio
    os.remove(destination_video)
    os.rename(outname, destination_video)

ffmpeg_cmd

ffmpeg_cmd(command, total_time, pb_prefix='Progress', print_cmd=False, stream=True, pipe=None)

Run an ffmpeg command in a subprocess and show progress using an MgProgressbar.

Parameters:

Name Type Description Default
command list

The ffmpeg command to execute as a list. Eg. ['ffmpeg', '-y', '-i', 'myVid.mp4', 'myVid.mov']

required
total_time float

The length of the output. Needed mainly for the progress bar.

required
pb_prefix str

The prefix for the progress bar. Defaults to 'Progress'.

'Progress'
print_cmd bool

Whether to print the full ffmpeg command to the console before executing it. Good for debugging. Defaults to False.

False
stream bool

Whether to have a continuous output stream or just (the last) one. Defaults to True (continuous stream).

True
pipe str

Whether to pipe video frames from FFmpeg to numpy array. Possible to read the video frame by frame with pipe='read', to load video in memory with pipe='load', or to write the frames of a numpy array to a video file with pipe='write'. Defaults to None.

None

Raises:

Type Description
KeyboardInterrupt

If the user stops the process.

FFmpegError

If the ffmpeg process was unsuccessful.

Source code in musicalgestures/_utils.py
1805
1806
1807
1808
1809
1810
1811
1812
1813
1814
1815
1816
1817
1818
1819
1820
1821
1822
1823
1824
1825
1826
1827
1828
1829
1830
1831
1832
1833
1834
1835
1836
1837
1838
1839
1840
1841
1842
1843
1844
1845
1846
1847
1848
1849
1850
1851
1852
1853
1854
1855
1856
1857
1858
1859
1860
1861
1862
1863
1864
1865
1866
1867
1868
1869
1870
1871
1872
1873
1874
1875
1876
1877
1878
1879
1880
1881
1882
1883
1884
1885
1886
1887
1888
1889
1890
1891
def ffmpeg_cmd(command: list, total_time: float, pb_prefix: str = 'Progress', print_cmd: bool = False, stream: bool = True, pipe: str | None = None):
    """
    Run an ffmpeg command in a subprocess and show progress using an MgProgressbar.

    Args:
        command (list): The ffmpeg command to execute as a list. Eg. ['ffmpeg', '-y', '-i', 'myVid.mp4', 'myVid.mov']
        total_time (float): The length of the output. Needed mainly for the progress bar.
        pb_prefix (str, optional): The prefix for the progress bar. Defaults to 'Progress'.
        print_cmd (bool, optional): Whether to print the full ffmpeg command to the console before executing it. Good for debugging. Defaults to False.
        stream (bool, optional): Whether to have a continuous output stream or just (the last) one. Defaults to True (continuous stream).
        pipe (str, optional): Whether to pipe video frames from FFmpeg to numpy array. Possible to read the video frame by frame with pipe='read', to load video in memory with pipe='load', or to write the frames of a numpy array to a video file with pipe='write'. Defaults to None.

    Raises:
        KeyboardInterrupt: If the user stops the process.
        FFmpegError: If the ffmpeg process was unsuccessful.
    """
    import subprocess

    pb = MgProgressbar(total=total_time, prefix=pb_prefix)

    # Hide banner and quiet report printing
    command = ['ffmpeg', '-hide_banner', '-loglevel', 'quiet'] + command[1:]

    if print_cmd:
        if isinstance(command, list):
            print(' '.join(command))
        else:
            print(command)

    if pipe == 'read':
        # Define ffmpeg command and read frame by frame
        command = command + ['-f', 'image2pipe', '-pix_fmt', 'bgr24', '-vcodec', 'rawvideo', '-preset', 'ultrafast', '-']
        process = subprocess.Popen(command, stdout=subprocess.PIPE, bufsize=-1)
        return process

    elif pipe == 'load':
        # Define ffmpeg command and load all video frames in memory
        command = command + ['-f', 'image2pipe', '-pix_fmt', 'bgr24', '-vcodec', 'rawvideo', '-preset', 'ultrafast', '-']
        process = subprocess.run(command, stdout=subprocess.PIPE, bufsize=-1)
        return process

    elif pipe == 'write':
        # Write the frames of a numpy array to a video file
        process = subprocess.Popen(command, stdin=subprocess.PIPE, bufsize=-1)
        return process

    else:
        process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
        returncode = None
        all_out = ''

        try:
            while True:
                if stream:
                    out = process.stdout.readline()
                else:
                    out = process.stdout.read()
                all_out += out

                if out == '':
                    process.wait()
                    returncode = process.returncode
                    break

                elif out.startswith('frame='):
                    try:
                        out_list = out.split()
                        time_ind = [elem.startswith('time=') for elem in out_list].index(True)
                        time_str = out_list[time_ind][5:]
                        time_sec = str2sec(time_str)
                        pb.progress(time_sec)
                    except ValueError:
                        # New version of FFmpeg outputs N/A values
                        pass

            if returncode in [None, 0]:
                pb.progress(total_time)
            else:
                raise FFmpegError(f"return code: {returncode}"+all_out)

        except KeyboardInterrupt:
            try:
                process.terminate()
            except OSError:
                pass
            process.wait()
            raise KeyboardInterrupt

str2sec

str2sec(time_string)

Converts a time code string into seconds.

Parameters:

Name Type Description Default
time_string str

The time code to convert. Eg. '01:33:42'.

required

Returns:

Name Type Description
float float

The time code converted to seconds.

Source code in musicalgestures/_utils.py
1894
1895
1896
1897
1898
1899
1900
1901
1902
1903
1904
1905
def str2sec(time_string: str) -> float:
    """
    Converts a time code string into seconds.

    Args:
        time_string (str): The time code to convert. Eg. '01:33:42'.

    Returns:
        float: The time code converted to seconds.
    """
    elems = [float(elem) for elem in time_string.split(':')]
    return elems[0]*3600 + elems[1]*60 + elems[2]

wrap_str

wrap_str(string, matchers=[' ', '(', ')'])

Wraps a string in double quotes if it contains any of matchers - by default: space or parentheses. Useful when working with shell commands.

Parameters:

Name Type Description Default
string str

The string to inspect.

required
matchers list

The list of characters to look for in the string. Defaults to [" ", "(", ")"].

[' ', '(', ')']

Returns:

Name Type Description
str str

The (wrapped) string.

Source code in musicalgestures/_utils.py
1908
1909
1910
1911
1912
1913
1914
1915
1916
1917
1918
1919
1920
1921
1922
1923
1924
1925
1926
def wrap_str(string: str, matchers: list = [" ", "(", ")"]) -> str:
    """
    Wraps a string in double quotes if it contains any of `matchers` - by default: space or parentheses.
    Useful when working with shell commands.

    Args:
        string (str): The string to inspect.
        matchers (list, optional): The list of characters to look for in the string. Defaults to [" ", "(", ")"].

    Returns:
        str: The (wrapped) string.
    """

    matchers = [" ", "(", ")"]

    if any(True for char in string if char in matchers) and '"' not in [string[0], string[-1]]:
        return '"' + string + '"'
    else:
        return string

unwrap_str

unwrap_str(string)

Unwraps a string from quotes.

Parameters:

Name Type Description Default
string str

The string to inspect.

required

Returns:

Name Type Description
str str

The (unwrapped) string.

Source code in musicalgestures/_utils.py
1929
1930
1931
1932
1933
1934
1935
1936
1937
1938
1939
1940
1941
1942
1943
1944
def unwrap_str(string: str) -> str:
    """
    Unwraps a string from quotes.

    Args:
        string (str): The string to inspect.

    Returns:
        str: The (unwrapped) string.
    """
    if '"' in [string[0], string[-1]]:
        return string[1:-1]
    elif "'" in [string[0], string[-1]]:
        return string[1:-1]
    else:
        return string

get_cuda_device_count

get_cuda_device_count()

Returns the number of CUDA-capable GPU devices visible to OpenCV.

Returns:

Name Type Description
int int

Number of available CUDA devices, or 0 if the OpenCV CUDA module is unavailable or no devices are detected.

Source code in musicalgestures/_utils.py
1947
1948
1949
1950
1951
1952
1953
1954
1955
1956
1957
1958
1959
def get_cuda_device_count() -> int:
    """
    Returns the number of CUDA-capable GPU devices visible to OpenCV.

    Returns:
        int: Number of available CUDA devices, or 0 if the OpenCV CUDA
             module is unavailable or no devices are detected.
    """
    try:
        import cv2
        return cv2.cuda.getCudaEnabledDeviceCount()
    except Exception:
        return 0

cuda_build_available

cuda_build_available()

Returns whether the installed OpenCV was compiled with CUDA support.

Returns:

Name Type Description
bool bool

True if OpenCV's build information reports CUDA support, else False.

Source code in musicalgestures/_utils.py
1962
1963
1964
1965
1966
1967
1968
1969
1970
1971
1972
1973
1974
1975
1976
1977
1978
def cuda_build_available() -> bool:
    """
    Returns whether the installed OpenCV was compiled with CUDA support.

    Returns:
        bool: True if OpenCV's build information reports CUDA support, else False.
    """
    try:
        import cv2
        info = cv2.getBuildInformation()
    except Exception:
        return False
    for line in info.splitlines():
        stripped = line.strip()
        if stripped.startswith('NVIDIA CUDA') or stripped.startswith('CUDA:'):
            return 'YES' in stripped.upper()
    return False

cuda_unavailable_reason

cuda_unavailable_reason()

Returns a short, actionable explanation of why the OpenCV CUDA backend is unavailable.

Distinguishes the common case (the pip OpenCV wheels are built without CUDA) from the case where OpenCV has CUDA but no GPU/driver is detected.

Returns:

Name Type Description
str str

A human-readable explanation.

Source code in musicalgestures/_utils.py
1981
1982
1983
1984
1985
1986
1987
1988
1989
1990
1991
1992
1993
1994
1995
1996
1997
1998
1999
2000
2001
def cuda_unavailable_reason() -> str:
    """
    Returns a short, actionable explanation of why the OpenCV CUDA backend is unavailable.

    Distinguishes the common case (the pip OpenCV wheels are built without CUDA) from the
    case where OpenCV has CUDA but no GPU/driver is detected.

    Returns:
        str: A human-readable explanation.
    """
    if not cuda_build_available():
        return (
            'The installed OpenCV (pip "opencv-python"/"opencv-contrib-python") is built '
            'WITHOUT CUDA, so the GPU cannot be used even if your machine has one. GPU '
            'acceleration requires an OpenCV compiled with CUDA + cuDNN — build it from '
            'source with -D WITH_CUDA=ON, or install a CUDA-enabled OpenCV build.'
        )
    return (
        'OpenCV was built with CUDA but no CUDA-capable GPU was detected. Check your '
        'NVIDIA driver and CUDA runtime installation.'
    )

in_colab

in_colab()

Check's if the environment is a Google Colab document.

Returns:

Name Type Description
bool bool

True if the environment is a Colab document, otherwise False.

Source code in musicalgestures/_utils.py
2004
2005
2006
2007
2008
2009
2010
2011
2012
2013
2014
2015
2016
def in_colab() -> bool:
    """
    Check's if the environment is a Google Colab document.

    Returns:
        bool: True if the environment is a Colab document, otherwise False.
    """
    result = None
    try:
        result = 'google.colab' in str(get_ipython())
    except NameError:
        result = False
    return result

in_ipynb

in_ipynb()

Check if the environment is a Jupyter notebook. Taken from https://stackoverflow.com/questions/15411967/how-can-i-check-if-code-is-executed-in-the-ipython-notebook.

Returns:

Name Type Description
bool bool

True if the environment is a Jupyter notebook, otherwise False.

Source code in musicalgestures/_utils.py
2019
2020
2021
2022
2023
2024
2025
2026
2027
2028
2029
2030
2031
2032
2033
2034
2035
2036
def in_ipynb() -> bool:
    """
    Check if the environment is a Jupyter notebook.
    Taken from https://stackoverflow.com/questions/15411967/how-can-i-check-if-code-is-executed-in-the-ipython-notebook.

    Returns:
        bool: True if the environment is a Jupyter notebook, otherwise False.
    """
    try:
        shell = get_ipython().__class__.__name__
        if shell == 'ZMQInteractiveShell':
            return True   # Jupyter notebook or qtconsole
        elif shell == 'TerminalInteractiveShell':
            return False  # Terminal running IPython
        else:
            return False  # Other type (?)
    except NameError:
        return False      # Probably standard Python interpreter

merge_videos

merge_videos(media_paths, target_name=None, overwrite=False, print_cmd=False)

Merges a list of video files into a single video file using ffmpeg.

Parameters:

Name Type Description Default
media_paths list

List of paths to the video files to merge.

required
target_name str

The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_merged" should be used).

None
overwrite bool

Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

False

Returns:

Name Type Description
str str

Path to the output video.

Source code in musicalgestures/_utils.py
2044
2045
2046
2047
2048
2049
2050
2051
2052
2053
2054
2055
2056
2057
2058
2059
2060
2061
2062
2063
2064
2065
2066
2067
2068
2069
2070
2071
2072
2073
2074
2075
2076
2077
2078
2079
2080
2081
2082
2083
2084
2085
2086
2087
2088
2089
2090
2091
2092
2093
2094
2095
2096
2097
2098
2099
2100
2101
2102
2103
2104
2105
2106
2107
2108
2109
2110
2111
2112
2113
def merge_videos(
    media_paths: list, target_name: str = None, overwrite: bool = False, print_cmd: bool = False
) -> str:
    """
    Merges a list of video files into a single video file using ffmpeg.

    Args:
        media_paths (list): List of paths to the video files to merge.
        target_name (str, optional): The name of the output video. Defaults to None (which assumes that the input filename with the suffix "_merged" should be used).
        overwrite (bool, optional): Whether to allow overwriting existing files or to automatically increment target filename to avoid overwriting. Defaults to True.

    Returns:
        str: Path to the output video.
    """

    if len(media_paths) == 0:
        raise ValueError("The list of media paths is empty.")
    elif len(media_paths) == 1:
        return media_paths[0]

    import os
    from musicalgestures._utils import generate_outfilename

    # check if all media files have the same container, same resolution and same fps
    try:
        for media in media_paths:
            pass_if_containers_match(media, media_paths[0])
            assert get_widthheight(media) == get_widthheight(media_paths[0])
            assert get_fps(media) == get_fps(media_paths[0])
    except WrongContainer:
        raise FilesNotMatchError("All media files must be in the same container.")
    except AssertionError:
        raise FilesNotMatchError("All media files must have the same resolution and fps.")

    # set target name, a new file in the same directory as the first media file
    of, fex = os.path.splitext(media_paths[0])
    of = os.path.abspath(of)
    # create a tmp .txt file for concat
    txt_path = os.path.join(os.path.dirname(media_paths[0]), "tmp.txt")
    with open(os.path.join(txt_path), "w") as f:
        for media in media_paths:
            f.write(f"file '{os.path.abspath(media)}'\n")

    # if files are in certain containers, remain the same;
    # otherwise, convert to .mkv
    if fex.lower() not in [".mp4", ".mov", ".avi"]:
        fex = ".mkv"
    # set target name, a new file in the same directory as the first media file
    if target_name is None:
        target_name = of + "_merged" + fex.lower()
    if not overwrite:
        target_name = generate_outfilename(target_name)

    total_length = sum([get_length(media) for media in media_paths])

    cmd = [
        "ffmpeg",
        "-y" if overwrite else "-n",
        "-f", "concat",
        "-safe", "0",
        "-i", txt_path,
        "-c", "copy",
        target_name,
    ]
    ffmpeg_cmd(cmd, total_length, pb_prefix="Merging videos:", print_cmd=print_cmd)

    # remove tmp.txt
    os.remove(txt_path)

    return target_name