path: root/src/mml_composemusic_mcp/server.py

# MCP server exposing compose_mml tool.

import argparse
import json
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path

from fastmcp import FastMCP
from fastmcp.tools import ToolResult
from fastmcp.utilities.types import Audio
from mcp.types import TextContent

from .ir import ErrorCode, ErrorDetail, ErrorPhase, NoteSequence
from .lexer import tokenize
from .parser_ppmck import parse_ppmck
from .parser_pyxel import parse_pyxel
from .semantic_ppmck import analyze_ppmck
from .semantic_pyxel import analyze_pyxel
from .synthesizer import build_channel_summary, synthesize, write_wav
from .templates import TEMPLATES, get_template

mcp = FastMCP("mml-composemusic-mcp")
OUTPUT_DIR = Path("./data")


@dataclass(frozen=True)
class ToolExecution:
    """Internal result kept separate from the public MCP payload."""

    result: dict
    is_error: bool = False
    wav_path: Path | None = None


def _is_error(errors: list[ErrorDetail]) -> bool:
    return any(e.severity == "error" for e in errors)


def _create_output_directory() -> Path:
    """Create and return a unique timestamped output directory.

    Format: YYYYMMDD_HHMMSS_mmm
    If the millisecond is already in use, advance the timestamp until a free
    directory can be created. Exclusive creation also handles concurrent calls.
    """
    now = datetime.now()
    for offset_ms in range(1000):
        candidate_time = now + timedelta(milliseconds=offset_ms)
        timestamp = candidate_time.strftime("%Y%m%d_%H%M%S_%f")[:-3]
        output_dir = OUTPUT_DIR / timestamp
        try:
            output_dir.mkdir(parents=True, exist_ok=False)
            return output_dir
        except FileExistsError:
            continue
    raise RuntimeError("一意な出力ディレクトリを作成できませんでした。")


def _split_errors(errors: list[ErrorDetail]) -> tuple[list[dict], list[dict]]:
    errs = [e.to_dict() for e in errors if e.severity == "error"]
    warns = [e.to_dict() for e in errors if e.severity == "warning"]
    return errs, warns


def _internal_error_execution() -> ToolExecution:
    """Build a path-free MCP tool error for unexpected failures."""
    return ToolExecution(
        {
            "success": False,
            "audio": None,
            "duration_sec": 0,
            "note_sequence": None,
            "validation": {
                "errors": [
                    {
                        "code": ErrorCode.RUNTIME_INTERNAL_ERROR.value,
                        "phase": ErrorPhase.RUNTIME.value,
                        "line": 0,
                        "column": 0,
                        "message": "内部エラーが発生しました。",
                        "severity": "error",
                        "hint": "MMLの内容を確認の上、再度お試しください。",
                    }
                ],
                "warnings": [],
            },
        },
        is_error=True,
    )


def _parse_mml(mml: str, mode: str) -> tuple[dict | None, list[ErrorDetail]]:
    """Run lexer, parser, and semantic analyzer for the given mode."""
    tokens = tokenize(mml, mode)

    if mode == "ppmck":
        program, errors = parse_ppmck(mml, tokens)
        if not _is_error(errors):
            note_sequence, semantic_errors = analyze_ppmck(mml, program)
            errors.extend(semantic_errors)
    elif mode == "pyxel":
        program, errors = parse_pyxel(mml, tokens)
        if not _is_error(errors):
            note_sequence, semantic_errors = analyze_pyxel(mml, program)
            errors.extend(semantic_errors)
    else:
        errors = [
            ErrorDetail(
                code=ErrorCode.VALIDATION_INVALID_MODE,
                phase=ErrorPhase.API,
                line=0,
                column=0,
                message=f"未知のモード '{mode}' です。",
                severity="error",
                hint="mode は 'ppmck' または 'pyxel' を指定してください。",
            )
        ]
        note_sequence = None

    if _is_error(errors):
        note_sequence = None
    return (note_sequence.to_dict() if note_sequence else None), errors


