util API

maai.util

candidate_variants(mode, mc, mono)

List the checkpoint variants to try for a public mode, best first.

Parameters:

Name	Type	Description	Default
mode	str	Public mode ('vap', 'vad', 'bc_det', 'bc', ...).	required
mc	bool	Prefer the noise-robust (multi-condition) variant.	required
mono	bool	Prefer the single-channel variant.	required

Returns:

Type	Description
list	List[str]: Variant names, e.g. ['vap_mc_mono', 'vap_mono', 'vap_mc', 'vap'].

Source code in src/maai/util.py

def candidate_variants(mode: str, mc: bool, mono: bool) -> list:
    """List the checkpoint variants to try for a public mode, best first.

    Args:
        mode (str): Public mode ('vap', 'vad', 'bc_det', 'bc', ...).
        mc (bool): Prefer the noise-robust (multi-condition) variant.
        mono (bool): Prefer the single-channel variant.

    Returns:
        List[str]: Variant names, e.g. ``['vap_mc_mono', 'vap_mono', 'vap_mc', 'vap']``.
    """
    stereo = [mode]
    if mc and mode in MC_CAPABLE_MODES:
        stereo = [f"{mode}_mc", mode]
    if mono and mode in MONO_CAPABLE_MODES:
        return [mono_variant(v) for v in stereo] + stereo
    return stereo

conv_2float_2_byte(val1, val2)

Convert two double-precision floats into a combined byte array.

Parameters:

Name	Type	Description	Default
val1	float	First float.	required
val2	float	Second float.	required

Returns:

Name	Type	Description
bytes		Combined byte array.

Source code in src/maai/util.py

def conv_2float_2_byte(val1, val2):
    """Convert two double-precision floats into a combined byte array.

    Args:
        val1 (float): First float.
        val2 (float): Second float.

    Returns:
        bytes: Combined byte array.
    """
    b1 = struct.pack('<d', val1)
    b2 = struct.pack('<d', val2)

    b = b1 + b2

    return b

conv_2int16_2_byte(val1, val2)

Convert two integers into a combined byte array using 2 bytes each.

Parameters:

Name	Type	Description	Default
val1	int	First integer.	required
val2	int	Second integer.	required

Returns:

Name	Type	Description
bytes		Combined byte array.

Source code in src/maai/util.py

def conv_2int16_2_byte(val1, val2):
    """Convert two integers into a combined byte array using 2 bytes each.

    Args:
        val1 (int): First integer.
        val2 (int): Second integer.

    Returns:
        bytes: Combined byte array.
    """
    b1 = val1.to_bytes(2, BYTE_ORDER)
    b2 = val2.to_bytes(2, BYTE_ORDER)

    # print(b1)
    # print(b2)
    # concatenate two bytes
    b = b1 + b2

    #print(b)

    return b

conv_byte_2_2float(b1, b2)

Convert two double-precision byte blocks back into floats.

Parameters:

Name	Type	Description	Default
b1	bytes	First byte block.	required
b2	bytes	Second byte block.	required

Returns:

Type	Description
	Tuple[float, float]: The decoded floats.

Source code in src/maai/util.py

def conv_byte_2_2float(b1, b2):
    """Convert two double-precision byte blocks back into floats.

    Args:
        b1 (bytes): First byte block.
        b2 (bytes): Second byte block.

    Returns:
        Tuple[float, float]: The decoded floats.
    """
    val1 = struct.unpack('<d', b1)[0]
    val2 = struct.unpack('<d', b2)[0]

    return val1, val2

conv_bytearray_2_vapresult(barr)

Deserialize a byte array back into a VAP result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded VAP result data.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult(barr):
    """Deserialize a byte array back into a VAP result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded VAP result data.
    """
    idx = 0
    t = struct.unpack('<d', barr[idx:8])[0]
    idx += 8

    len_x1 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x1 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x1])
    idx += 8*len_x1

    len_x2 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x2 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x2])
    idx += 8 * len_x2

    len_p_now = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    p_now = conv_bytearray_2_floatarray(barr[idx:idx+8*len_p_now])
    idx += 8*len_p_now

    len_p_future = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    p_future = conv_bytearray_2_floatarray(barr[idx:idx+8*len_p_future])
    idx += 8*len_p_future

    len_vad = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    vad = conv_bytearray_2_floatarray(barr[idx:idx+8*len_vad])
    idx += 8*len_vad

    result_vap = {
        't': t,
        'x1': x1,
        'x2': x2,
        'p_now': p_now,
        'p_future': p_future,
        'vad': vad
    }

    return result_vap

