plugin.py
python
| 1 | """Music domain plugin — reference implementation of :class:`MuseDomainPlugin`. |
| 2 | |
| 3 | This plugin implements the five Muse domain interfaces for MIDI state: |
| 4 | notes, velocities, controller events (CC), pitch bends, and aftertouch. |
| 5 | |
| 6 | It is the domain that proved the abstraction. Every other domain — scientific |
| 7 | simulation, genomics, 3D spatial design — is a new plugin that implements |
| 8 | the same five interfaces. |
| 9 | |
| 10 | Live State |
| 11 | ---------- |
| 12 | For the music domain, ``LiveState`` is either: |
| 13 | |
| 14 | 1. A ``muse-work/`` directory path (``pathlib.Path``) — the CLI path where |
| 15 | MIDI files live on disk and are managed by ``muse commit / checkout``. |
| 16 | 2. A dict snapshot previously captured by :meth:`snapshot` — used when |
| 17 | constructing merges and diffs in memory. |
| 18 | |
| 19 | Both forms are supported. The plugin detects which form it received by |
| 20 | checking for ``pathlib.Path`` vs ``dict``. |
| 21 | |
| 22 | Snapshot Format |
| 23 | --------------- |
| 24 | A music snapshot is a JSON-serialisable dict: |
| 25 | |
| 26 | .. code-block:: json |
| 27 | |
| 28 | { |
| 29 | "files": { |
| 30 | "tracks/drums.mid": "<sha256>", |
| 31 | "tracks/bass.mid": "<sha256>" |
| 32 | }, |
| 33 | "domain": "music" |
| 34 | } |
| 35 | |
| 36 | The ``files`` key maps POSIX paths (relative to ``muse-work/``) to their |
| 37 | SHA-256 content digests. This is the same structure that the core file-based |
| 38 | store uses as a snapshot manifest — the music plugin does not add an |
| 39 | abstraction layer on top of the existing content-addressed object store. |
| 40 | |
| 41 | For more sophisticated use cases (DAW-level integration, per-note diffs, |
| 42 | emotion vectors, harmonic analysis), the snapshot can be extended with |
| 43 | additional top-level keys. The core DAG engine only requires that the |
| 44 | snapshot be JSON-serialisable and content-addressable. |
| 45 | """ |
| 46 | from __future__ import annotations |
| 47 | |
| 48 | import hashlib |
| 49 | import json |
| 50 | import pathlib |
| 51 | |
| 52 | from muse.domain import ( |
| 53 | DeltaManifest, |
| 54 | DriftReport, |
| 55 | LiveState, |
| 56 | MergeResult, |
| 57 | MuseDomainPlugin, |
| 58 | SnapshotManifest, |
| 59 | StateDelta, |
| 60 | StateSnapshot, |
| 61 | ) |
| 62 | |
| 63 | _DOMAIN_TAG = "music" |
| 64 | |
| 65 | |
| 66 | class MusicPlugin: |
| 67 | """Music domain plugin for the Muse VCS. |
| 68 | |
| 69 | Implements :class:`~muse.domain.MuseDomainPlugin` for MIDI state stored |
| 70 | as files in ``muse-work/``. Use this plugin when running ``muse`` against |
| 71 | a directory of MIDI, audio, or other music production files. |
| 72 | |
| 73 | This is the reference implementation. It demonstrates the five-interface |
| 74 | contract that every other domain plugin must satisfy. |
| 75 | """ |
| 76 | |
| 77 | # ------------------------------------------------------------------ |
| 78 | # 1. snapshot — capture live state as a content-addressed dict |
| 79 | # ------------------------------------------------------------------ |
| 80 | |
| 81 | def snapshot(self, live_state: LiveState) -> StateSnapshot: |
| 82 | """Capture the current ``muse-work/`` directory as a snapshot dict. |
| 83 | |
| 84 | Args: |
| 85 | live_state: Either a ``pathlib.Path`` pointing to ``muse-work/`` |
| 86 | or an existing snapshot dict (returned as-is). |
| 87 | |
| 88 | Returns: |
| 89 | A JSON-serialisable ``{"files": {path: sha256}, "domain": "music"}`` |
| 90 | dict. The ``files`` mapping is the canonical snapshot manifest used |
| 91 | by the core VCS engine for commit / checkout / diff. |
| 92 | """ |
| 93 | if isinstance(live_state, pathlib.Path): |
| 94 | workdir = live_state |
| 95 | files: dict[str, str] = {} |
| 96 | for file_path in sorted(workdir.rglob("*")): |
| 97 | if not file_path.is_file(): |
| 98 | continue |
| 99 | if file_path.name.startswith("."): |
| 100 | continue |
| 101 | rel = file_path.relative_to(workdir).as_posix() |
| 102 | files[rel] = _hash_file(file_path) |
| 103 | return SnapshotManifest(files=files, domain=_DOMAIN_TAG) |
| 104 | |
| 105 | return live_state |
| 106 | |
| 107 | # ------------------------------------------------------------------ |
| 108 | # 2. diff — compute the minimal delta between two snapshots |
| 109 | # ------------------------------------------------------------------ |
| 110 | |
| 111 | def diff(self, base: StateSnapshot, target: StateSnapshot) -> StateDelta: |
| 112 | """Compute the file-level delta between two music snapshots. |
| 113 | |
| 114 | Args: |
| 115 | base: The ancestor snapshot. |
| 116 | target: The later snapshot. |
| 117 | |
| 118 | Returns: |
| 119 | A delta dict with three keys: |
| 120 | - ``added``: list of paths present in *target* but not *base*. |
| 121 | - ``removed``: list of paths present in *base* but not *target*. |
| 122 | - ``modified``: list of paths present in both with different digests. |
| 123 | """ |
| 124 | base_files = base["files"] |
| 125 | target_files = target["files"] |
| 126 | |
| 127 | base_paths = set(base_files) |
| 128 | target_paths = set(target_files) |
| 129 | |
| 130 | added = sorted(target_paths - base_paths) |
| 131 | removed = sorted(base_paths - target_paths) |
| 132 | common = base_paths & target_paths |
| 133 | modified = sorted(p for p in common if base_files[p] != target_files[p]) |
| 134 | |
| 135 | return DeltaManifest( |
| 136 | domain=_DOMAIN_TAG, |
| 137 | added=added, |
| 138 | removed=removed, |
| 139 | modified=modified, |
| 140 | ) |
| 141 | |
| 142 | # ------------------------------------------------------------------ |
| 143 | # 3. merge — three-way reconciliation |
| 144 | # ------------------------------------------------------------------ |
| 145 | |
| 146 | def merge( |
| 147 | self, |
| 148 | base: StateSnapshot, |
| 149 | left: StateSnapshot, |
| 150 | right: StateSnapshot, |
| 151 | ) -> MergeResult: |
| 152 | """Three-way merge two divergent music state lines against a common base. |
| 153 | |
| 154 | A file is auto-merged when only one side changed it. A conflict is |
| 155 | recorded when both sides changed the same file relative to *base*. |
| 156 | |
| 157 | Args: |
| 158 | base: The common ancestor snapshot. |
| 159 | left: The current branch snapshot (ours). |
| 160 | right: The target branch snapshot (theirs). |
| 161 | |
| 162 | Returns: |
| 163 | A :class:`~muse.domain.MergeResult` with the merged snapshot and |
| 164 | any conflict descriptions. |
| 165 | """ |
| 166 | base_files = base["files"] |
| 167 | left_files = left["files"] |
| 168 | right_files = right["files"] |
| 169 | |
| 170 | left_changed: set[str] = _changed_paths(base_files, left_files) |
| 171 | right_changed: set[str] = _changed_paths(base_files, right_files) |
| 172 | conflict_paths: set[str] = left_changed & right_changed |
| 173 | |
| 174 | merged = dict(base_files) |
| 175 | |
| 176 | for path in left_changed - conflict_paths: |
| 177 | if path in left_files: |
| 178 | merged[path] = left_files[path] |
| 179 | else: |
| 180 | merged.pop(path, None) |
| 181 | |
| 182 | for path in right_changed - conflict_paths: |
| 183 | if path in right_files: |
| 184 | merged[path] = right_files[path] |
| 185 | else: |
| 186 | merged.pop(path, None) |
| 187 | |
| 188 | # If both sides deleted the same file, that is consensus — not a conflict. |
| 189 | real_conflicts = { |
| 190 | p for p in conflict_paths |
| 191 | if not (p not in left_files and p not in right_files) |
| 192 | } |
| 193 | |
| 194 | # Apply consensus deletions (both sides removed the same file). |
| 195 | for path in conflict_paths - real_conflicts: |
| 196 | merged.pop(path, None) |
| 197 | |
| 198 | return MergeResult( |
| 199 | merged=SnapshotManifest(files=merged, domain=_DOMAIN_TAG), |
| 200 | conflicts=sorted(real_conflicts), |
| 201 | ) |
| 202 | |
| 203 | # ------------------------------------------------------------------ |
| 204 | # 4. drift — compare committed state vs live state |
| 205 | # ------------------------------------------------------------------ |
| 206 | |
| 207 | def drift( |
| 208 | self, |
| 209 | committed: StateSnapshot, |
| 210 | live: LiveState, |
| 211 | ) -> DriftReport: |
| 212 | """Detect uncommitted changes in ``muse-work/`` relative to *committed*. |
| 213 | |
| 214 | Args: |
| 215 | committed: The last committed snapshot. |
| 216 | live: Either a ``pathlib.Path`` (``muse-work/``) or a snapshot |
| 217 | dict representing current live state. |
| 218 | |
| 219 | Returns: |
| 220 | A :class:`~muse.domain.DriftReport` describing whether and how the |
| 221 | live state differs from the committed snapshot. |
| 222 | """ |
| 223 | live_snapshot = self.snapshot(live) |
| 224 | delta = self.diff(committed, live_snapshot) |
| 225 | |
| 226 | added = delta["added"] |
| 227 | removed = delta["removed"] |
| 228 | modified = delta["modified"] |
| 229 | has_drift = bool(added or removed or modified) |
| 230 | |
| 231 | parts: list[str] = [] |
| 232 | if added: |
| 233 | parts.append(f"{len(added)} added") |
| 234 | if removed: |
| 235 | parts.append(f"{len(removed)} removed") |
| 236 | if modified: |
| 237 | parts.append(f"{len(modified)} modified") |
| 238 | |
| 239 | summary = ", ".join(parts) if parts else "working tree clean" |
| 240 | |
| 241 | return DriftReport(has_drift=has_drift, summary=summary, delta=delta) |
| 242 | |
| 243 | # ------------------------------------------------------------------ |
| 244 | # 5. apply — execute a delta against live state (checkout) |
| 245 | # ------------------------------------------------------------------ |
| 246 | |
| 247 | def apply(self, delta: StateDelta, live_state: LiveState) -> LiveState: |
| 248 | """Apply a delta to produce a new live state. |
| 249 | |
| 250 | For the music plugin in CLI mode, this returns the *target* snapshot |
| 251 | dict. The actual file restoration (writing MIDI files back to |
| 252 | ``muse-work/``) is handled by ``muse checkout`` using the core |
| 253 | object store, not by this method. |
| 254 | |
| 255 | This method is the semantic entry point for DAW-level integrations |
| 256 | that want to apply a delta to a live project without going through |
| 257 | the filesystem. |
| 258 | |
| 259 | Args: |
| 260 | delta: A delta produced by :meth:`diff`. |
| 261 | live_state: The current live state to patch. |
| 262 | |
| 263 | Returns: |
| 264 | The updated live state as a snapshot dict. |
| 265 | """ |
| 266 | current = self.snapshot(live_state) |
| 267 | current_files = dict(current["files"]) |
| 268 | |
| 269 | for path in delta["removed"]: |
| 270 | current_files.pop(path, None) |
| 271 | |
| 272 | for path in delta["added"] + delta["modified"]: |
| 273 | pass |
| 274 | |
| 275 | return SnapshotManifest(files=current_files, domain=_DOMAIN_TAG) |
| 276 | |
| 277 | |
| 278 | # --------------------------------------------------------------------------- |
| 279 | # Helpers |
| 280 | # --------------------------------------------------------------------------- |
| 281 | |
| 282 | |
| 283 | def _hash_file(path: pathlib.Path) -> str: |
| 284 | """Return the SHA-256 hex digest of a file's raw bytes.""" |
| 285 | h = hashlib.sha256() |
| 286 | with path.open("rb") as fh: |
| 287 | for chunk in iter(lambda: fh.read(65536), b""): |
| 288 | h.update(chunk) |
| 289 | return h.hexdigest() |
| 290 | |
| 291 | |
| 292 | def _changed_paths( |
| 293 | base: dict[str, str], other: dict[str, str] |
| 294 | ) -> set[str]: |
| 295 | """Return paths that differ between *base* and *other*.""" |
| 296 | base_p = set(base) |
| 297 | other_p = set(other) |
| 298 | added = other_p - base_p |
| 299 | deleted = base_p - other_p |
| 300 | common = base_p & other_p |
| 301 | modified = {p for p in common if base[p] != other[p]} |
| 302 | return added | deleted | modified |
| 303 | |
| 304 | |
| 305 | def content_hash(snapshot: StateSnapshot) -> str: |
| 306 | """Return a stable SHA-256 digest of a snapshot for content-addressing.""" |
| 307 | canonical = json.dumps(snapshot, sort_keys=True, separators=(",", ":")) |
| 308 | return hashlib.sha256(canonical.encode()).hexdigest() |
| 309 | |
| 310 | |
| 311 | #: Module-level singleton — import and use directly. |
| 312 | plugin = MusicPlugin() |
| 313 | |
| 314 | assert isinstance(plugin, MuseDomainPlugin), ( |
| 315 | "MusicPlugin does not satisfy the MuseDomainPlugin protocol" |
| 316 | ) |