def _execute_compose_mml(
    action: str,
    mml: str = "",
    mode: str = "",
    template: str = "basic",
    sample_rate: int = 44100,
    normalize: bool = True,
) -> ToolExecution:
    """Execute a request without exposing internal file paths."""
    if action == "template":
        if mode not in ("ppmck", "pyxel"):
            mode = "ppmck"
        if template not in TEMPLATES[mode]:
            template = "basic"
        mml_text, description = get_template(mode, template)
        return ToolExecution({"mml": mml_text, "description": description})

    if action in ("compose", "validate"):
        if not mml or not mode:
            return ToolExecution(
                {
                    "success": False,
                    "valid": False,
                    "note_sequence": None,
                    "validation": {
                        "errors": [
                            {
                                "code": ErrorCode.VALIDATION_MISSING_PARAMETER.value,
                                "phase": ErrorPhase.API.value,
                                "line": 0,
                                "column": 0,
                                "message": "mml と mode は compose/validate の必須パラメータです。",
                                "severity": "error",
                                "hint": "mml に MML 文字列、mode に 'ppmck' または 'pyxel' を指定してください。",
                            }
                        ],
                        "warnings": [],
                    },
                },
                is_error=True,
            )
        try:
            note_sequence_dict, errors = _parse_mml(mml, mode)
            errors_list, warnings_list = _split_errors(errors)
            if action == "validate":
                return ToolExecution(
                    {
                        "valid": not _is_error(errors),
                        "errors": errors_list,
                        "warnings": warnings_list,
                        "note_sequence": note_sequence_dict,
                        "channel_summary": build_channel_summary(note_sequence_dict)
                        if note_sequence_dict
                        else [],
                    },
                    is_error=mode not in ("ppmck", "pyxel"),
                )
            # compose
            if _is_error(errors):
                return ToolExecution(
                    {
                        "success": False,
                        "audio": None,
                        "duration_sec": 0,
                        "note_sequence": note_sequence_dict,
                        "validation": {
                            "errors": errors_list,
                            "warnings": warnings_list,
                        },
                    },
                    is_error=True,
                )

            ns = _dict_to_note_sequence(note_sequence_dict)
            try:
                wave_data, duration, synth_errors = synthesize(
                    ns, mode, sample_rate, normalize
                )
            except Exception:
                return ToolExecution(
                    {
                        "success": False,
                        "audio": None,
                        "duration_sec": 0,
                        "note_sequence": note_sequence_dict,
                        "validation": {
                            "errors": [
                                {
                                    "code": ErrorCode.RUNTIME_SYNTHESIS_FAILED.value,
                                    "phase": ErrorPhase.RUNTIME.value,
                                    "line": 0,
                                    "column": 0,
                                    "message": "音声合成中にエラーが発生しました。",
                                    "severity": "error",
                                    "hint": "MMLの内容を確認の上、再度お試しください。問題が続く場合は、短いMMLから試してください。",
                                }
                            ],
                            "warnings": warnings_list,
                        },
                    },
                    is_error=True,
                )
            errors.extend(synth_errors)
            errors_list, warnings_list = _split_errors(errors)
            if _is_error(errors):
                return ToolExecution(
                    {
                        "success": False,
                        "audio": None,
                        "duration_sec": 0,
                        "note_sequence": note_sequence_dict,
                        "validation": {
                            "errors": errors_list,
                            "warnings": warnings_list,
                        },
                    },
                    is_error=True,
                )
            output_dir = _create_output_directory()
            wav_path = output_dir / "output.wav"
            wav_errors = write_wav(wav_path, wave_data, sample_rate)
            errors.extend(wav_errors)
            if not _is_error(errors):
                try:
                    (output_dir / "output.mml").write_text(mml, encoding="utf-8")
                except Exception:
                    errors.append(
                        ErrorDetail(
                            code=ErrorCode.RUNTIME_MML_WRITE_FAILED,
                            phase=ErrorPhase.RUNTIME,
                            line=0,
                            column=0,
                            message="MMLファイルの出力に失敗しました。",
                            severity="error",
                            hint="出力先を確認してください。",
                        )
                    )
            errors_list, warnings_list = _split_errors(errors)
            failed = _is_error(errors)
            result = {
                "success": not failed,
                "audio": None
                if failed
                else {
                    "mime_type": "audio/wav",
                    "size_bytes": wav_path.stat().st_size,
                    "sample_rate": sample_rate,
                },
                "duration_sec": duration,
                "note_sequence": note_sequence_dict,
                "validation": {"errors": errors_list, "warnings": warnings_list},
            }
            return ToolExecution(
                result,
                is_error=failed,
                wav_path=None if failed else wav_path,
            )
        except Exception:
            return _internal_error_execution()

    return ToolExecution(
        {
            "success": False,
            "validation": {
                "errors": [
                    {
                        "code": ErrorCode.VALIDATION_INVALID_ACTION.value,
                        "phase": ErrorPhase.API.value,
                        "line": 0,
                        "column": 0,
                        "message": f"未知の action '{action}' です。",
                        "severity": "error",
                        "hint": "action は 'compose', 'validate', 'template' のいずれかを指定してください。",
                    }
                ],
                "warnings": [],
            },
        },
        is_error=True,
    )


def _build_tool_result(execution: ToolExecution) -> ToolResult:
    """Convert an internal execution result to the public MCP response."""
    result = execution.result
    if "wav_path" in result:
        raise ValueError("Internal paths must not be included in MCP results")
    if "success" in result and result["success"] is execution.is_error:
        raise ValueError("success and is_error are inconsistent")
    if execution.is_error and execution.wav_path is not None:
        raise ValueError("Error results must not contain audio")

    content = [
        TextContent(
            type="text",
            text=json.dumps(result, ensure_ascii=False),
        )
    ]
    if execution.wav_path is not None:
        content.append(Audio(path=execution.wav_path).to_audio_content())

    return ToolResult(
        content=content,
        structured_content=result,
        is_error=execution.is_error,
    )