conv_bytearray_2_vapresult_bc_det(barr)

Deserialize a byte array back into a BC detection result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded BC detection result data.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult_bc_det(barr):
    """Deserialize a byte array back into a BC detection result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded BC detection result data.
    """
    idx = 0
    t = struct.unpack('<d', barr[idx:8])[0]
    idx += 8

    len_x1 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x1 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x1])
    idx += 8*len_x1

    len_x2 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x2 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x2])
    idx += 8*len_x2

    len_bc_det = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    p_bc_det = conv_bytearray_2_floatarray(barr[idx:idx+8*len_bc_det])
    idx += 8*len_bc_det

    result_vap = {
        't': t,
        'x1': x1,
        'x2': x2,
        'p_bc_det': p_bc_det
    }

    return result_vap

conv_bytearray_2_vapresult_bc_det_mono(barr)

Deserialize a byte array back into a mono BC detection result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded mono BC detection result data with a
	scalar p_bc_det.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult_bc_det_mono(barr):
    """Deserialize a byte array back into a mono BC detection result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded mono BC detection result data with a
        scalar ``p_bc_det``.
    """
    result_vap = conv_bytearray_2_vapresult_bc_det(barr)
    result_vap['p_bc_det'] = result_vap['p_bc_det'][0]
    return result_vap

conv_bytearray_2_vapresult_mono(barr)

Deserialize a byte array back into a mono VAP result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded mono VAP result data with scalar
	p_now / p_future / vad.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult_mono(barr):
    """Deserialize a byte array back into a mono VAP result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded mono VAP result data with scalar
        ``p_now`` / ``p_future`` / ``vad``.
    """
    result_vap = conv_bytearray_2_vapresult(barr)
    for k in ('p_now', 'p_future', 'vad'):
        result_vap[k] = result_vap[k][0]
    return result_vap

conv_bytearray_2_vapresult_vad(barr)

Deserialize a byte array back into a VAD result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded VAD result data.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult_vad(barr):
    """Deserialize a byte array back into a VAD result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded VAD result data.
    """
    idx = 0
    t = struct.unpack('<d', barr[idx:8])[0]
    idx += 8

    len_x1 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x1 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x1])
    idx += 8*len_x1

    len_x2 = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    x2 = conv_bytearray_2_floatarray(barr[idx:idx+8*len_x2])
    idx += 8*len_x2

    len_vad = struct.unpack('<I', barr[idx:idx+4])[0]
    idx += 4
    vad = conv_bytearray_2_floatarray(barr[idx:idx+8*len_vad])
    idx += 8*len_vad

    result_vap = {
        't': t,
        'x1': x1,
        'x2': x2,
        'vad': vad
    }

    return result_vap

conv_bytearray_2_vapresult_vad_mono(barr)

Deserialize a byte array back into a mono VAD result dictionary.

Parameters:

Name	Type	Description	Default
barr	bytes	Serialized byte array.	required

Returns:

Type	Description
	Dict[str, Any]: The decoded mono VAD result data with a scalar vad.

Source code in src/maai/util.py

def conv_bytearray_2_vapresult_vad_mono(barr):
    """Deserialize a byte array back into a mono VAD result dictionary.

    Args:
        barr (bytes): Serialized byte array.

    Returns:
        Dict[str, Any]: The decoded mono VAD result data with a scalar ``vad``.
    """
    result_vap = conv_bytearray_2_vapresult_vad(barr)
    result_vap['vad'] = result_vap['vad'][0]
    return result_vap

conv_vapresult_2_bytearray(vap_result)

