For more information about the examples, such as how the Python and Mojo files interact with each other, see the Examples Overview

SpectralFreezeExample

Python Code

import sys
from pathlib import Path

# This example is able to run by pressing the "play" button in VSCode
# that executes the whole file.
# In order to do this, it needs to add the parent directory to the path
# (the next line here) so that it can find the mmm_src and mmm_utils packages.
# If you want to run it line by line in a REPL, skip this line!
sys.path.insert(0, str(Path(__file__).parent.parent))

from mmm_src.MMMAudio import MMMAudio
from PySide6.QtWidgets import QApplication, QWidget, QVBoxLayout, QPushButton

mmm_audio = MMMAudio(256, graph_name="SpectralFreezeExample", package_name="examples")
mmm_audio.start_audio()

app = QApplication([])

# Create the main window
window = QWidget()
window.setWindowTitle("Spectral Freeze Controller")
window.resize(100, 100)
# stop audio when window is closed
window.closeEvent = lambda event: (mmm_audio.stop_audio(), event.accept())

# Create layout
layout = QVBoxLayout()

gatebutton = QPushButton("Freeze Gate")
gatebutton.setCheckable(True)
gatebutton.setChecked(False)
gatebutton.clicked.connect(lambda checked: mmm_audio.send_bool("freeze_gate", checked))
layout.addWidget(gatebutton)
window.setLayout(layout)
window.show()
app.exec()

Mojo Code

from mmm_src.MMMWorld import *
from mmm_utils.Messenger import Messenger
from mmm_dsp.PlayBuf import PlayBuf
from mmm_utils.functions import select
from mmm_dsp.FFTProcess import *
from mmm_utils.Windows import WindowTypes
from mmm_dsp.Env import ASREnv
from random import random_float64

alias two_pi = 2.0 * pi

struct SpectralFreezeWindow[window_size: Int](FFTProcessable):
    var world: UnsafePointer[MMMWorld]
    var m: Messenger
    var bin: Int64
    var freeze_gate: Bool
    var stored_phases: List[SIMD[DType.float64, 2]]
    var stored_mags: List[SIMD[DType.float64, 2]]

    fn __init__(out self, world: UnsafePointer[MMMWorld], namespace: Optional[String] = None):
        self.world = world
        self.bin = (window_size // 2) + 1
        self.m = Messenger(world, namespace)
        self.freeze_gate = False
        self.stored_phases = [SIMD[DType.float64, 2](0.0) for _ in range(window_size)]
        self.stored_mags = [SIMD[DType.float64, 2](0.0) for _ in range(window_size)]

    fn get_messages(mut self) -> None:
        self.m.update(self.freeze_gate, "freeze_gate")

    fn next_stereo_frame(mut self, mut mags: List[SIMD[DType.float64, 2]], mut phases: List[SIMD[DType.float64, 2]]) -> None:
        if not self.freeze_gate:
            # self.stored_phases = phases.copy()
            self.stored_mags = mags.copy()
        else:
            mags = self.stored_mags.copy()
        for i in range(window_size):
            phases[i] += SIMD[DType.float64, 2](random_float64(0, two_pi), random_float64(0, two_pi))


struct SpectralFreeze[window_size: Int](Movable, Copyable):
    """
     Spectral Freeze

    """

    alias hop_size = window_size // 4
    var world: UnsafePointer[MMMWorld]
    var freeze: FFTProcess[SpectralFreezeWindow[window_size],window_size,Self.hop_size,WindowTypes.hann,WindowTypes.hann]
    var m: Messenger
    var freeze_gate: Bool
    var asr: ASREnv

    fn __init__(out self, world: UnsafePointer[MMMWorld], namespace: Optional[String] = None):
        self.world = world
        self.freeze = FFTProcess[
                SpectralFreezeWindow[window_size],
                window_size,
                self.hop_size,
                WindowTypes.hann,
                WindowTypes.hann
            ](self.world,process=SpectralFreezeWindow[window_size](self.world, namespace))
        self.m = Messenger(world, namespace)
        self.freeze_gate = False
        self.asr = ASREnv(world)

    fn next(mut self, sample: SIMD[DType.float64, 2]) -> SIMD[DType.float64, 2]:
        self.m.update(self.freeze_gate, "freeze_gate")
        env = self.asr.next(0.01, 1.0, 0.01, self.freeze_gate, 1.0)
        freeze = self.freeze.next_stereo(sample)
        return select(env, [sample, freeze]) * 0.3

# this really should have a window size of 8192 or more, but the numpy FFT seems to barf on this
alias window_size = 2048

struct SpectralFreezeExample(Movable, Copyable):
    var world: UnsafePointer[MMMWorld]
    var buffer: Buffer
    var play_buf: PlayBuf   
    var spectral_freeze: SpectralFreeze[window_size]
    var m: Messenger
    var stereo_switch: Bool

    fn __init__(out self, world: UnsafePointer[MMMWorld], namespace: Optional[String] = None):
        self.world = world
        self.buffer = Buffer("resources/Shiverer.wav")
        self.play_buf = PlayBuf(world) 
        self.spectral_freeze = SpectralFreeze[window_size](world)
        self.m = Messenger(world)
        self.stereo_switch: Bool = False

    fn next(mut self) -> SIMD[DType.float64,2]:
        self.m.update(self.stereo_switch,"stereo_switch")

        out = self.play_buf.next[2](self.buffer, 0, 1)

        out = self.spectral_freeze.next(out)

        return out