import_cmd.py
python
| 1 | """muse import — import a MIDI or MusicXML file as a new Muse commit. |
| 2 | |
| 3 | Workflow |
| 4 | -------- |
| 5 | 1. Validate the file extension (supported: .mid, .midi, .xml, .musicxml). |
| 6 | 2. Parse the file into Muse's internal :class:`MuseImportData` representation. |
| 7 | 3. Apply any ``--track-map`` channel→name remapping. |
| 8 | 4. Copy the source file into ``muse-work/imports/<filename>``. |
| 9 | 5. Write a ``muse-work/imports/<filename>.meta.json`` with note count, tracks, |
| 10 | and tempo metadata (for downstream processing and commit diffs). |
| 11 | 6. Run :func:`_commit_async` to create the Muse commit. |
| 12 | 7. If ``--analyze``, print a multi-dimensional analysis of the imported content. |
| 13 | |
| 14 | Flags |
| 15 | ----- |
| 16 | ``<file>`` File to import (.mid/.midi/.xml/.musicxml). |
| 17 | ``--message TEXT`` Commit message (default: "Import <filename>"). |
| 18 | ``--track-map TEXT`` Channel→name mapping, e.g. ``ch0=bass,ch1=piano,ch9=drums``. |
| 19 | ``--section TEXT`` Section tag stored in the commit metadata JSON. |
| 20 | ``--analyze`` Run analysis and display results after importing. |
| 21 | ``--dry-run`` Validate only — do not write files or create a commit. |
| 22 | """ |
| 23 | from __future__ import annotations |
| 24 | |
| 25 | import asyncio |
| 26 | import json |
| 27 | import logging |
| 28 | import pathlib |
| 29 | import shutil |
| 30 | import typer |
| 31 | from sqlalchemy.ext.asyncio import AsyncSession |
| 32 | |
| 33 | from maestro.muse_cli._repo import require_repo |
| 34 | from maestro.muse_cli.commands.commit import _commit_async |
| 35 | from maestro.muse_cli.db import open_session |
| 36 | from maestro.muse_cli.errors import ExitCode |
| 37 | from maestro.muse_cli.midi_parser import ( |
| 38 | MuseImportData, |
| 39 | analyze_import, |
| 40 | apply_track_map, |
| 41 | parse_file, |
| 42 | parse_track_map_arg, |
| 43 | ) |
| 44 | |
| 45 | logger = logging.getLogger(__name__) |
| 46 | |
| 47 | app = typer.Typer() |
| 48 | |
| 49 | |
| 50 | # --------------------------------------------------------------------------- |
| 51 | # Testable async core |
| 52 | # --------------------------------------------------------------------------- |
| 53 | |
| 54 | |
| 55 | async def _import_async( |
| 56 | *, |
| 57 | file_path: pathlib.Path, |
| 58 | root: pathlib.Path, |
| 59 | session: AsyncSession | None, |
| 60 | message: str | None = None, |
| 61 | track_map: dict[str, str] | None = None, |
| 62 | section: str | None = None, |
| 63 | analyze: bool = False, |
| 64 | dry_run: bool = False, |
| 65 | ) -> str | None: |
| 66 | """Core import pipeline — fully injectable for tests. |
| 67 | |
| 68 | Parses, copies, commits, and optionally analyses the given file. |
| 69 | |
| 70 | Returns: |
| 71 | The new ``commit_id`` on success, or ``None`` when ``dry_run=True``. |
| 72 | |
| 73 | Raises: |
| 74 | ``typer.Exit`` with the appropriate exit code on validation failures so |
| 75 | the Typer callback surfaces a clean message instead of a traceback. |
| 76 | """ |
| 77 | # ── Validate and parse ─────────────────────────────────────────────── |
| 78 | try: |
| 79 | data: MuseImportData = parse_file(file_path) |
| 80 | except FileNotFoundError: |
| 81 | typer.echo(f"❌ File not found: {file_path}") |
| 82 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 83 | except ValueError as exc: |
| 84 | typer.echo(f"❌ {exc}") |
| 85 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 86 | except RuntimeError as exc: |
| 87 | typer.echo(f"❌ {exc}") |
| 88 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 89 | |
| 90 | # ── Apply track map ────────────────────────────────────────────────── |
| 91 | if track_map: |
| 92 | data = MuseImportData( |
| 93 | source_path=data.source_path, |
| 94 | format=data.format, |
| 95 | ticks_per_beat=data.ticks_per_beat, |
| 96 | tempo_bpm=data.tempo_bpm, |
| 97 | notes=apply_track_map(data.notes, track_map), |
| 98 | tracks=list({n.channel_name for n in apply_track_map(data.notes, track_map)}), |
| 99 | raw_meta=data.raw_meta, |
| 100 | ) |
| 101 | # Preserve insertion order for tracks |
| 102 | seen: set[str] = set() |
| 103 | ordered_tracks: list[str] = [] |
| 104 | for n in data.notes: |
| 105 | if n.channel_name not in seen: |
| 106 | seen.add(n.channel_name) |
| 107 | ordered_tracks.append(n.channel_name) |
| 108 | data = MuseImportData( |
| 109 | source_path=data.source_path, |
| 110 | format=data.format, |
| 111 | ticks_per_beat=data.ticks_per_beat, |
| 112 | tempo_bpm=data.tempo_bpm, |
| 113 | notes=data.notes, |
| 114 | tracks=ordered_tracks, |
| 115 | raw_meta=data.raw_meta, |
| 116 | ) |
| 117 | |
| 118 | effective_message = message or f"Import {file_path.name}" |
| 119 | |
| 120 | # ── Dry-run: validate only ──────────────────────────────────────────── |
| 121 | if dry_run: |
| 122 | typer.echo(f"✅ Dry run: '{file_path.name}' is valid ({data.format})") |
| 123 | typer.echo(f" Notes: {len(data.notes)}, Tracks: {len(data.tracks)}, Tempo: {data.tempo_bpm:.1f} BPM") |
| 124 | typer.echo(f" Would commit: {effective_message!r}") |
| 125 | if section: |
| 126 | typer.echo(f" Section: {section!r}") |
| 127 | if analyze: |
| 128 | typer.echo("\nAnalysis:") |
| 129 | typer.echo(analyze_import(data)) |
| 130 | return None |
| 131 | |
| 132 | # ── Copy file into muse-work/imports/ ──────────────────────────────── |
| 133 | imports_dir = root / "muse-work" / "imports" |
| 134 | imports_dir.mkdir(parents=True, exist_ok=True) |
| 135 | |
| 136 | dest = imports_dir / file_path.name |
| 137 | shutil.copy2(str(file_path), str(dest)) |
| 138 | logger.debug("✅ Copied %s → %s", file_path, dest) |
| 139 | |
| 140 | # ── Write metadata JSON ─────────────────────────────────────────────── |
| 141 | meta: dict[str, object] = { |
| 142 | "source": str(file_path), |
| 143 | "format": data.format, |
| 144 | "ticks_per_beat": data.ticks_per_beat, |
| 145 | "tempo_bpm": round(data.tempo_bpm, 3), |
| 146 | "note_count": len(data.notes), |
| 147 | "tracks": data.tracks, |
| 148 | "track_map": track_map or {}, |
| 149 | "section": section, |
| 150 | "raw_meta": data.raw_meta, |
| 151 | } |
| 152 | meta_path = imports_dir / f"{file_path.name}.meta.json" |
| 153 | meta_path.write_text(json.dumps(meta, indent=2)) |
| 154 | |
| 155 | # ── Commit ──────────────────────────────────────────────────────────── |
| 156 | # dry_run returns before this point, so session is always non-None here. |
| 157 | assert session is not None |
| 158 | commit_id = await _commit_async( |
| 159 | message=effective_message, |
| 160 | root=root, |
| 161 | session=session, |
| 162 | ) |
| 163 | |
| 164 | typer.echo(f"✅ Imported '{file_path.name}' as commit {commit_id[:8]}") |
| 165 | if section: |
| 166 | typer.echo(f" Section: {section!r}") |
| 167 | |
| 168 | # ── Analysis ───────────────────────────────────────────────────────── |
| 169 | if analyze: |
| 170 | typer.echo("\nAnalysis:") |
| 171 | typer.echo(analyze_import(data)) |
| 172 | |
| 173 | return commit_id |
| 174 | |
| 175 | |
| 176 | # --------------------------------------------------------------------------- |
| 177 | # Typer command |
| 178 | # --------------------------------------------------------------------------- |
| 179 | |
| 180 | |
| 181 | @app.callback(invoke_without_command=True) |
| 182 | def import_file( |
| 183 | ctx: typer.Context, |
| 184 | file: str = typer.Argument(..., help="Path to the MIDI or MusicXML file to import."), |
| 185 | message: str | None = typer.Option( |
| 186 | None, |
| 187 | "--message", |
| 188 | "-m", |
| 189 | help='Commit message (default: "Import <filename>").', |
| 190 | ), |
| 191 | track_map: str | None = typer.Option( |
| 192 | None, |
| 193 | "--track-map", |
| 194 | help='Map MIDI channels to track names, e.g. "ch0=bass,ch1=piano,ch9=drums".', |
| 195 | ), |
| 196 | section: str | None = typer.Option( |
| 197 | None, |
| 198 | "--section", |
| 199 | help="Tag the imported content as a specific section.", |
| 200 | ), |
| 201 | analyze: bool = typer.Option( |
| 202 | False, |
| 203 | "--analyze", |
| 204 | help="Run multi-dimensional analysis on the imported content and display it.", |
| 205 | ), |
| 206 | dry_run: bool = typer.Option( |
| 207 | False, |
| 208 | "--dry-run", |
| 209 | help="Validate the import without committing.", |
| 210 | ), |
| 211 | ) -> None: |
| 212 | """Import a MIDI or MusicXML file as a new Muse commit.""" |
| 213 | file_path = pathlib.Path(file).expanduser().resolve() |
| 214 | |
| 215 | parsed_track_map: dict[str, str] | None = None |
| 216 | if track_map is not None: |
| 217 | try: |
| 218 | parsed_track_map = parse_track_map_arg(track_map) |
| 219 | except ValueError as exc: |
| 220 | typer.echo(f"❌ --track-map: {exc}") |
| 221 | raise typer.Exit(code=ExitCode.USER_ERROR) |
| 222 | |
| 223 | root = require_repo() |
| 224 | |
| 225 | async def _run() -> None: |
| 226 | if dry_run: |
| 227 | # Dry-run does not need a DB session |
| 228 | await _import_async( |
| 229 | file_path=file_path, |
| 230 | root=root, |
| 231 | session=None, |
| 232 | message=message, |
| 233 | track_map=parsed_track_map, |
| 234 | section=section, |
| 235 | analyze=analyze, |
| 236 | dry_run=True, |
| 237 | ) |
| 238 | else: |
| 239 | async with open_session() as session: |
| 240 | await _import_async( |
| 241 | file_path=file_path, |
| 242 | root=root, |
| 243 | session=session, |
| 244 | message=message, |
| 245 | track_map=parsed_track_map, |
| 246 | section=section, |
| 247 | analyze=analyze, |
| 248 | dry_run=False, |
| 249 | ) |
| 250 | |
| 251 | try: |
| 252 | asyncio.run(_run()) |
| 253 | except typer.Exit: |
| 254 | raise |
| 255 | except Exception as exc: |
| 256 | typer.echo(f"❌ muse import failed: {exc}") |
| 257 | logger.error("❌ muse import error: %s", exc, exc_info=True) |
| 258 | raise typer.Exit(code=ExitCode.INTERNAL_ERROR) |