Serialize a VAP result dictionary into a byte array.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	VAP result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray(vap_result):
    """Serialize a VAP result dictionary into a byte array.

    Args:
        vap_result (Dict[str, Any]): VAP result data.

    Returns:
        bytes: The serialized byte array.
    """
    b = b''
    #print(type(vap_result['t']))
    b += struct.pack('<d', vap_result['t'])

    b += len(vap_result['x1']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x1'])

    b += len(vap_result['x2']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x2'])

    b += len(vap_result['p_now']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['p_now'])

    b += len(vap_result['p_future']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['p_future'])

    b += len(vap_result['vad']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['vad'])

    return b

conv_vapresult_2_bytearray_bc_det(vap_result)

Serialize a BC detection result dictionary into a byte array.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	BC detection result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray_bc_det(vap_result):
    """Serialize a BC detection result dictionary into a byte array.

    Args:
        vap_result (Dict[str, Any]): BC detection result data.

    Returns:
        bytes: The serialized byte array.
    """
    b = b''
    b += struct.pack('<d', vap_result['t'])

    b += len(vap_result['x1']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x1'])

    b += len(vap_result['x2']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x2'])

    b += len(vap_result['p_bc_det']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['p_bc_det'])

    return b

conv_vapresult_2_bytearray_bc_det_mono(vap_result)

Serialize a mono BC detection result dictionary into a byte array.

The scalar p_bc_det value is wrapped into a length-1 array so the wire format stays identical to the two-channel one.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	Mono BC detection result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray_bc_det_mono(vap_result):
    """Serialize a mono BC detection result dictionary into a byte array.

    The scalar ``p_bc_det`` value is wrapped into a length-1 array so the
    wire format stays identical to the two-channel one.

    Args:
        vap_result (Dict[str, Any]): Mono BC detection result data.

    Returns:
        bytes: The serialized byte array.
    """
    wrapped = dict(vap_result)
    wrapped['p_bc_det'] = [vap_result['p_bc_det']]
    return conv_vapresult_2_bytearray_bc_det(wrapped)

conv_vapresult_2_bytearray_mono(vap_result)

Serialize a mono VAP result dictionary into a byte array.

Scalar p_now / p_future / vad values are wrapped into length-1 arrays so the wire format stays identical to the stereo one.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	Mono VAP result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray_mono(vap_result):
    """Serialize a mono VAP result dictionary into a byte array.

    Scalar ``p_now`` / ``p_future`` / ``vad`` values are wrapped into
    length-1 arrays so the wire format stays identical to the stereo one.

    Args:
        vap_result (Dict[str, Any]): Mono VAP result data.

    Returns:
        bytes: The serialized byte array.
    """
    wrapped = dict(vap_result)
    wrapped['p_now'] = [vap_result['p_now']]
    wrapped['p_future'] = [vap_result['p_future']]
    wrapped['vad'] = [vap_result['vad']]
    return conv_vapresult_2_bytearray(wrapped)

conv_vapresult_2_bytearray_vad(vap_result)

Serialize a VAD result dictionary into a byte array.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	VAD result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray_vad(vap_result):
    """Serialize a VAD result dictionary into a byte array.

    Args:
        vap_result (Dict[str, Any]): VAD result data.

    Returns:
        bytes: The serialized byte array.
    """
    b = b''
    b += struct.pack('<d', vap_result['t'])

    b += len(vap_result['x1']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x1'])

    b += len(vap_result['x2']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['x2'])

    b += len(vap_result['vad']).to_bytes(4, BYTE_ORDER)
    b += conv_floatarray_2_byte(vap_result['vad'])

    return b

conv_vapresult_2_bytearray_vad_mono(vap_result)

Serialize a mono VAD result dictionary into a byte array.

The scalar vad value is wrapped into a length-1 array so the wire format stays identical to the two-channel one.

Parameters:

Name	Type	Description	Default
vap_result	Dict[str, Any]	Mono VAD result data.	required

Returns:

Name	Type	Description
bytes		The serialized byte array.

Source code in src/maai/util.py

def conv_vapresult_2_bytearray_vad_mono(vap_result):
    """Serialize a mono VAD result dictionary into a byte array.

    The scalar ``vad`` value is wrapped into a length-1 array so the wire
    format stays identical to the two-channel one.

    Args:
        vap_result (Dict[str, Any]): Mono VAD result data.

    Returns:
        bytes: The serialized byte array.
    """
    wrapped = dict(vap_result)
    wrapped['vad'] = [vap_result['vad']]
    return conv_vapresult_2_bytearray_vad(wrapped)

download_continuous_mimi_onnx(precision='fp32', cache_dir=None, force_download=False, frames_per_call=1, variant=None)

Resolve paths to the streaming Mimi ONNX model and JSON sidecar on disk.

Files are fetched from maai-kyoto/continuous-mimi-onnx via hf_hub_download (cached under the usual Hugging Face cache layout, or under cache_dir when set).

frames_per_call selects the microbatch export contract: 1 (default) is the original one-frame-per-call export (continuous_mimi_{precision}.onnx); N > 1 selects the N-frames-per-call export (continuous_mimi_{precision}_{N}f.onnx), which produces N VAP frames per ONNX call.

variant selects a fine-tuned Mimi (e.g. "ft-zoom-tri" -> continuous_mimi_ft-zoom-tri_{precision}.onnx). None is the original Mimi.

Source code in src/maai/util.py

def download_continuous_mimi_onnx(
    precision: str = "fp32",
    cache_dir: str | None = None,
    force_download: bool = False,
    frames_per_call: int = 1,
    variant: str | None = None,
) -> tuple[str, str]:
    """
    Resolve paths to the streaming Mimi ONNX model and JSON sidecar on disk.

    Files are fetched from ``maai-kyoto/continuous-mimi-onnx`` via ``hf_hub_download``
    (cached under the usual Hugging Face cache layout, or under ``cache_dir`` when set).

    ``frames_per_call`` selects the microbatch export contract: 1 (default) is the
    original one-frame-per-call export (``continuous_mimi_{precision}.onnx``); N > 1
    selects the N-frames-per-call export (``continuous_mimi_{precision}_{N}f.onnx``),
    which produces N VAP frames per ONNX call.

    ``variant`` selects a fine-tuned Mimi (e.g. ``"ft-zoom-tri"`` ->
    ``continuous_mimi_ft-zoom-tri_{precision}.onnx``). None is the original Mimi.
    """
    precision = str(precision).strip().lower()
    frames_per_call = int(frames_per_call)
    if frames_per_call < 1:
        raise ValueError("frames_per_call must be at least 1.")
    if precision not in ("fp32", "int8"):
        raise ValueError(f"Unsupported precision for continuous Mimi ONNX: {precision}")
    suffix = "" if frames_per_call == 1 else f"_{frames_per_call}f"
    if (variant, precision, frames_per_call) in CONTINUOUS_MIMI_ONNX_VARIANT_FILES:
        onnx_fn, meta_fn = CONTINUOUS_MIMI_ONNX_VARIANT_FILES[(variant, precision, frames_per_call)]
    else:
        prefix = "continuous_mimi_" if not variant else f"continuous_mimi_{variant}_"
        onnx_fn = f"{prefix}{precision}{suffix}.onnx"
        meta_fn = f"{prefix}{precision}{suffix}.json"

    onnx_path = hf_hub_download(
        repo_id=CONTINUOUS_MIMI_ONNX_REPO_ID,
        filename=onnx_fn,
        cache_dir=cache_dir,
        force_download=force_download,
    )
    meta_path = hf_hub_download(
        repo_id=CONTINUOUS_MIMI_ONNX_REPO_ID,
        filename=meta_fn,
        cache_dir=cache_dir,
        force_download=force_download,
    )
    return str(onnx_path), str(meta_path)

euler_to_quaternion(rx, ry, rz)

XYZ intrinsic Euler angles (radians) to quaternion (qx, qy, qz, qw).

Source code in src/maai/util.py

def euler_to_quaternion(rx: float, ry: float, rz: float) -> tuple[float, float, float, float]:
    """XYZ intrinsic Euler angles (radians) to quaternion (qx, qy, qz, qw)."""
    cx = math.cos(rx / 2.0)
    sx = math.sin(rx / 2.0)
    cy = math.cos(ry / 2.0)
    sy = math.sin(ry / 2.0)
    cz = math.cos(rz / 2.0)
    sz = math.sin(rz / 2.0)
    qx = sx * cy * cz + cx * sy * sz
    qy = cx * sy * cz - sx * cy * sz
    qz = cx * cy * sz + sx * sy * cz
    qw = cx * cy * cz - sx * sy * sz
    return (qx, qy, qz, qw)

generate_natural_nod(range_rad, count, use_pre_rise, velocity, fps=30, decay_rate=0.6, pre_rise_ratio=0.8)

Generate a natural nodding motion sequence (pitch in radians vs time in seconds).

Interpolation is cubic spline (CubicSpline) when scipy is available, else cosine interpolation between keyframes.

Parameters

range_rad : float Nod depth in radians (absolute value). count : int Number of nods (>= 1). use_pre_rise : bool Whether to include a pre-rise before the first nod. velocity : float Target average angular velocity (rad/s). fps : int Output frame rate. decay_rate : float Amplitude decay per nod (0–1). pre_rise_ratio : float Pre-rise amplitude as a ratio of range_rad.

Returns

motion : np.ndarray Pitch values in radians, one per frame. time_axis : np.ndarray Time stamps in seconds, same length as motion.

Source code in src/maai/util.py

def generate_natural_nod(
    range_rad: float,
    count: int,
    use_pre_rise: bool,
    velocity: float,
    fps: int = 30,
    decay_rate: float = 0.6,
    pre_rise_ratio: float = 0.8,
) -> tuple[np.ndarray, np.ndarray]:
    """Generate a natural nodding motion sequence (pitch in radians vs time in seconds).

    Interpolation is cubic spline (``CubicSpline``) when scipy is available,
    else cosine interpolation between keyframes.

    Parameters
    ----------
    range_rad : float
        Nod depth in radians (absolute value).
    count : int
        Number of nods (>= 1).
    use_pre_rise : bool
        Whether to include a pre-rise before the first nod.
    velocity : float
        Target average angular velocity (rad/s).
    fps : int
        Output frame rate.
    decay_rate : float
        Amplitude decay per nod (0–1).
    pre_rise_ratio : float
        Pre-rise amplitude as a ratio of *range_rad*.

    Returns
    -------
    motion : np.ndarray
        Pitch values in radians, one per frame.
    time_axis : np.ndarray
        Time stamps in seconds, same length as *motion*.
    """
    count = max(1, int(count))
    fps = max(1, int(fps))

    keyframe_vals: list[float] = [0.0]
    current_amp = abs(float(range_rad))

    if use_pre_rise:
        keyframe_vals.extend(
            [
                current_amp * float(pre_rise_ratio),
                -current_amp,
                0.0,
            ]
        )
    else:
        keyframe_vals.extend([-current_amp, 0.0])

    for _ in range(count - 1):
        current_amp *= float(decay_rate)
        keyframe_vals.extend([-current_amp, 0.0])

    n_seg = len(keyframe_vals) - 1
    distances: list[float] = []
    vel_scales: list[float] = []
    for i in range(n_seg):
        d = abs(keyframe_vals[i + 1] - keyframe_vals[i])
        distances.append(d)
        if d < 1e-6:
            vel_scales.append(1.0)
        else:
            if use_pre_rise:
                nod_idx = 0 if i < 3 else 1 + (i - 3) // 2
            else:
                nod_idx = i // 2
            vel_scales.append(float(decay_rate) ** (nod_idx * 0.5))

    total_distance = sum(distances)
    raw_total = sum(d / vs for d, vs in zip(distances, vel_scales))
    total_time = total_distance / float(velocity) if float(velocity) > 1e-9 else 0.0

    keyframe_frames: list[int] = [0]
    for d, vs in zip(distances, vel_scales):
        if raw_total > 1e-9 and d >= 1e-6:
            duration = (d / vs) * total_time / raw_total
            n_frames = max(1, int(round(duration * fps)))
        else:
            n_frames = 0
        keyframe_frames.append(keyframe_frames[-1] + n_frames)

    keyframe_times = [f / fps for f in keyframe_frames]
    total_frames = keyframe_frames[-1]

    if total_frames <= 0:
        return np.array([0.0]), np.array([0.0])

    time_axis = np.arange(total_frames) / fps

    motion = _nod_motion_cubic_spline_or_cosine(
        keyframe_times,
        keyframe_vals,
        keyframe_frames,
        time_axis,
    )
    return motion, time_axis

get_available_models()

Retrieve a dictionary of available pre-trained models from the Hugging Face hub.

Returns:

Type	Description
	Dict[str, List[str]]: A mapping of repository IDs to their available model files.

Source code in src/maai/util.py

def get_available_models():
    """Retrieve a dictionary of available pre-trained models from the Hugging Face hub.

    Returns:
        Dict[str, List[str]]: A mapping of repository IDs to their available model files.
    """
    available_models = {}
    for repo_id in repo_ids.values():
        files = list_repo_files(repo_id)
        available_models[repo_id] = [file for file in files if file.endswith(".pt")]
    return available_models

load_vap_model(mode, frame_rate, context_len_sec, language='jp', device='cpu', cache_dir=None, force_download=False, model_type='normal')

Load a pretrained VAP model from the Hugging Face hub.

Parameters:

Name	Type	Description	Default
mode	str	The checkpoint variant to load (e.g., 'vap', 'vap_mc', 'vap_mc_mono', 'bc', 'nod'). See :func:resolve_and_load_model for picking the variant from a public mode.	required
frame_rate	float	The frame rate expected by the model.	required
context_len_sec	float	The context length in seconds.	required
language	str	The language identifier for the model (e.g., 'jp', 'en').	'jp'
device	str	The device to load the model onto ('cpu', 'cuda').	'cpu'
cache_dir	str	Directory to cache the downloaded model.	None
force_download	bool	If True, forces download even if cached.	False
model_type	str	The general model architecture type.	'normal'

Returns:

Type	Description
	Dict[str, Any]: The loaded state dictionary of the model.

Source code in src/maai/util.py

def load_vap_model(mode: str, frame_rate: float, context_len_sec: float, language: str = "jp", device: str = "cpu", cache_dir: str = None, force_download: bool = False, model_type: str = "normal"):
    """Load a pretrained VAP model from the Hugging Face hub.

    Args:
        mode (str): The checkpoint variant to load (e.g., 'vap', 'vap_mc', 'vap_mc_mono', 'bc', 'nod').
            See :func:`resolve_and_load_model` for picking the variant from a public mode.
        frame_rate (float): The frame rate expected by the model.
        context_len_sec (float): The context length in seconds.
        language (str): The language identifier for the model (e.g., 'jp', 'en').
        device (str): The device to load the model onto ('cpu', 'cuda').
        cache_dir (str, optional): Directory to cache the downloaded model.
        force_download (bool): If True, forces download even if cached.
        model_type (str): The general model architecture type.

    Returns:
        Dict[str, Any]: The loaded state dictionary of the model.
    """
    frame_rate_label = _format_frame_rate(frame_rate)
    encoder_type = resolve_encoder_type(model_type)
    encoder_suffix = ""
    if encoder_type == "mimi":
        encoder_suffix = "_mimi"
    elif encoder_type != "cpc":
        raise ValueError(f"Unsupported encoder_type for pretrained model lookup: {encoder_type}")

    if mode == "vap":
        if language == "jp":
            repo_id = repo_ids["vap_jp"]
            file_path = f"vap{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en":
            repo_id = repo_ids["vap_en"]
            file_path = f"vap{encoder_suffix}_state_dict_en_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ch":
            repo_id = repo_ids["vap_ch"]
            file_path = f"vap{encoder_suffix}_state_dict_ch_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "tri":
            repo_id = repo_ids["vap_tri"]
            file_path = f"vap{encoder_suffix}_state_dict_tri_ecj_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "jp_kyoto":
            repo_id = repo_ids["vap_jp_kyoto"]
            file_path = f"vap{encoder_suffix}_state_dict_jp_kyoto_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en_kyoto":
            repo_id = repo_ids["vap_en_kyoto"]
            file_path = f"vap{encoder_suffix}_state_dict_en_kyoto_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ch_kyoto":
            repo_id = repo_ids["vap_ch_kyoto"]
            file_path = f"vap{encoder_suffix}_state_dict_ch_kyoto_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "tri_kyoto":
            repo_id = repo_ids["vap_tri_kyoto"]
            file_path = f"vap{encoder_suffix}_state_dict_tri_kyoto_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ca":
            repo_id = repo_ids["vap_ca"]
            file_path = f"vap{encoder_suffix}_state_dict_ca_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "fr":
            repo_id = repo_ids["vap_fr"]
            file_path = f"vap{encoder_suffix}_state_dict_fr_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch", "tri", "jp_kyoto", "en_kyoto", "ch_kyoto", "tri_kyoto", "ca", "fr"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    # vap_mono is a single-channel model with its own checkpoints
    elif mode == "vap_mono":
        if language in ("jp", "en", "ch"):
            repo_id = repo_ids[f"vap_{language}"]
            file_path = f"vap_mono{encoder_suffix}_state_dict_{language}_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    # vap_mc is the noise-robust (multi-condition) model, stored with the vap checkpoints
    elif mode == "vap_mc":
        if language in ("jp", "en", "ch", "tri", "fr", "jp_kyoto", "en_kyoto", "ch_kyoto", "tri_kyoto"):
            repo_id = repo_ids[f"vap_{language}"]
            file_path = f"vap_mc{encoder_suffix}_state_dict_{language}_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch", "tri", "jp_kyoto", "en_kyoto", "ch_kyoto", "tri_kyoto", "fr"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    # vap_mc_mono is the noise-robust single-channel model, stored with the vap checkpoints
    elif mode == "vap_mc_mono":
        if language in ("jp", "en", "ch"):
            repo_id = repo_ids[f"vap_{language}"]
            file_path = f"vap_mono_mc{encoder_suffix}_state_dict_{language}_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"
        else:
            supported_languages = ["jp", "en", "ch"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "vad":
        if language == "jp":
            repo_id = repo_ids["vad_jp"]
            file_path = f"vad{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ch":
            repo_id = repo_ids["vad_ch"]
            file_path = f"vad{encoder_suffix}_state_dict_ch_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en":
            repo_id = repo_ids["vad_en"]
            file_path = f"vad{encoder_suffix}_state_dict_en_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "ch", "en"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "vad_mono":
        if language == "jp":
            repo_id = repo_ids["vad_jp"]
            file_path = f"vad_mono{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ch":
            repo_id = repo_ids["vad_ch"]
            file_path = f"vad_mono{encoder_suffix}_state_dict_ch_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en":
            repo_id = repo_ids["vad_en"]
            file_path = f"vad_mono{encoder_suffix}_state_dict_en_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "ch", "en"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "bc_det":
        if language in ("jp", "en", "ch"):
            repo_id = repo_ids[f"bc_det_{language}"]
            file_path = f"bc_det{encoder_suffix}_state_dict_{language}_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "bc_det_mono":
        if language in ("jp", "en", "ch"):
            repo_id = repo_ids[f"bc_det_{language}"]
            file_path = f"bc_det_mono{encoder_suffix}_state_dict_{language}_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "bc":
        if language == "jp":
            repo_id = repo_ids["vap_bc_jp"]
            file_path = f"vap-bc{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en":
            repo_id = repo_ids["vap_bc_en"]
            file_path = f"vap-bc{encoder_suffix}_state_dict_en_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "ch":
            repo_id = repo_ids["vap_bc_ch"]
            file_path = f"vap-bc{encoder_suffix}_state_dict_ch_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "tri":
            repo_id = repo_ids["vap_bc_tri"]
            file_path = f"vap-bc{encoder_suffix}_state_dict_tri_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "ch", "tri"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "bc_2type":

        if language == "jp":
            repo_id = repo_ids["vap_bc_2type_jp"]
            file_path = f"vap-bc-2type{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        # elif language == "en":
        #     repo_id = repo_ids["vap_bc_2type_en"]
        #     file_path = f"vap-bc_2type_state_dict_erica_{frame_rate}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "tri"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "nod":

        if language == "jp":
            repo_id = repo_ids["vap_nod_jp"]
            file_path = f"vap-nod{encoder_suffix}_state_dict_erica_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        elif language == "en":
            repo_id = repo_ids["vap_nod_en"]
            file_path = f"vap-nod{encoder_suffix}_state_dict_erica_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp", "en", "tri"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "nod_timing":
        # Timing-only nod: single merged occurrence gt_head (sigmoid), onset+-500ms
        # window label, vs. mode="nod"'s 4-way [none, short, long, long_p] softmax.
        if language == "jp":
            repo_id = repo_ids["vap_nod_timing_jp"]
            file_path = f"vap-nod{encoder_suffix}_state_dict_erica_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "vap_prompt":

        if language == "jp":
            repo_id = repo_ids["vap_prompt_jp"]
            file_path = f"vap_prompt{encoder_suffix}_state_dict_jp_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"

        else:
            supported_languages = ["jp"]
            raise ValueError(f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}")

    elif mode == "nod_para":
        if language != "jp":
            supported_languages = ["jp"]
            raise ValueError(
                f"Invalid language: {language}. Mode {mode} supports languages are: {supported_languages}"
            )
        repo_id = repo_ids["vap_nod_para_jp"]
        file_path = (
            f"vap-nod_para_state_dict_erica_{frame_rate_label}hz_{int(context_len_sec*1000)}msec.pt"
        )

    else:
        supported_modes = ["vap", "vap_mono", "vap_mc", "vap_mc_mono", "vad", "vad_mono", "bc_det", "bc_det_mono", "bc", "bc_2type", "nod", "nod_timing", "vap_prompt", "nod_para"]
        raise ValueError(f"Invalid mode: {mode}. Supported modes are: {supported_modes}")

    try:
        sd = hf_hub_download(repo_id=repo_id, filename=file_path, cache_dir=cache_dir, force_download=force_download)

    except Exception as e:
        raise ValueError(f"Invalid model: mode: {mode}, frame_rate: {frame_rate}, context_len_sec: {context_len_sec}, language: {language}. Run get_available_models() for available models.")

    sd = torch.load(sd, map_location=torch.device(device))

    return sd

load_vap_tri_model(language='jp', cache_dir=None, force_download=False)

Download and load the three-party VAP checkpoint used by MaaiTri.

Returns the checkpoint dict (state_dict, maai_tri_config, source).

Source code in src/maai/util.py

def load_vap_tri_model(language: str = "jp", cache_dir: str = None, force_download: bool = False):
    """Download and load the three-party VAP checkpoint used by MaaiTri.

    Returns the checkpoint dict
    (``state_dict``, ``maai_tri_config``, ``source``).
    """
    filename = f"vap_3party_mimi_state_dict_{language}_12.5hz_20000msec.pt"
    path = hf_hub_download(
        repo_id=VAP_3PARTY_REPO_ID.format(lang=language),
        filename=filename,
        cache_dir=cache_dir,
        force_download=force_download,
    )
    return torch.load(path, map_location="cpu")

mono_variant(mode)

Return the single-channel checkpoint variant name of a public mode.

Source code in src/maai/util.py

def mono_variant(mode: str) -> str:
    """Return the single-channel checkpoint variant name of a public mode."""
    return "vap_mc_mono" if mode == "vap_mc" else f"{mode}_mono"

resolve_and_load_model(mode, mc, mono, frame_rate, context_len_sec, language='jp', device='cpu', cache_dir=None, force_download=False, model_type='normal')

Pick the best available checkpoint variant for a public mode and load it.

Variants from :func:candidate_variants are tried in order, so a missing noise-robust model falls back to the standard one, and a missing single-channel model falls back to the 2-channel one (which is then fed a silent second channel).

Parameters:

Name	Type	Description	Default
mode	str	Public mode ('vap', 'vad', 'bc_det', 'bc', ...).	required
mc	bool	Prefer the noise-robust (multi-condition) variant.	required
mono	bool	Prefer the single-channel variant.	required
frame_rate	float	The frame rate expected by the model.	required
context_len_sec	float	The context length in seconds.	required
language	str	The language identifier for the model.	'jp'
device	str	The device to load the model onto.	'cpu'
cache_dir	str	Directory to cache the downloaded model.	None
force_download	bool	If True, forces download even if cached.	False
model_type	str	The general model architecture type.	'normal'

Returns:

Type	Description
	Tuple[str, Dict[str, Any]]: The chosen variant and its loaded state dictionary.

Source code in src/maai/util.py

def resolve_and_load_model(mode: str, mc: bool, mono: bool, frame_rate: float, context_len_sec: float, language: str = "jp", device: str = "cpu", cache_dir: str = None, force_download: bool = False, model_type: str = "normal"):
    """Pick the best available checkpoint variant for a public mode and load it.

    Variants from :func:`candidate_variants` are tried in order, so a missing
    noise-robust model falls back to the standard one, and a missing
    single-channel model falls back to the 2-channel one (which is then fed a
    silent second channel).

    Args:
        mode (str): Public mode ('vap', 'vad', 'bc_det', 'bc', ...).
        mc (bool): Prefer the noise-robust (multi-condition) variant.
        mono (bool): Prefer the single-channel variant.
        frame_rate (float): The frame rate expected by the model.
        context_len_sec (float): The context length in seconds.
        language (str): The language identifier for the model.
        device (str): The device to load the model onto.
        cache_dir (str, optional): Directory to cache the downloaded model.
        force_download (bool): If True, forces download even if cached.
        model_type (str): The general model architecture type.

    Returns:
        Tuple[str, Dict[str, Any]]: The chosen variant and its loaded state dictionary.
    """
    candidates = candidate_variants(mode, mc, mono)
    last_error = None
    for variant in candidates:
        try:
            sd = load_vap_model(variant, frame_rate, context_len_sec, language, device, cache_dir, force_download, model_type=model_type)
        except ValueError as e:
            last_error = e
            continue

        if mc and mode in MC_CAPABLE_MODES and "mc" not in variant:
            print(f"[Info] No noise-robust (mc) model for mode='{mode}', lang='{language}'; using the standard model.")
        if mono and mode in MONO_CAPABLE_MODES and not variant.endswith("_mono"):
            print(f"[Warning] No single-channel model for mode='{mode}', lang='{language}'; using the 2-channel model with a silent channel 2.")
        return variant, sd

    raise ValueError(f"No model found for mode: {mode} (tried {candidates}), language: {language}. {last_error}")

resolve_encoder_type(model_type='normal')

Resolve the encoder type based on the provided model type.

Parameters:

Name	Type	Description	Default
model_type	str	The type of model (e.g., 'normal', 'normal-ver2').	'normal'

Returns:

Name	Type	Description
str	str	The corresponding encoder type ('cpc' or 'mimi').

Source code in src/maai/util.py

def resolve_encoder_type(model_type: str = "normal") -> str:
    """Resolve the encoder type based on the provided model type.

    Args:
        model_type (str): The type of model (e.g., 'normal', 'normal-ver2').

    Returns:
        str: The corresponding encoder type ('cpc' or 'mimi').
    """
    try:
        return MODEL_TYPE_TO_ENCODER_TYPE[model_type]
    except KeyError as exc:
        supported_model_types = list(MODEL_TYPE_TO_ENCODER_TYPE.keys())
        raise ValueError(
            f"Unsupported model_type: {model_type}. Supported model_type values are: {supported_model_types}"
        ) from exc