Skip to content

Recording session

A recording session: capture to disk, optionally with a live draft transcript.

RecordingSession

Records to a new vault entry, optionally with a live draft transcript.

Source code in src/murmurvault/session.py 
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
class RecordingSession:
    """Records to a new vault entry, optionally with a live draft transcript."""

    def __init__(  # noqa: PLR0913
        self,
        vault: Vault,
        cfg: dict,
        *,
        title: str = "",
        folder: str = "",
        tags: list[str] | None = None,
        mic: bool = True,
        system: bool = True,
        live: bool = False,
        on_segment: Callable[[Segment], None] | None = None,
    ):
        """Create the recording entry and set up capture; nothing is recorded until :meth:`start`.

        Args:
            vault: The vault.
            cfg: The configuration.
            title: Recording title.
            folder: Destination folder.
            tags: Tags to apply.
            mic: Record the microphone.
            system: Record the computer's audio output.
            live: Transcribe a draft while recording.
            on_segment: Called with each live segment.
        """
        self.vault, self.cfg = vault, cfg
        self.rec = vault.new_recording(title=title, folder=folder, tags=tags)
        try:
            self.recorder = Recorder(self.rec.path, cfg["audio"], mic=mic, system=system)
            self.live: LiveTranscriber | None = None
            if live:
                self.live = LiveTranscriber(
                    live_engine(cfg),
                    on_segment or (lambda s: None),
                    language=cfg["transcription"].get("language") or None,
                    min_chunk_s=float(cfg["live"]["min_chunk_s"]),
                    max_chunk_s=float(cfg["live"]["max_chunk_s"]),
                )
                self.recorder.add_listener(self.live.listener)
        except Exception:
            vault.delete(self.rec)
            raise

    def start(self) -> None:
        """Start recording; on failure the empty recording entry is removed."""
        try:
            self.recorder.start()
        except Exception:
            if self.live:
                self.live.stop()
            self.vault.delete(self.rec)
            raise

    def elapsed(self) -> float:
        """Seconds since recording started.

        Returns:
            Elapsed time.
        """
        return self.recorder.elapsed()

    def stop(self) -> Recording:
        """Stop recording, save the tracks and the live draft, and index the draft.

        Returns:
            The saved recording.

        Raises:
            RuntimeError: If no audio was captured; the empty entry is removed.
        """
        tracks, offsets, length = self.recorder.stop()
        draft = self.live.stop() if self.live else []
        if not tracks:
            self.vault.delete(self.rec)
            raise RuntimeError("no audio was captured (check devices with `murmurvault devices`)")
        rec = self.rec
        rec.tracks, rec.track_offsets, rec.duration = tracks, offsets, round(length, 2)
        self.vault.save(rec)
        if self.live and draft:
            for s in draft:
                off = offsets.get(s.track or "", 0.0)
                s.start, s.end = round(s.start + off, 3), round(s.end + off, 3)
            draft.sort(key=lambda s: s.start)
            eng = self.live.engine
            self.vault.save_transcript(rec, "live", Transcript(engine=eng.name, model=eng.model, segments=draft))
            reindex_recording(self.vault, rec, self.cfg)
        return rec

__init__(vault, cfg, *, title='', folder='', tags=None, mic=True, system=True, live=False, on_segment=None)

Create the recording entry and set up capture; nothing is recorded until :meth:start.

Parameters:

Name Type Description Default
vault Vault

The vault.

 required
cfg dict

The configuration.

 required
title str

Recording title.

 ''
folder str

Destination folder.

 ''
tags list[str] | None

Tags to apply.

 None
mic bool

Record the microphone.

 True
system bool

Record the computer's audio output.

 True
live bool

Transcribe a draft while recording.

 False
on_segment Callable[[Segment], None] | None

Called with each live segment.

 None