@mcp.tool(name="compose_mml", output_schema=None)
def compose_mml(
    action: str,
    mml: str = "",
    mode: str = "",
    template: str = "basic",
    sample_rate: int = 44100,
    normalize: bool = True,
) -> ToolResult:
    """Compose, validate, or generate templates for retro chiptune-style MML."""
    try:
        execution = _execute_compose_mml(
            action=action,
            mml=mml,
            mode=mode,
            template=template,
            sample_rate=sample_rate,
            normalize=normalize,
        )
        return _build_tool_result(execution)
    except Exception:
        return _build_tool_result(_internal_error_execution())


def _dict_to_note_sequence(data: dict) -> NoteSequence:
    from .ir import (
        ChannelSequence,
        DetuneEvent,
        DutyEnvelopeEvent,
        DutyEvent,
        EnvelopeEvent,
        GlideEvent,
        LfoEvent,
        NoteEnvEvent,
        NoteEvent,
        NoteSequence,
        PitchEnvEvent,
        QuantizeEvent,
        RelativeVolumeEvent,
        RepeatEvent,
        RestEvent,
        SweepEvent,
        TempoEvent,
        VibratoEvent,
        VolumeEnvelopeEvent,
        VolumeEvent,
    )

    ns = NoteSequence(
        version=data["version"],
        bpm=data["bpm"],
        ticks_per_quarter=data["ticks_per_quarter"],
        mode=data.get("mode", "ppmck"),
        definitions=data.get("definitions", {}),
        channels={},
    )
    for name, ch_data in data["channels"].items():
        ch = ChannelSequence(
            channel_type=ch_data["channel_type"],
            events=[],
            total_ticks=ch_data["total_ticks"],
        )
        for ev in ch_data["events"]:
            ev_type = ev["type"]
            if ev_type == "note":
                ch.events.append(NoteEvent(**ev))
            elif ev_type == "rest":
                ch.events.append(RestEvent(**ev))
            elif ev_type == "volume":
                ch.events.append(VolumeEvent(**ev))
            elif ev_type == "duty":
                ch.events.append(DutyEvent(**ev))
            elif ev_type == "tempo":
                ch.events.append(TempoEvent(**ev))
            elif ev_type == "repeat":
                ch.events.append(RepeatEvent(**ev))
            elif ev_type == "envelope":
                ch.events.append(EnvelopeEvent(**ev))
            elif ev_type == "quantize":
                ch.events.append(QuantizeEvent(**ev))
            elif ev_type == "detune":
                ch.events.append(DetuneEvent(**ev))
            elif ev_type == "sweep":
                ch.events.append(SweepEvent(**ev))
            elif ev_type == "rel_volume":
                ch.events.append(RelativeVolumeEvent(**ev))
            elif ev_type == "vol_envelope":
                ch.events.append(VolumeEnvelopeEvent(**ev))
            elif ev_type == "duty_envelope":
                ch.events.append(DutyEnvelopeEvent(**ev))
            elif ev_type == "lfo":
                ch.events.append(LfoEvent(**ev))
            elif ev_type == "pitch_envelope":
                ch.events.append(PitchEnvEvent(**ev))
            elif ev_type == "note_envelope":
                ch.events.append(NoteEnvEvent(**ev))
            elif ev_type == "vibrato":
                ch.events.append(VibratoEvent(**ev))
            elif ev_type == "glide":
                ch.events.append(GlideEvent(**ev))
        ns.channels[name] = ch
    return ns


def main() -> None:
    parser = argparse.ArgumentParser(description="MML Compose MCP Server")
    parser.add_argument(
        "--output-dir",
        type=str,
        default="./data",
        help="Directory to store generated WAV files (default: ./data)",
    )
    parser.add_argument(
        "--transport",
        type=str,
        default="stdio",
        choices=["stdio", "http", "sse", "streamable-http"],
        help="MCP transport (default: stdio)",
    )
    parser.add_argument(
        "--host",
        type=str,
        default="127.0.0.1",
        help="Host to bind for HTTP-based transports (default: 127.0.0.1)",
    )
    parser.add_argument(
        "--port",
        type=int,
        default=8080,
        help="Port to bind for HTTP-based transports (default: 8080)",
    )
    args = parser.parse_args()
    global OUTPUT_DIR
    OUTPUT_DIR = Path(args.output_dir)
    if args.transport == "stdio":
        mcp.run(transport="stdio")
    else:
        mcp.run(
            transport=args.transport,
            host=args.host,
            port=args.port,
        )


if __name__ == "__main__":
    main()