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	float	Total iterations, or a duration in seconds. 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

def __init__(
        self,
        total: float = 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 (float, optional): Total iterations, or a duration in seconds. 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

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

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 bool(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

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

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

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

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

def __init__(self, figure=None, figure_type: str | None = None, data: Any = None,
             layers: list | None = 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

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

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

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

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)

deprecated_alias

deprecated_alias(old, new)

A property that reads and writes new under the retired name old.

Renaming a result attribute is a breaking change, because show(key=...) and user code both reach these by name. The old name keeps working through one release and warns, and it is a property rather than a second attribute so the two names cannot drift apart when either side is reassigned.

Parameters:

Name	Type	Description	Default
old	str	The retired attribute name, used only in the warning.	required
new	str	The canonical attribute this proxies to.	required

Returns:

Name	Type	Description
property	property	Assign it in a class body as old_name = deprecated_alias(...).

Source code in musicalgestures/_utils.py

def deprecated_alias(old: str, new: str) -> property:
    """A property that reads and writes `new` under the retired name `old`.

    Renaming a result attribute is a breaking change, because `show(key=...)`
    and user code both reach these by name. The old name keeps working through
    one release and warns, and it is a property rather than a second attribute
    so the two names cannot drift apart when either side is reassigned.

    Args:
        old (str): The retired attribute name, used only in the warning.
        new (str): The canonical attribute this proxies to.

    Returns:
        property: Assign it in a class body as ``old_name = deprecated_alias(...)``.
    """
    message = f"{old} is deprecated and will be removed in 2.0; use {new}."

    def getter(self):
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        try:
            return getattr(self, new)
        except AttributeError:
            # report the name that was asked for, so hasattr(obj, old) is False
            # rather than raising about an attribute the caller never mentioned
            raise AttributeError(
                f"{type(self).__name__!r} object has no attribute {old!r}") from None

    def setter(self, value):
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        setattr(self, new, value)

    def deleter(self):
        warnings.warn(message, DeprecationWarning, stacklevel=2)
        delattr(self, new)

    getter.__doc__ = f"Deprecated alias for :attr:`{new}`. Removed in 2.0."
    return property(getter, setter, deleter)

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

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

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

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

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

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

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

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	float	Frames per second. Not an integer: 29.97 and 23.976 are ordinary.	required

Returns:

Name	Type	Description
int	int	The rounded millisecond value of the input frame index.

Source code in musicalgestures/_utils.py

def frame2ms(frame: int, fps: float) -> int:
    """
    Converts frames to milliseconds.

    Args:
        frame (int): The index of the frame to be converted to milliseconds.
        fps (float): Frames per second. Not an integer: 29.97 and 23.976 are ordinary.

    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

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

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

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

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

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

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

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

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

def extract_frame(
    filename: str,
    frame: int | None = None,
    time: Union[str, float] | None = None,
    target_name: str | None = 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"
        else:
            # the guards above leave exactly one of frame/time set, so this is the time case
            t = cast(Union[str, float], time)
            time = t if isinstance(t, str) else datetime.datetime.fromtimestamp(t-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]
    else:
        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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

framecount_timeout

framecount_timeout(filename, fast=True)

How long to let ffprobe count frames in this file, in seconds.

Scaled to the file, because a flat value is a ceiling on video length. Separate from get_framecount so a test can shorten it and exercise the timeout path against a real ffprobe rather than against a mock of one.

Source code in musicalgestures/_utils.py

def framecount_timeout(filename: str, fast: bool = True) -> float:
    """How long to let ffprobe count frames in this file, in seconds.

    Scaled to the file, because a flat value is a ceiling on video length. Separate
    from `get_framecount` so a test can shorten it and exercise the timeout path
    against a real ffprobe rather than against a mock of one.
    """
    import os
    try:
        gb = os.path.getsize(filename) / 1e9
    except OSError:
        gb = 0.0
    return max(60.0, 120.0 * gb) if fast else max(300.0, 600.0 * gb)

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

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)
    #: THE TIMEOUT USED TO BE A FLAT 10 SECONDS, which put a hard ceiling on how long
    #: a video MGT could open at all --- nothing to do with any analysis. Counting
    #: packets on a 2 h 38 min 1080p file takes 22 s, so it timed out; the fallback
    #: then escalated to `-count_frames`, which fully DECODES and is far slower, so
    #: that timed out too and the file could not be read. Scaled to the file now: this
    #: is one ffprobe pass per video, not per frame.
    timeout = framecount_timeout(filename, fast)

    try:
        out, err = process.communicate(timeout=timeout)
    except subprocess.TimeoutExpired:
        process.kill()
        process.communicate()
        #: Do NOT escalate to the slower method on a timeout. The container's own
        #: count is instant, and an approximate answer beats refusing the file.
        fallback = _container_framecount(filename)
        if fallback:
            import warnings as _w
            _w.warn(
                f"counting frames timed out after {timeout:.0f}s; using the "
                f"container's nb_frames ({fallback}). That figure is off by one on "
                f"some AVIs and absent on some WebM, so treat the last frame as "
                f"uncertain.", RuntimeWarning, stacklevel=2)
            return fallback
        out, err = "", None

    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)
            elif _container_framecount(filename):
                return _container_framecount(filename)  # type: ignore[return-value]
            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)
        elif _container_framecount(filename):
            return _container_framecount(filename)  # type: ignore[return-value]
        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

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

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

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

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

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

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

Returns:

Name	Type	Description
Any	Any	What comes back depends on pipe, and the caller knows which it asked for. 'read' and 'write' give a subprocess.Popen to read frames from or write frames to, 'load' gives the subprocess.CompletedProcess of a finished run, and the default of None runs the command to completion behind a progress bar and returns nothing. Annotating that precisely would need overloads on the value of a string argument, which would describe the signature more exactly than it deserves.

Raises:

Type	Description
KeyboardInterrupt	If the user stops the process.
FFmpegError	If the ffmpeg process was unsuccessful.

Source code in musicalgestures/_utils.py

def ffmpeg_cmd(command: list, total_time: float, pb_prefix: str = 'Progress', print_cmd: bool = False, stream: bool = True, pipe: str | None = None) -> Any:
    """
    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.

    Returns:
        Any: What comes back depends on `pipe`, and the caller knows which it
            asked for. `'read'` and `'write'` give a `subprocess.Popen` to read
            frames from or write frames to, `'load'` gives the
            `subprocess.CompletedProcess` of a finished run, and the default of
            None runs the command to completion behind a progress bar and
            returns nothing. Annotating that precisely would need overloads on
            the value of a string argument, which would describe the signature
            more exactly than it deserves.

    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', '-']
        completed = subprocess.run(command, stdout=subprocess.PIPE, bufsize=-1)
        return completed

    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:
        proc = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, universal_newlines=True)
        # stdout=PIPE guarantees a stream; the annotation is Optional for the general case
        assert proc.stdout is not None, "ffmpeg was started without a readable output stream"
        returncode = None
        all_out = ''

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

                if out == '':
                    proc.wait()
                    returncode = proc.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:
                proc.terminate()
            except OSError:
                pass
            proc.wait()
            raise KeyboardInterrupt

stop_ffmpeg_process

stop_ffmpeg_process(process, timeout=5.0)

Stop a piped ffmpeg from ffmpeg_cmd and reap it, unconditionally.

terminate() alone is not enough for a process on the other end of a pipe. A reader that stops before end of stream leaves ffmpeg blocked in a write to the full pipe; the blocked write restarts around SIGTERM, so the graceful handler never runs, and an unreaped Popen parks itself in subprocess._active, which keeps the pipe's read end open for the life of the calling process. The result is a live ffmpeg per stopped stream, invisible until they are counted in gigabytes.

Closing the pipes first is what actually stops it: the next write dies of EPIPE whatever the signal handling does. The kill-after-timeout is a backstop for a process wedged somewhere other than a pipe write.

Parameters:

Name	Type	Description	Default
process		The subprocess.Popen returned by ffmpeg_cmd(pipe='read') or pipe='write'. For a write pipe, flush and close stdin first if the buffered tail matters; this discards it.	required
timeout	float	Seconds to wait after SIGTERM before SIGKILL. Defaults to 5.0.	5.0

Source code in musicalgestures/_utils.py

def stop_ffmpeg_process(process, timeout: float = 5.0) -> None:
    """Stop a piped ffmpeg from `ffmpeg_cmd` and reap it, unconditionally.

    `terminate()` alone is not enough for a process on the other end of a pipe.
    A reader that stops before end of stream leaves ffmpeg blocked in a write to
    the full pipe; the blocked write restarts around SIGTERM, so the graceful
    handler never runs, and an unreaped Popen parks itself in `subprocess._active`,
    which keeps the pipe's read end open for the life of the calling process. The
    result is a live ffmpeg per stopped stream, invisible until they are counted
    in gigabytes.

    Closing the pipes first is what actually stops it: the next write dies of
    EPIPE whatever the signal handling does. The kill-after-timeout is a backstop
    for a process wedged somewhere other than a pipe write.

    Args:
        process: The `subprocess.Popen` returned by `ffmpeg_cmd(pipe='read')` or
            `pipe='write'`. For a write pipe, flush and close `stdin` first if the
            buffered tail matters; this discards it.
        timeout (float, optional): Seconds to wait after SIGTERM before SIGKILL.
            Defaults to 5.0.
    """
    import subprocess

    for stream in (process.stdin, process.stdout, process.stderr):
        if stream is not None:
            try:
                stream.close()
            except OSError:
                pass
    try:
        process.terminate()
    except OSError:
        pass
    try:
        process.wait(timeout=timeout)
    except subprocess.TimeoutExpired:
        process.kill()
        process.wait()

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

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

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

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

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 int(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

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

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

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())  # type: ignore[name-defined]  # IPython injects this into builtins; NameError is caught below
    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

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__  # type: ignore[name-defined]  # IPython injects this into builtins; NameError is caught below
        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

def merge_videos(
    media_paths: list[str], target_name: str | None = 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