Source code in src/murmurvault/session.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
def __init__(  # noqa: PLR0913
    self,
    vault: Vault,
    cfg: dict,
    *,
    title: str = "",
    folder: str = "",
    tags: list[str] | None = None,
    mic: bool = True,
    system: bool = True,
    live: bool = False,
    on_segment: Callable[[Segment], None] | None = None,
):
    """Create the recording entry and set up capture; nothing is recorded until :meth:`start`.

    Args:
        vault: The vault.
        cfg: The configuration.
        title: Recording title.
        folder: Destination folder.
        tags: Tags to apply.
        mic: Record the microphone.
        system: Record the computer's audio output.
        live: Transcribe a draft while recording.
        on_segment: Called with each live segment.
    """
    self.vault, self.cfg = vault, cfg
    self.rec = vault.new_recording(title=title, folder=folder, tags=tags)
    try:
        self.recorder = Recorder(self.rec.path, cfg["audio"], mic=mic, system=system)
        self.live: LiveTranscriber | None = None
        if live:
            self.live = LiveTranscriber(
                live_engine(cfg),
                on_segment or (lambda s: None),
                language=cfg["transcription"].get("language") or None,
                min_chunk_s=float(cfg["live"]["min_chunk_s"]),
                max_chunk_s=float(cfg["live"]["max_chunk_s"]),
            )
            self.recorder.add_listener(self.live.listener)
    except Exception:
        vault.delete(self.rec)
        raise

elapsed()

Seconds since recording started.

Returns:

Type Description
float

Elapsed time.
Source code in src/murmurvault/session.py 
87
88
89
90
91
92
93
def elapsed(self) -> float:
    """Seconds since recording started.

    Returns:
        Elapsed time.
    """
    return self.recorder.elapsed()

start()

Start recording; on failure the empty recording entry is removed.

Source code in src/murmurvault/session.py 
77
78
79
80
81
82
83
84
85
def start(self) -> None:
    """Start recording; on failure the empty recording entry is removed."""
    try:
        self.recorder.start()
    except Exception:
        if self.live:
            self.live.stop()
        self.vault.delete(self.rec)
        raise

stop()

Stop recording, save the tracks and the live draft, and index the draft.

Returns:

Type Description
Recording

The saved recording.

Raises:

Type Description
RuntimeError

If no audio was captured; the empty entry is removed.
Source code in src/murmurvault/session.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
def stop(self) -> Recording:
    """Stop recording, save the tracks and the live draft, and index the draft.

    Returns:
        The saved recording.

    Raises:
        RuntimeError: If no audio was captured; the empty entry is removed.
    """
    tracks, offsets, length = self.recorder.stop()
    draft = self.live.stop() if self.live else []
    if not tracks:
        self.vault.delete(self.rec)
        raise RuntimeError("no audio was captured (check devices with `murmurvault devices`)")
    rec = self.rec
    rec.tracks, rec.track_offsets, rec.duration = tracks, offsets, round(length, 2)
    self.vault.save(rec)
    if self.live and draft:
        for s in draft:
            off = offsets.get(s.track or "", 0.0)
            s.start, s.end = round(s.start + off, 3), round(s.end + off, 3)
        draft.sort(key=lambda s: s.start)
        eng = self.live.engine
        self.vault.save_transcript(rec, "live", Transcript(engine=eng.name, model=eng.model, segments=draft))
        reindex_recording(self.vault, rec, self.cfg)
    return rec

live_engine(cfg)

Build the engine for the live draft pass.

Parameters:

Name Type Description Default
cfg dict

The configuration.

 required

Returns:

Type Description
Engine

The engine named by live.engine, using live.model if it is the built-in engine.
Source code in src/murmurvault/session.py 
14
15
16
17
18
19
20
21
22
23
24
25
26
27
def live_engine(cfg: dict) -> Engine:
    """Build the engine for the live draft pass.

    Args:
        cfg: The configuration.

    Returns:
        The engine named by ``live.engine``, using ``live.model`` if it is the built-in engine.
    """
    name = cfg["live"].get("engine") or cfg["transcription"]["engine"]
    section = cfg["engines"].get(name, {})
    # live.model only makes sense for the built-in engine; API engines keep their own model name.
    model = cfg["live"].get("model") if section.get("type", "faster-whisper") == "faster-whisper" else None
    return get_engine(cfg, name, model)