diff options
| author | tanaka takashi <isazaka@gmail.com> | 2026-07-11 19:02:14 +0900 |
|---|---|---|
| committer | tanaka takashi <isazaka@gmail.com> | 2026-07-11 19:02:14 +0900 |
| commit | ca35520338057512a71acd42bb116ffbb82cc192 (patch) | |
| tree | 9faa2428aeb71b62497f21ceec93b99ef7f2a5ac /src | |
| parent | 7dca1616b8c891f5e140c7267323d1fcdde6741a (diff) | |
feat: implement Famicom-style MML composer MCP server
- Add FastMCP server with compose_mml tool
- Support ppmck and pyxel MML syntax
- Add lexer and parsers for both formats
- Add NES APU-style synthesizer (pulse/triangle/noise) to WAV
- Add MML templates (basic/melody/chord/drum/empty) for both modes
- Add tests for lexer, parser, and server
- Support stdio, http, sse, streamable-http transports
- Update README, pyproject.toml, and .gitignore
Diffstat (limited to 'src')
| -rw-r--r-- | src/mml_composemusic_mcp/__init__.py (renamed from src/__init__.py) | 0 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/ir.py | 185 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/lexer.py | 314 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/parser_base.py | 92 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/parser_ppmck.py | 451 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/parser_pyxel.py | 652 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/server.py | 254 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/synthesizer.py | 301 | ||||
| -rw-r--r-- | src/mml_composemusic_mcp/templates.py | 32 |
9 files changed, 2281 insertions, 0 deletions
diff --git a/src/__init__.py b/src/mml_composemusic_mcp/__init__.py index e69de29..e69de29 100644 --- a/src/__init__.py +++ b/src/mml_composemusic_mcp/__init__.py diff --git a/src/mml_composemusic_mcp/ir.py b/src/mml_composemusic_mcp/ir.py new file mode 100644 index 0000000..05ef37a --- /dev/null +++ b/src/mml_composemusic_mcp/ir.py @@ -0,0 +1,185 @@ +"""Intermediate representation for MML NoteSequence and errors.""" + +from dataclasses import dataclass, field +from enum import Enum +from typing import Any + + +class ErrorCode(Enum): + """Machine-readable error codes.""" + + SYNTAX_INVALID_TOKEN = "SYNTAX_INVALID_TOKEN" + SYNTAX_INVALID_NUMBER = "SYNTAX_INVALID_NUMBER" + SYNTAX_VALUE_OUT_OF_RANGE = "SYNTAX_VALUE_OUT_OF_RANGE" + SYNTAX_UNEXPECTED_TOKEN = "SYNTAX_UNEXPECTED_TOKEN" + SYNTAX_UNTERMINATED_REPEAT = "SYNTAX_UNTERMINATED_REPEAT" + SYNTAX_UNMATCHED_REPEAT_END = "SYNTAX_UNMATCHED_REPEAT_END" + SYNTAX_UNTERMINATED_TIE = "SYNTAX_UNTERMINATED_TIE" + SYNTAX_INVALID_TRACK_HEADER = "SYNTAX_INVALID_TRACK_HEADER" + SYNTAX_DUPLICATE_TRACK = "SYNTAX_DUPLICATE_TRACK" + SYNTAX_EMPTY_TRACK = "SYNTAX_EMPTY_TRACK" + SYNTAX_NOTE_OUT_OF_RANGE = "SYNTAX_NOTE_OUT_OF_RANGE" + SYNTAX_CHANNEL_MISMATCH = "SYNTAX_CHANNEL_MISMATCH" + SYNTAX_UNTERMINATED_HEADER = "SYNTAX_UNTERMINATED_HEADER" + SYNTAX_UNDEFINED_REFERENCE = "SYNTAX_UNDEFINED_REFERENCE" + + SYSTEM_SYNTHESIS_FAILED = "SYSTEM_SYNTHESIS_FAILED" + SYSTEM_WAV_WRITE_FAILED = "SYSTEM_WAV_WRITE_FAILED" + SYSTEM_INTERNAL_ERROR = "SYSTEM_INTERNAL_ERROR" + + +@dataclass +class ErrorDetail: + """Structured error/warning detail.""" + + code: ErrorCode + line: int + column: int + message: str + severity: str # "error" | "warning" + hint: str = "" + context: str = "" + + def to_dict(self) -> dict[str, Any]: + return { + "code": self.code.value, + "line": self.line, + "column": self.column, + "message": self.message, + "severity": self.severity, + "hint": self.hint, + "context": self.context, + } + + +@dataclass +class NoteEvent: + type: str = "note" + tick_position: int = 0 + duration: int = 0 + note_number: int = 69 + velocity: int = 15 + duty: int = 2 + gate_time: float = 1.0 + + +@dataclass +class RestEvent: + type: str = "rest" + tick_position: int = 0 + duration: int = 0 + + +@dataclass +class VolumeEvent: + type: str = "volume" + tick_position: int = 0 + value: int = 15 + + +@dataclass +class DutyEvent: + type: str = "duty" + tick_position: int = 0 + value: int = 2 + + +@dataclass +class TempoEvent: + type: str = "tempo" + tick_position: int = 0 + bpm: int = 120 + + +@dataclass +class RepeatEvent: + type: str = "repeat" + tick_position: int = 0 + start_tick: int = 0 + end_tick: int = 0 + repeat_count: int = 1 + duration: int = 0 + + +@dataclass +class EnvelopeEvent: + type: str = "envelope" + tick_position: int = 0 + slot: int = 0 + points: list[dict[str, int]] = field(default_factory=list) + + +@dataclass +class VibratoEvent: + type: str = "vibrato" + tick_position: int = 0 + slot: int = 0 + params: dict[str, int] = field(default_factory=dict) + + +@dataclass +class GlideEvent: + type: str = "glide" + tick_position: int = 0 + slot: int = 0 + params: dict[str, int] = field(default_factory=dict) + + +Event = ( + NoteEvent + | RestEvent + | VolumeEvent + | DutyEvent + | TempoEvent + | RepeatEvent + | EnvelopeEvent + | VibratoEvent + | GlideEvent +) + + +@dataclass +class ChannelSequence: + channel_type: str = "pulse" + events: list[Event] = field(default_factory=list) + total_ticks: int = 0 + + def to_dict(self) -> dict[str, Any]: + return { + "channel_type": self.channel_type, + "events": [event_to_dict(e) for e in self.events], + "total_ticks": self.total_ticks, + } + + +@dataclass +class NoteSequence: + version: str = "1.0" + bpm: int = 120 + ticks_per_quarter: int = 192 + channels: dict[str, ChannelSequence] = field( + default_factory=lambda: { + "Pulse1": ChannelSequence("pulse"), + "Pulse2": ChannelSequence("pulse"), + "Triangle": ChannelSequence("triangle"), + "Noise": ChannelSequence("noise"), + } + ) + + def to_dict(self) -> dict[str, Any]: + return { + "version": self.version, + "bpm": self.bpm, + "ticks_per_quarter": self.ticks_per_quarter, + "channels": {k: v.to_dict() for k, v in self.channels.items()}, + } + + +def event_to_dict(event: Event) -> dict[str, Any]: + if isinstance( + event, (NoteEvent, RestEvent, VolumeEvent, DutyEvent, TempoEvent, RepeatEvent) + ): + return event.__dict__ + if isinstance(event, (EnvelopeEvent, VibratoEvent, GlideEvent)): + return event.__dict__ + raise TypeError(f"Unsupported event type: {type(event)}") diff --git a/src/mml_composemusic_mcp/lexer.py b/src/mml_composemusic_mcp/lexer.py new file mode 100644 index 0000000..05861cf --- /dev/null +++ b/src/mml_composemusic_mcp/lexer.py @@ -0,0 +1,314 @@ +"""Common lexer for MML.""" + +from collections.abc import Iterable +from dataclasses import dataclass +from enum import Enum, auto + + +class TokenType(Enum): + NOTE = auto() + REST = auto() + OCTAVE = auto() + OCTAVE_UP = auto() + OCTAVE_DOWN = auto() + LENGTH = auto() + VOLUME = auto() + DUTY = auto() + TEMPO = auto() + GATE_TIME = auto() + TIE = auto() + REPEAT_START = auto() + REPEAT_END = auto() + BAR = auto() + TRACK_HEADER = auto() + HEADER = auto() + COMMENT = auto() + NUMBER = auto() + SHARP = auto() + FLAT = auto() + DOT = auto() + TEXT = auto() + AT = auto() + TRANSPOSE = auto() + DETUNE = auto() + EXT_CMD = auto() # @ENV, @VIB, @GLI + EOF = auto() + + +@dataclass +class Token: + type: TokenType + value: str + line: int + column: int + raw: str = "" + + +def _track_header_char(ch: str) -> bool: + return ch.isupper() and ch in "ABTNL" + + +class Lexer: + def __init__(self, source: str, mode: str = "ppmck") -> None: + self.source = source + self.mode = mode.lower() + self.tokens: list[Token] = [] + self.line = 1 + self.column = 1 + self.pos = 0 + + def _peek(self, offset: int = 0) -> str: + idx = self.pos + offset + if idx >= len(self.source): + return "" + return self.source[idx] + + def _advance(self) -> str: + ch = self.source[self.pos] + self.pos += 1 + if ch == "\n": + self.line += 1 + self.column = 1 + else: + self.column += 1 + return ch + + def _skip_whitespace(self) -> None: + while self.pos < len(self.source) and self.source[self.pos] in " \t\r": + self._advance() + + def _read_number(self) -> str: + start = "" + while self.pos < len(self.source) and self.source[self.pos].isdigit(): + start += self._advance() + return start + + def _read_until_newline(self) -> str: + start = "" + while self.pos < len(self.source) and self.source[self.pos] not in "\r\n": + start += self._advance() + return start + + def _read_header_value(self) -> str: + value = "" + if self._peek() == '"': + self._advance() + while self.pos < len(self.source) and self._peek() not in '"\r\n': + value += self._advance() + if self._peek() == '"': + self._advance() + else: + # Return raw unterminated header; parser reports error + pass + return value + + def tokenize(self) -> list[Token]: + while self.pos < len(self.source): + self._skip_whitespace() + if self.pos >= len(self.source): + break + ch = self._peek() + start_line = self.line + start_col = self.column + + if self.mode == "ppmck": + self._tokenize_ppmck(ch, start_line, start_col) + else: + self._tokenize_pyxel(ch, start_line, start_col) + self.tokens.append(Token(TokenType.EOF, "", self.line, self.column)) + return self.tokens + + def _emit(self, ttype: TokenType, value: str, raw: str = "") -> None: + self.tokens.append( + Token(ttype, value, self.line, self.column - len(raw), raw or value) + ) + + def _tokenize_ppmck(self, ch: str, start_line: int, start_col: int) -> None: # noqa: ARG002 + if ch == "\n": + self._advance() + return + if ch == ";": + raw = self._read_until_newline() + self._emit(TokenType.COMMENT, raw, raw) + return + if ch == "#": + raw = "#" + self._read_until_newline() + self._emit(TokenType.HEADER, raw, raw) + return + if ch == "|": + self._advance() + self._emit(TokenType.BAR, "|", "|") + return + if ch == ">": + self._advance() + self._emit(TokenType.OCTAVE_UP, ">", ">") + return + if ch == "<": + self._advance() + self._emit(TokenType.OCTAVE_DOWN, "<", "<") + return + if ch == "&": + self._advance() + self._emit(TokenType.TIE, "&", "&") + return + if ch == "[": + self._advance() + self._emit(TokenType.REPEAT_START, "[", "[") + return + if ch == "]": + self._advance() + count = self._read_number() + self._emit(TokenType.REPEAT_END, count, "]" + count) + return + if ch == ".": + self._advance() + self._emit(TokenType.DOT, ".", ".") + return + if ch == "+": + self._advance() + self._emit(TokenType.SHARP, "+", "+") + return + if ch == "-": + self._advance() + self._emit(TokenType.FLAT, "-", "-") + return + if ch.isdigit(): + num = self._read_number() + self._emit(TokenType.NUMBER, num, num) + return + if _track_header_char(ch): + raw = self._advance() + self._emit(TokenType.TRACK_HEADER, raw.upper(), raw) + return + if ch.isalpha(): + cmd = self._advance() + if ch in "olvtq": + num = self._read_number() + raw = cmd + num + if ch == "o": + self._emit(TokenType.OCTAVE, num, raw) + elif ch == "l": + self._emit(TokenType.LENGTH, num, raw) + elif ch == "v": + self._emit(TokenType.VOLUME, num, raw) + elif ch == "t": + self._emit(TokenType.TEMPO, num, raw) + elif ch == "q": + self._emit(TokenType.DUTY, num, raw) + return + # Note commands + note = ch.lower() + if note in "cdefgabr": + if note == "r": + self._emit(TokenType.REST, note, note) + else: + self._emit(TokenType.NOTE, note, note) + return + # Unknown command + self._emit(TokenType.TEXT, cmd, cmd) + return + # Fallback invalid char + raw = self._advance() + self._emit(TokenType.TEXT, raw, raw) + + def _tokenize_pyxel(self, ch: str, start_line: int, start_col: int) -> None: # noqa: ARG002 + if ch == "\n": + self._advance() + return + if ch == "|": + self._advance() + self._emit(TokenType.BAR, "|", "|") + return + if ch == ">": + self._advance() + self._emit(TokenType.OCTAVE_UP, ">", ">") + return + if ch == "<": + self._advance() + self._emit(TokenType.OCTAVE_DOWN, "<", "<") + return + if ch == "&": + self._advance() + self._emit(TokenType.TIE, "&", "&") + return + if ch == "[": + self._advance() + self._emit(TokenType.REPEAT_START, "[", "[") + return + if ch == "]": + self._advance() + count = self._read_number() + self._emit(TokenType.REPEAT_END, count, "]" + count) + return + if ch == ".": + self._advance() + self._emit(TokenType.DOT, ".", ".") + return + if ch == "+": + self._advance() + self._emit(TokenType.SHARP, "+", "+") + return + if ch == "-": + self._advance() + self._emit(TokenType.FLAT, "-", "-") + return + if ch.isdigit(): + num = self._read_number() + if self._peek() == ":": + self._advance() + self._emit(TokenType.TRACK_HEADER, num, num + ":") + else: + self._emit(TokenType.NUMBER, num, num) + return + if ch == "@": + self._advance() + if self.source[self.pos : self.pos + 3].upper() in ("ENV", "VIB", "GLI"): + ext = self.source[self.pos : self.pos + 3].upper() + self.pos += 3 + self.column += 3 + raw = "@" + ext + self._emit(TokenType.EXT_CMD, ext, raw) + return + num = self._read_number() + self._emit(TokenType.AT, num, "@" + num) + return + if ch.isalpha(): + cmd = self._advance() + if cmd in "OLVTQKY": + num = self._read_number() + raw = cmd + num + if cmd == "O": + self._emit(TokenType.OCTAVE, num, raw) + elif cmd == "L": + self._emit(TokenType.LENGTH, num, raw) + elif cmd == "V": + self._emit(TokenType.VOLUME, num, raw) + elif cmd == "T": + self._emit(TokenType.TEMPO, num, raw) + elif cmd == "Q": + self._emit(TokenType.GATE_TIME, num, raw) + elif cmd == "K": + self._emit(TokenType.TRANSPOSE, num, raw) + elif cmd == "Y": + self._emit(TokenType.DETUNE, num, raw) + return + note = cmd.upper() + if note in "CDEFGABR": + if note == "R": + self._emit(TokenType.REST, note, note) + else: + self._emit(TokenType.NOTE, note, note) + return + self._emit(TokenType.TEXT, cmd, cmd) + return + # Fallback + raw = self._advance() + self._emit(TokenType.TEXT, raw, raw) + + +def tokenize(source: str, mode: str = "ppmck") -> list[Token]: + return Lexer(source, mode).tokenize() + + +def iter_token_text(tokens: Iterable[Token]) -> str: + return " ".join(f"{t.type.name}:{t.value}" for t in tokens) diff --git a/src/mml_composemusic_mcp/parser_base.py b/src/mml_composemusic_mcp/parser_base.py new file mode 100644 index 0000000..0427a4c --- /dev/null +++ b/src/mml_composemusic_mcp/parser_base.py @@ -0,0 +1,92 @@ +"""Base parser utilities shared between ppmck and pyxel parsers.""" + +from dataclasses import dataclass + +from .ir import ErrorCode, ErrorDetail +from .lexer import Token, TokenType + + +@dataclass +class ParserContext: + tokens: list[Token] + pos: int = 0 + errors: list[ErrorDetail] = None # type: ignore[assignment] + + def __post_init__(self) -> None: + if self.errors is None: + self.errors = [] + + def peek(self, offset: int = 0) -> Token: + idx = self.pos + offset + if idx >= len(self.tokens): + return self.tokens[-1] + return self.tokens[idx] + + def advance(self) -> Token: + token = self.peek() + self.pos += 1 + return token + + def match(self, *types: TokenType) -> Token | None: + token = self.peek() + if token.type in types: + return self.advance() + return None + + def expect(self, *types: TokenType) -> Token | None: + return self.match(*types) + + def add_error( + self, + code: ErrorCode, + line: int, + column: int, + message: str, + severity: str, + hint: str = "", + context: str = "", + ) -> None: + self.errors.append( + ErrorDetail( + code=code, + line=line, + column=column, + message=message, + severity=severity, + hint=hint, + context=context, + ) + ) + + +class ParserError(Exception): + pass + + +def length_value_to_ticks(length: int, dots: int, ticks_per_quarter: int = 192) -> int: + base = ticks_per_quarter * 4 // length + total = base + add = base + for _ in range(dots): + add //= 2 + total += add + return total + + +NOTE_OFFSET = { + "c": 0, + "d": 2, + "e": 4, + "f": 5, + "g": 7, + "a": 9, + "b": 11, +} + + +def note_to_midi(note: str, octave: int, accidental: int = 0) -> int: + return (octave + 1) * 12 + NOTE_OFFSET[note] + accidental + + +def clamp(value: int, min_val: int, max_val: int) -> int: + return max(min_val, min(value, max_val)) diff --git a/src/mml_composemusic_mcp/parser_ppmck.py b/src/mml_composemusic_mcp/parser_ppmck.py new file mode 100644 index 0000000..c2a1fa3 --- /dev/null +++ b/src/mml_composemusic_mcp/parser_ppmck.py @@ -0,0 +1,451 @@ +"""PPMCK-style MML parser.""" + +from .ir import ( + DutyEvent, + ErrorCode, + ErrorDetail, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, +) +from .lexer import Token, TokenType +from .parser_base import ( + ParserContext, + clamp, + length_value_to_ticks, + note_to_midi, +) + +CHANNEL_MAP = { + "A": "Pulse1", + "B": "Pulse2", + "T": "Triangle", + "N": "Noise", +} +PULSE_CHANNELS = {"Pulse1", "Pulse2"} + + +class PpmckParser: + def __init__(self, source: str, tokens: list[Token]) -> None: + self.source = source + self.ctx = ParserContext(tokens) + self.note_sequence = NoteSequence() + self.current_channel: str | None = None + self.octave = 4 + self.default_length = 4 + self.velocity = 15 + self.duty = 2 + self.tick_position = 0 + self.headers: list[tuple[str, str]] = [] + self.loop_start: int | None = None + self.tracks_seen: set[str] = set() + + def _context_line(self, token: Token) -> str: + lines = self.source.splitlines() + if 1 <= token.line <= len(lines): + return lines[token.line - 1] + return "" + + def parse(self) -> tuple[NoteSequence, list[ErrorDetail]]: + while self.ctx.peek().type != TokenType.EOF: + token = self.ctx.peek() + if token.type == TokenType.HEADER: + self._parse_header() + continue + if token.type == TokenType.COMMENT: + self.ctx.advance() + continue + if token.type == TokenType.TRACK_HEADER: + self._parse_track_header() + continue + if self.current_channel is None: + self.ctx.advance() + continue + self._parse_statement() + + self._validate_tracks() + + if self.loop_start is not None: + max_tick = max( + ch.total_ticks for ch in self.note_sequence.channels.values() + ) + for ch in self.note_sequence.channels.values(): + ch.events.append( + RepeatEvent( + start_tick=self.loop_start, + end_tick=max_tick, + repeat_count=0, + ) + ) + + for ch in self.note_sequence.channels.values(): + ch.total_ticks = max( + (e.tick_position + getattr(e, "duration", 0) for e in ch.events), + default=0, + ) + + return self.note_sequence, self.ctx.errors + + def _parse_header(self) -> None: + token = self.ctx.advance() + raw = token.value + parts = raw.split(None, 1) + key = parts[0] if parts else raw + value = "" + if len(parts) > 1 and parts[1].startswith('"'): + value = parts[1].strip('"') + if not parts[1].endswith('"') or parts[1] == '"': + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_HEADER, + line=token.line, + column=token.column, + message=f"ヘッダー '{key}' の引用符が閉じられていません。", + severity="error", + hint='二重引用符 " で値を閉じてください。例: #TITLE "My Song"', + context=self._context_line(token), + ) + self.headers.append((key, value)) + + def _parse_track_header(self) -> None: + token = self.ctx.advance() + ch = token.value.upper() + if ch == "L": + self.loop_start = 0 + return + if ch not in CHANNEL_MAP: + self.ctx.add_error( + code=ErrorCode.SYNTAX_INVALID_TRACK_HEADER, + line=token.line, + column=token.column, + message=f"無効なトラックヘッダー '{ch}' です。", + severity="error", + hint="A, B, T, N, L のいずれかを使用してください。", + context=self._context_line(token), + ) + self.current_channel = None + return + self.current_channel = CHANNEL_MAP[ch] + if self.current_channel in self.tracks_seen: + self.ctx.add_error( + code=ErrorCode.SYNTAX_DUPLICATE_TRACK, + line=token.line, + column=token.column, + message=f"トラック '{ch}' が複数回定義されています。", + severity="error", + hint="各トラックは1回のみ定義できます。重複定義を削除してください。", + context=self._context_line(token), + ) + self.tracks_seen.add(self.current_channel) + self.octave = 4 + self.default_length = 4 + self.velocity = 15 + self.duty = 2 + self.tick_position = 0 + + def _parse_statement(self) -> None: + token = self.ctx.peek() + handlers = { + TokenType.NOTE: self._parse_note, + TokenType.REST: self._parse_rest, + TokenType.OCTAVE: self._parse_octave, + TokenType.OCTAVE_UP: self._parse_octave_up, + TokenType.OCTAVE_DOWN: self._parse_octave_down, + TokenType.LENGTH: self._parse_length, + TokenType.VOLUME: self._parse_volume, + TokenType.DUTY: self._parse_duty, + TokenType.TEMPO: self._parse_tempo, + TokenType.TIE: self._parse_tie, + TokenType.REPEAT_START: self._parse_repeat_start, + TokenType.REPEAT_END: self._parse_repeat_end, + TokenType.BAR: self._parse_bar, + TokenType.COMMENT: lambda: self.ctx.advance(), + } + handler = handlers.get(token.type) + if handler: + handler() + else: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"'{token.raw}' はここでは使用できません。", + severity="error", + hint="音符、休符、コマンド、小節線を配置してください。", + context=self._context_line(token), + ) + self.ctx.advance() + + def _read_length(self) -> tuple[int, int]: + dots = 0 + length_token = self.ctx.match(TokenType.NUMBER) + if length_token is None: + length = self.default_length + else: + length = int(length_token.value) + while self.ctx.match(TokenType.DOT): + dots += 1 + return length, dots + + def _parse_note(self) -> None: + token = self.ctx.advance() + note = token.value.lower() + length, dots = self._read_length() + ticks = length_value_to_ticks(length, dots) + accidental = 0 + while True: + if self.ctx.match(TokenType.SHARP): + accidental += 1 + elif self.ctx.match(TokenType.FLAT): + accidental -= 1 + else: + break + + if self.current_channel == "Noise": + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message="チャンネル 'Noise' で音高指定は使用できません。", + severity="warning", + hint="Noiseチャンネルでは音高ではなく音長と音量のみ指定してください。", + context=self._context_line(token), + ) + midi = 0 + else: + midi = note_to_midi(note, self.octave, accidental) + if not 0 <= midi <= 127: + self.ctx.add_error( + code=ErrorCode.SYNTAX_NOTE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"音高が範囲外です(MIDI音番号 {midi})。", + severity="warning", + hint="オクターブを調整してください。有効範囲: o0〜o7。", + context=self._context_line(token), + ) + midi = clamp(midi, 0, 127) + + event = NoteEvent( + tick_position=self.tick_position, + duration=ticks, + note_number=midi, + velocity=self.velocity, + duty=self.duty, + gate_time=1.0, + ) + self._add_event(event) + self.tick_position += ticks + + def _parse_rest(self) -> None: + self.ctx.advance() + length, dots = self._read_length() + ticks = length_value_to_ticks(length, dots) + event = RestEvent(tick_position=self.tick_position, duration=ticks) + self._add_event(event) + self.tick_position += ticks + + def _parse_octave(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not 0 <= value <= 7: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'o' の値 {value} は範囲外です。有効範囲: 0〜7。", + severity="error", + hint="例: o4 のように指定してください。", + context=self._context_line(token), + ) + return + self.octave = value + + def _parse_octave_up(self) -> None: + self.ctx.advance() + if self.octave < 7: + self.octave += 1 + + def _parse_octave_down(self) -> None: + self.ctx.advance() + if self.octave > 0: + self.octave -= 1 + + def _parse_length(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not 1 <= value <= 192: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'l' の値 {value} は範囲外です。有効範囲: 1〜192。", + severity="error", + hint="例: l8 のように指定してください。", + context=self._context_line(token), + ) + return + self.default_length = value + + def _parse_volume(self) -> None: + token = self.ctx.advance() + if self.current_channel == "Triangle": + # Silently ignored per spec + return + value = int(token.value) + if not 0 <= value <= 15: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'v' の値 {value} は範囲外です。有効範囲: 0〜15。", + severity="error", + hint="例: v15 のように指定してください。", + context=self._context_line(token), + ) + return + self.velocity = value + + def _parse_duty(self) -> None: + token = self.ctx.advance() + if self.current_channel not in PULSE_CHANNELS: + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message=f"チャンネル '{self.current_channel}' で 'q' は使用できません。", + severity="warning", + hint="Triangleチャンネルはデューティ比を持ちません。q コマンドを削除してください。", + context=self._context_line(token), + ) + return + value = int(token.value) + if not 0 <= value <= 3: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'q' の値 {value} は範囲外です。有効範囲: 0〜3。", + severity="error", + hint="例: q2 のように指定してください。", + context=self._context_line(token), + ) + return + self.duty = value + self._add_event(DutyEvent(tick_position=self.tick_position, value=value)) + + def _parse_tempo(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if value < 1: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'t' の値 {value} は範囲外です。有効範囲: 1以上。", + severity="error", + hint="例: t150 のように指定してください。", + context=self._context_line(token), + ) + return + self.note_sequence.bpm = value + self._add_event(TempoEvent(tick_position=self.tick_position, bpm=value)) + + def _parse_tie(self) -> None: + token = self.ctx.advance() + next_token = self.ctx.peek() + if next_token.type not in (TokenType.NOTE, TokenType.REST, TokenType.NUMBER): + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_TIE, + line=token.line, + column=token.column, + message="タイ '&' の後に音符が見つかりません。", + severity="error", + hint="& の後に音符(例: c4)を続けてください。", + context=self._context_line(token), + ) + return + if next_token.type == TokenType.NUMBER: + self.ctx.advance() + ticks = length_value_to_ticks(int(next_token.value), 0) + self._extend_last_note(ticks) + self.tick_position += ticks + return + # For note/rest, parser will create next event; we just note that previous + # event should be merged. Simple approach: reduce tick_position and mark. + # Here we just allow the next note to overwrite position and merge later. + # Implementation: back up tick position so next note duration adds to previous. + self.tick_position = self._last_note_end() + + def _last_note_end(self) -> int: + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + for ev in reversed(ch.events): + if isinstance(ev, (NoteEvent, RestEvent)): + return ev.tick_position + return self.tick_position + + def _extend_last_note(self, ticks: int) -> None: + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + for ev in reversed(ch.events): + if isinstance(ev, (NoteEvent, RestEvent)): + ev.duration += ticks + return + + def _parse_repeat_start(self) -> None: + # Stage 2 feature: keep token but do not expand + token = self.ctx.advance() + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message="区間リピート '[' は第2段階で実装されます。", + severity="warning", + hint="現時点では無視されます。", + context=self._context_line(token), + ) + + def _parse_repeat_end(self) -> None: + token = self.ctx.advance() + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message="区間リピート ']' は第2段階で実装されます。", + severity="warning", + hint="現時点では無視されます。", + context=self._context_line(token), + ) + + def _parse_bar(self) -> None: + self.ctx.advance() + + def _add_event(self, event) -> None: # type: ignore[no-untyped-def] + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + ch.events.append(event) + + def _validate_tracks(self) -> None: + for raw, name in CHANNEL_MAP.items(): + ch = self.note_sequence.channels[name] + if name in self.tracks_seen and not ch.events: + # Determine approximate line of header (last track header token) + header_token = None + for tok in self.ctx.tokens: + if tok.type == TokenType.TRACK_HEADER and tok.value.upper() == raw: + header_token = tok + self.ctx.add_error( + code=ErrorCode.SYNTAX_EMPTY_TRACK, + line=header_token.line if header_token else 1, + column=header_token.column if header_token else 1, + message=f"トラック '{raw}' に音符または休符がありません。", + severity="error", + hint="少なくとも1つの音符または休符を記述してください。", + context=self._context_line(header_token) if header_token else "", + ) + + +def parse_ppmck( + source: str, tokens: list[Token] +) -> tuple[NoteSequence, list[ErrorDetail]]: + parser = PpmckParser(source, tokens) + return parser.parse() diff --git a/src/mml_composemusic_mcp/parser_pyxel.py b/src/mml_composemusic_mcp/parser_pyxel.py new file mode 100644 index 0000000..7d3707b --- /dev/null +++ b/src/mml_composemusic_mcp/parser_pyxel.py @@ -0,0 +1,652 @@ +"""Pyxel MML parser.""" + +from .ir import ( + DutyEvent, + EnvelopeEvent, + ErrorCode, + ErrorDetail, + GlideEvent, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, + VibratoEvent, + VolumeEvent, +) +from .lexer import Token, TokenType +from .parser_base import ( + ParserContext, + clamp, + length_value_to_ticks, + note_to_midi, +) + +CHANNEL_MAP = { + "0": "Pulse1", + "1": "Pulse2", + "2": "Triangle", + "3": "Noise", +} +PULSE_CHANNELS = {"Pulse1", "Pulse2"} + +PYXEL_DUTY = { + "1": 2, # Square 50% + "2": 1, # Pulse 25% +} + + +class PyxelParser: + def __init__(self, source: str, tokens: list[Token]) -> None: + self.source = source + self.ctx = ParserContext(tokens) + self.note_sequence = NoteSequence() + self.current_channel: str | None = None + self.octave = 4 + self.default_length = 4 + self.velocity = 100 + self.gate_time = 0.8 + self.duty = 2 + self.transpose = 0 + self.detune_cents = 0 + self.tick_position = 0 + self.tracks_seen: set[str] = set() + self.repeat_stack: list[int] = [] + + def _context_line(self, token: Token) -> str: + lines = self.source.splitlines() + if 1 <= token.line <= len(lines): + return lines[token.line - 1] + return "" + + def parse(self) -> tuple[NoteSequence, list[ErrorDetail]]: + while self.ctx.peek().type != TokenType.EOF: + token = self.ctx.peek() + if token.type == TokenType.TRACK_HEADER: + self._parse_track_header() + continue + if self.current_channel is None: + self.ctx.advance() + continue + self._parse_statement() + + if self.repeat_stack: + for _start_tick in self.repeat_stack: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_REPEAT, + line=1, + column=1, + message="リピート '[' に対応する ']' が見つかりません。", + severity="error", + hint="] を追加してリピートを閉じてください。回数指定(例: ]2)も可能です。", + context="", + ) + + self._validate_tracks() + + for ch in self.note_sequence.channels.values(): + ch.total_ticks = max( + (e.tick_position + getattr(e, "duration", 0) for e in ch.events), + default=0, + ) + + return self.note_sequence, self.ctx.errors + + def _parse_track_header(self) -> None: + token = self.ctx.advance() + ch = token.value + if ch not in CHANNEL_MAP: + self.ctx.add_error( + code=ErrorCode.SYNTAX_INVALID_TRACK_HEADER, + line=token.line, + column=token.column, + message=f"無効なトラックヘッダー '{ch}:' です。", + severity="error", + hint="0:, 1:, 2:, 3: のいずれかを使用してください。", + context=self._context_line(token), + ) + self.current_channel = None + return + self.current_channel = CHANNEL_MAP[ch] + if self.current_channel in self.tracks_seen: + self.ctx.add_error( + code=ErrorCode.SYNTAX_DUPLICATE_TRACK, + line=token.line, + column=token.column, + message=f"トラック '{ch}:' が複数回定義されています。", + severity="error", + hint="各トラックは1回のみ定義できます。重複定義を削除してください。", + context=self._context_line(token), + ) + self.tracks_seen.add(self.current_channel) + self.octave = 4 + self.default_length = 4 + self.velocity = 100 + self.gate_time = 0.8 + self.duty = 2 + self.transpose = 0 + self.detune_cents = 0 + self.tick_position = 0 + + def _parse_statement(self) -> None: + token = self.ctx.peek() + handlers = { + TokenType.NOTE: self._parse_note, + TokenType.REST: self._parse_rest, + TokenType.OCTAVE: self._parse_octave, + TokenType.OCTAVE_UP: self._parse_octave_up, + TokenType.OCTAVE_DOWN: self._parse_octave_down, + TokenType.LENGTH: self._parse_length, + TokenType.VOLUME: self._parse_volume, + TokenType.GATE_TIME: self._parse_gate_time, + TokenType.DUTY: self._parse_duty, + TokenType.AT: self._parse_at, + TokenType.TEMPO: self._parse_tempo, + TokenType.TRANSPOSE: self._parse_transpose, + TokenType.DETUNE: self._parse_detune, + TokenType.TIE: self._parse_tie, + TokenType.REPEAT_START: self._parse_repeat_start, + TokenType.REPEAT_END: self._parse_repeat_end, + TokenType.EXT_CMD: self._parse_ext_cmd, + TokenType.BAR: self._parse_bar, + } + handler = handlers.get(token.type) + if handler: + handler() + else: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"'{token.raw}' はここでは使用できません。", + severity="error", + hint="音符、休符、コマンド、小節線を配置してください。", + context=self._context_line(token), + ) + self.ctx.advance() + + def _read_length(self) -> tuple[int, int]: + dots = 0 + length_token = self.ctx.match(TokenType.NUMBER) + if length_token is None: + length = self.default_length + else: + length = int(length_token.value) + while self.ctx.match(TokenType.DOT): + dots += 1 + return length, dots + + def _parse_note(self) -> None: + token = self.ctx.advance() + note = token.value.lower() + length, dots = self._read_length() + ticks = length_value_to_ticks(length, dots) + accidental = 0 + while True: + if self.ctx.match(TokenType.SHARP): + accidental += 1 + elif self.ctx.match(TokenType.FLAT): + accidental -= 1 + else: + break + + if self.current_channel == "Noise": + midi = note_to_midi(note, self.octave, accidental) + else: + midi = note_to_midi(note, self.octave, accidental) + self.transpose + if not 0 <= midi <= 127: + self.ctx.add_error( + code=ErrorCode.SYNTAX_NOTE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"音高が範囲外です(MIDI音番号 {midi})。", + severity="warning", + hint="オクターブ/トランスポーズを調整してください。有効範囲: O0〜O7。", + context=self._context_line(token), + ) + midi = clamp(midi, 0, 127) + + event = NoteEvent( + tick_position=self.tick_position, + duration=ticks, + note_number=midi, + velocity=self._normalize_velocity(self.velocity), + duty=self.duty, + gate_time=self.gate_time, + ) + self._add_event(event) + self.tick_position += ticks + + def _parse_rest(self) -> None: + self.ctx.advance() + length, dots = self._read_length() + ticks = length_value_to_ticks(length, dots) + event = RestEvent(tick_position=self.tick_position, duration=ticks) + self._add_event(event) + self.tick_position += ticks + + def _parse_octave(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not -1 <= value <= 9: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'O' の値 {value} は範囲外です。有効範囲: -1〜9。", + severity="error", + hint="例: O4 のように指定してください。", + context=self._context_line(token), + ) + return + self.octave = value + + def _parse_octave_up(self) -> None: + self.ctx.advance() + if self.octave < 9: + self.octave += 1 + + def _parse_octave_down(self) -> None: + self.ctx.advance() + if self.octave > -1: + self.octave -= 1 + + def _parse_length(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not 1 <= value <= 192: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'L' の値 {value} は範囲外です。有効範囲: 1〜192。", + severity="error", + hint="例: L8 のように指定してください。", + context=self._context_line(token), + ) + return + self.default_length = value + + def _parse_volume(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not 0 <= value <= 127: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'V' の値 {value} は範囲外です。有効範囲: 0〜127。", + severity="error", + hint="例: V100 のように指定してください。", + context=self._context_line(token), + ) + return + self.velocity = value + self._add_event( + VolumeEvent( + tick_position=self.tick_position, + value=self._normalize_velocity(value), + ) + ) + + def _normalize_velocity(self, value: int) -> int: + return round(value / 127 * 15) + + def _parse_gate_time(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if not 0 <= value <= 100: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'Q' の値 {value} は範囲外です。有効範囲: 0〜100。", + severity="error", + hint="例: Q80 のように指定してください。", + context=self._context_line(token), + ) + return + self.gate_time = value / 100.0 + + def _parse_duty(self, cmd_raw: str = "Q") -> None: + # Q command maps to duty per Design_mml legacy section + token = self.ctx.peek() + value = int(token.value) + if not 0 <= value <= 3: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'{cmd_raw}' の値 {value} は範囲外です。有効範囲: 0〜3。", + severity="error", + hint=f"例: {cmd_raw}2 のように指定してください。", + context=self._context_line(token), + ) + self.ctx.advance() + return + if self.current_channel not in PULSE_CHANNELS: + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message=f"チャンネル '{self.current_channel}' で '{cmd_raw}' は使用できません。", + severity="warning", + hint=f"{self.current_channel}チャンネルではデューティ比を持ちません。{cmd_raw} コマンドを削除してください。", + context=self._context_line(token), + ) + self.ctx.advance() + return + self.duty = value + self._add_event(DutyEvent(tick_position=self.tick_position, value=value)) + self.ctx.advance() + + def _parse_at(self) -> None: + token = self.ctx.advance() + value = token.value + if value in PYXEL_DUTY: + if self.current_channel not in PULSE_CHANNELS: + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message=f"チャンネル '{self.current_channel}' で '@{value}' は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、Pulseチャンネル(0: または 1:)に移動してください。", + context=self._context_line(token), + ) + return + self.duty = PYXEL_DUTY[value] + self._add_event( + DutyEvent(tick_position=self.tick_position, value=self.duty) + ) + return + if value == "0": + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message=f"チャンネル '{self.current_channel}' で '@0' (Triangle) は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、適切なチャンネルに移動してください。", + context=self._context_line(token), + ) + elif value == "3": + self.ctx.add_error( + code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + line=token.line, + column=token.column, + message=f"チャンネル '{self.current_channel}' で '@3' (Noise) は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、適切なチャンネルに移動してください。", + context=self._context_line(token), + ) + else: + self.ctx.add_error( + code=ErrorCode.SYNTAX_INVALID_NUMBER, + line=token.line, + column=token.column, + message=f"'@' の後に無効な値 '{value}' が見つかりました。", + severity="error", + hint="例: @1 のように指定してください。", + context=self._context_line(token), + ) + + def _parse_tempo(self) -> None: + token = self.ctx.advance() + value = int(token.value) + if value < 1: + self.ctx.add_error( + code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + line=token.line, + column=token.column, + message=f"'T' の値 {value} は範囲外です。有効範囲: 1以上。", + severity="error", + hint="例: T150 のように指定してください。", + context=self._context_line(token), + ) + return + self.note_sequence.bpm = value + self._add_event(TempoEvent(tick_position=self.tick_position, bpm=value)) + + def _parse_transpose(self) -> None: + token = self.ctx.advance() + self.transpose = int(token.value) + + def _parse_detune(self) -> None: + token = self.ctx.advance() + self.detune_cents = int(token.value) + + def _parse_tie(self) -> None: + token = self.ctx.advance() + next_token = self.ctx.peek() + if next_token.type not in (TokenType.NOTE, TokenType.REST, TokenType.NUMBER): + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_TIE, + line=token.line, + column=token.column, + message="タイ '&' の後に音符が見つかりません。", + severity="error", + hint="& の後に音符(例: C4)を続けてください。音長のみ(例: &16)も可能です。", + context=self._context_line(token), + ) + return + if next_token.type == TokenType.NUMBER: + self.ctx.advance() + ticks = length_value_to_ticks(int(next_token.value), 0) + self._extend_last_note(ticks) + self.tick_position += ticks + return + self.tick_position = self._last_note_end() + + def _last_note_end(self) -> int: + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + for ev in reversed(ch.events): + if isinstance(ev, (NoteEvent, RestEvent)): + return ev.tick_position + return self.tick_position + + def _extend_last_note(self, ticks: int) -> None: + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + for ev in reversed(ch.events): + if isinstance(ev, (NoteEvent, RestEvent)): + ev.duration += ticks + return + + def _parse_repeat_start(self) -> None: + self.ctx.advance() + self.repeat_stack.append(self.tick_position) + + def _parse_repeat_end(self) -> None: + token = self.ctx.advance() + if not self.repeat_stack: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNMATCHED_REPEAT_END, + line=token.line, + column=token.column, + message="']' に対応する '[' が見つかりません。", + severity="error", + hint="直前の [ を確認するか、余分な ] を削除してください。", + context=self._context_line(token), + ) + return + start_tick = self.repeat_stack.pop() + count_str = token.value + if count_str == "": + count = 0 # infinite + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message="無限リピートはWAV生成時に2回で打ち切られます。", + severity="warning", + hint="回数を指定するか、2回の再生で十分であることを確認してください。", + context=self._context_line(token), + ) + else: + count = int(count_str) + segment = [ + e + for e in self.note_sequence.channels[self.current_channel].events # type: ignore[index] + if start_tick <= e.tick_position < self.tick_position + ] + repeat_count = 2 if count == 0 else count + for i in range(1, repeat_count): + offset = i * (self.tick_position - start_tick) + for ev in segment: + new_event = self._clone_event(ev, ev.tick_position + offset) + self._add_event(new_event) + if isinstance(new_event, (NoteEvent, RestEvent)): + end = new_event.tick_position + new_event.duration + if end > self.tick_position: + self.tick_position = end + self.note_sequence.channels[self.current_channel].events.sort( # type: ignore[index] + key=lambda e: e.tick_position + ) + self._add_event( + RepeatEvent( + start_tick=start_tick, + end_tick=self.tick_position, + repeat_count=count, + ) + ) + + def _clone_event(self, event, tick_position: int): # type: ignore[no-untyped-def] + import copy + + new = copy.copy(event) + new.tick_position = tick_position + return new + + def _parse_ext_cmd(self) -> None: + token = self.ctx.advance() + cmd = token.value # ENV, VIB, GLI + slot_token = self.ctx.match(TokenType.NUMBER) + if slot_token is None: + self.ctx.add_error( + code=ErrorCode.SYNTAX_INVALID_NUMBER, + line=token.line, + column=token.column, + message=f"'@{cmd}' の後にスロット番号が必要です。", + severity="error", + hint=f"例: @{cmd}1 のように指定してください。", + context=self._context_line(token), + ) + return + slot = int(slot_token.value) + if slot == 0: + # OFF + if cmd == "ENV": + ev = EnvelopeEvent(tick_position=self.tick_position, slot=0) + elif cmd == "VIB": + ev = VibratoEvent(tick_position=self.tick_position, slot=0) + else: + ev = GlideEvent(tick_position=self.tick_position, slot=0) + self._add_event(ev) + return + if self.ctx.match(TokenType.NUMBER): + # definition without braces; simplistic parse numbers until non-number + # Real Pyxel uses { ... }; we accept a flat sequence of ints for now. + values = [int(slot_token.value)] + while True: + num = self.ctx.match(TokenType.NUMBER) + if num is None: + break + values.append(int(num.value)) + if cmd == "ENV": + points = [] + for i in range(1, len(values), 2): + duration = values[i] + target = values[i - 1] if i - 1 < len(values) else values[-1] + points.append({"target_volume": target, "duration_ticks": duration}) + ev = EnvelopeEvent( + tick_position=self.tick_position, + slot=slot, + points=points, + ) + elif cmd == "VIB": + if len(values) >= 3: + params = { + "delay_ticks": values[0], + "period_ticks": values[1], + "depth_cents": values[2], + } + else: + params = {} + ev = VibratoEvent( + tick_position=self.tick_position, + slot=slot, + params=params, + ) + else: + if len(values) >= 2: + params = { + "initial_offset_cents": values[0], + "duration_ticks": values[1], + } + else: + params = {} + ev = GlideEvent( + tick_position=self.tick_position, + slot=slot, + params=params, + ) + self._add_event(ev) + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"@{cmd} は第1段階では未サポート、無視されます。", + severity="warning", + hint="合成には反映されませんが、IRとして保持されます。", + context=self._context_line(token), + ) + return + # Slot switch only + if cmd == "ENV": + ev = EnvelopeEvent(tick_position=self.tick_position, slot=slot) + elif cmd == "VIB": + ev = VibratoEvent(tick_position=self.tick_position, slot=slot) + else: + ev = GlideEvent(tick_position=self.tick_position, slot=slot) + self._add_event(ev) + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"@{cmd} は第1段階では未サポート、無視されます。", + severity="warning", + hint="合成には反映されませんが、IRとして保持されます。", + context=self._context_line(token), + ) + + def _parse_bar(self) -> None: + self.ctx.advance() + + def _add_event(self, event) -> None: # type: ignore[no-untyped-def] + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + ch.events.append(event) + + def _validate_tracks(self) -> None: + for raw, name in CHANNEL_MAP.items(): + ch = self.note_sequence.channels[name] + if name in self.tracks_seen and not ch.events: + header_token = None + for tok in self.ctx.tokens: + if tok.type == TokenType.TRACK_HEADER and tok.value == raw: + header_token = tok + self.ctx.add_error( + code=ErrorCode.SYNTAX_EMPTY_TRACK, + line=header_token.line if header_token else 1, + column=header_token.column if header_token else 1, + message=f"トラック '{raw}:' に音符または休符がありません。", + severity="error", + hint="少なくとも1つの音符または休符を記述してください。", + context=self._context_line(header_token) if header_token else "", + ) + + +def parse_pyxel( + source: str, tokens: list[Token] +) -> tuple[NoteSequence, list[ErrorDetail]]: + parser = PyxelParser(source, tokens) + return parser.parse() diff --git a/src/mml_composemusic_mcp/server.py b/src/mml_composemusic_mcp/server.py new file mode 100644 index 0000000..1205b70 --- /dev/null +++ b/src/mml_composemusic_mcp/server.py @@ -0,0 +1,254 @@ +# MCP server exposing compose_mml tool. + +import argparse +from pathlib import Path + +from fastmcp import FastMCP + +from .ir import ErrorCode, ErrorDetail, NoteSequence +from .lexer import tokenize +from .parser_ppmck import parse_ppmck +from .parser_pyxel import parse_pyxel +from .synthesizer import build_channel_summary, synthesize, write_wav +from .templates import get_template + +mcp = FastMCP("mml-composemusic-mcp") +OUTPUT_DIR = Path("./data") + + +def _is_error(errors: list[ErrorDetail]) -> bool: + return any(e.severity == "error" for e in errors) + + +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 _parse_mml(mml: str, mode: str) -> tuple[dict, list[ErrorDetail]]: + tokens = tokenize(mml, mode) + if mode == "ppmck": + note_sequence, errors = parse_ppmck(mml, tokens) + elif mode == "pyxel": + note_sequence, errors = parse_pyxel(mml, tokens) + else: + errors = [ + ErrorDetail( + code=ErrorCode.SYNTAX_INVALID_TOKEN, + line=0, + column=0, + message=f"未知のモード '{mode}' です。", + severity="error", + hint="mode は 'ppmck' または 'pyxel' を指定してください。", + ) + ] + note_sequence = None + return (note_sequence.to_dict() if note_sequence else None), errors + + +@mcp.tool() +def compose_mml( + action: str, + mml: str = "", + mode: str = "", + template: str = "basic", + sample_rate: int = 44100, + normalize: bool = True, +) -> dict: + """Compose, validate, or generate templates for Famicom-style MML.""" + if action == "template": + if mode not in ("ppmck", "pyxel"): + mode = "ppmck" + if template not in ("basic", "melody", "chord", "drum", "empty"): + template = "basic" + mml_text, description = get_template(mode, template) + return {"mml": mml_text, "description": description} + + if action in ("compose", "validate"): + if not mml or not mode: + return { + "success": False, + "valid": False, + "errors": [ + { + "code": ErrorCode.SYNTAX_INVALID_TOKEN.value, + "line": 0, + "column": 0, + "message": "mml と mode は compose/validate の必須パラメータです。", + "severity": "error", + "hint": "mml に MML 文字列、mode に 'ppmck' または 'pyxel' を指定してください。", + } + ], + "warnings": [], + } + note_sequence_dict, errors = _parse_mml(mml, mode) + errors_list, warnings_list = _split_errors(errors) + if action == "validate": + return { + "valid": not _is_error(errors), + "errors": errors_list, + "note_sequence": note_sequence_dict, + "channel_summary": build_channel_summary(note_sequence_dict) + if note_sequence_dict + else [], + } + # compose + if _is_error(errors): + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + + ns = _dict_to_note_sequence(note_sequence_dict) + try: + wave_data, duration, synth_errors = synthesize( + ns, mode, sample_rate, normalize + ) + except Exception as exc: + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": { + "errors": [ + { + "code": ErrorCode.SYSTEM_SYNTHESIS_FAILED.value, + "line": 0, + "column": 0, + "message": f"音声合成中にエラーが発生しました: {exc}", + "severity": "error", + "hint": "MMLの内容を確認の上、再度お試しください。問題が続く場合は、短いMMLから試してください。", + } + ], + "warnings": warnings_list, + }, + } + errors.extend(synth_errors) + errors_list, warnings_list = _split_errors(errors) + if _is_error(errors): + return { + "success": False, + "wav_path": None, + "duration_sec": 0, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + wav_path = OUTPUT_DIR / "output.wav" + wav_errors = write_wav(wav_path, wave_data, sample_rate) + errors.extend(wav_errors) + errors_list, warnings_list = _split_errors(errors) + return { + "success": not _is_error(errors), + "wav_path": str(wav_path) if not _is_error(errors) else None, + "duration_sec": duration, + "note_sequence": note_sequence_dict, + "validation": {"errors": errors_list, "warnings": warnings_list}, + } + + return { + "success": False, + "errors": [ + { + "code": ErrorCode.SYNTAX_INVALID_TOKEN.value, + "line": 0, + "column": 0, + "message": f"未知の action '{action}' です。", + "severity": "error", + "hint": "action は 'compose', 'validate', 'template' のいずれかを指定してください。", + } + ], + "warnings": [], + } + + +def _dict_to_note_sequence(data: dict) -> NoteSequence: + from .ir import ( + ChannelSequence, + DutyEvent, + EnvelopeEvent, + GlideEvent, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, + VibratoEvent, + VolumeEvent, + ) + + ns = NoteSequence( + version=data["version"], + bpm=data["bpm"], + ticks_per_quarter=data["ticks_per_quarter"], + 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 == "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) + mcp.run(transport=args.transport, host=args.host, port=args.port) + + +if __name__ == "__main__": + main() diff --git a/src/mml_composemusic_mcp/synthesizer.py b/src/mml_composemusic_mcp/synthesizer.py new file mode 100644 index 0000000..42a3c11 --- /dev/null +++ b/src/mml_composemusic_mcp/synthesizer.py @@ -0,0 +1,301 @@ +"""NES APU-style synthesizer: NoteSequence -> WAV.""" + +import wave +from math import floor +from pathlib import Path + +import numpy as np + +from .ir import ( + DutyEvent, + ErrorCode, + ErrorDetail, + NoteEvent, + NoteSequence, + RestEvent, + TempoEvent, + VolumeEvent, +) + +CPU_CLOCK = 1_789_772.5 +SAMPLE_RATE_DEFAULT = 44100 + + +def midi_to_freq(note_number: int) -> float: + return 440.0 * 2.0 ** ((note_number - 69) / 12.0) + + +def noise_period_to_rate(period: int, mode: int, sample_rate: int) -> float: + # NES noise: rate = CPU_CLOCK / (NTSC divider for period) + # Approximate using pyxel mapping described in design + dividers = [ + 4, + 8, + 16, + 32, + 64, + 96, + 128, + 160, + 202, + 254, + 380, + 508, + 762, + 1016, + 2034, + 4068, + ] + div = dividers[period & 0x0F] + # Long mode (0) vs short mode (1) affects LFSR length, not frequency divider + return CPU_CLOCK / div + + +class ChannelSynthesizer: + def __init__( + self, channel_type: str, sample_rate: int = SAMPLE_RATE_DEFAULT + ) -> None: + self.channel_type = channel_type + self.sample_rate = sample_rate + self.wave = np.zeros(0, dtype=np.float64) + + def render( + self, + events: list, + total_ticks: int, + bpm: int, + mode: str, + ) -> np.ndarray: + seconds = tick_to_second(total_ticks, bpm) + length = int(seconds * self.sample_rate) + self.wave = np.zeros(length, dtype=np.float64) + last_velocity = 15 if self.channel_type != "triangle" else 15 + last_duty = 2 + for event in events: + if isinstance(event, VolumeEvent): + last_velocity = event.value + continue + if isinstance(event, DutyEvent): + last_duty = event.value + continue + if isinstance(event, TempoEvent): + bpm = event.bpm + continue + if not isinstance(event, (NoteEvent, RestEvent)): + continue + start_sample = int( + tick_to_second(event.tick_position, bpm) * self.sample_rate + ) + end_tick = event.tick_position + event.duration + end_sample = min( + int(tick_to_second(end_tick, bpm) * self.sample_rate), length + ) + if start_sample >= length or end_sample <= start_sample: + continue + if isinstance(event, RestEvent): + continue + self._render_note( + start_sample, + end_sample, + event, + last_velocity, + last_duty, + mode, + ) + return self.wave + + def _render_note( + self, + start: int, + end: int, + event: NoteEvent, + velocity: int, + duty: int, + mode: str, + ) -> None: + duration_samples = end - start + if duration_samples <= 0: + return + if self.channel_type == "noise": + self._render_noise(start, end, event, velocity, mode) + return + if self.channel_type == "triangle": + self._render_triangle(start, end, event, velocity, mode) + return + self._render_pulse(start, end, event, velocity, duty) + + def _render_pulse( + self, + start: int, + end: int, + event: NoteEvent, + velocity: int, + duty: int, + ) -> None: + freq = midi_to_freq(event.note_number) + if freq <= 0: + return + phase_inc = freq / self.sample_rate + duty_cycle = {0: 0.125, 1: 0.25, 2: 0.5, 3: 0.75}.get(duty, 0.5) + samples = np.arange(end - start) + phase = (samples * phase_inc) % 1.0 + waveform = np.where(phase < duty_cycle, 1.0, -1.0) + amp = velocity / 15.0 + gate = max(0.0, min(1.0, event.gate_time)) + if gate < 1.0: + gate_sample = int((end - start) * gate) + if gate_sample < len(waveform): + waveform[gate_sample:] = 0.0 + self.wave[start:end] += waveform * amp + + def _render_triangle( + self, + start: int, + end: int, + event: NoteEvent, + velocity: int, + mode: str, + ) -> None: + freq = midi_to_freq(event.note_number) + if freq <= 0: + return + phase_inc = freq / self.sample_rate + samples = np.arange(end - start) + phase = (samples * phase_inc) % 1.0 + # 32-step triangle approximated by abs(phase*2-1) + waveform = 2.0 * np.abs(phase * 2.0 - 1.0) - 1.0 + if mode == "pyxel": + amp = velocity / 15.0 + else: + amp = 1.0 + gate = max(0.0, min(1.0, event.gate_time)) + if gate < 1.0: + gate_sample = int((end - start) * gate) + if gate_sample < len(waveform): + waveform[gate_sample:] = 0.0 + self.wave[start:end] += waveform * amp + + def _render_noise( + self, + start: int, + end: int, + event: NoteEvent, + velocity: int, + mode: str, + ) -> None: + if mode == "pyxel": + period = max(0, min(15, 15 - floor(event.note_number / 8))) + mode_flag = 1 if period >= 13 else 0 + else: + period = 8 + mode_flag = 0 + rate = noise_period_to_rate(period, mode_flag, self.sample_rate) + phase_inc = rate / self.sample_rate + samples = np.arange(end - start) + phase = (samples * phase_inc) % 1.0 + diff = np.diff(phase, prepend=0) + transitions = diff < 0 + # LFSR-like pseudo-random toggling on each period rollover + state = np.zeros(len(samples), dtype=np.int32) + lfsr = 1 + for i in range(len(samples)): + if transitions[i]: + # feedback bit + if mode_flag: + bit = (lfsr ^ (lfsr >> 6)) & 1 + lfsr = ((lfsr >> 1) | (bit << 14)) & 0x7FFF + else: + bit = (lfsr ^ (lfsr >> 1)) & 1 + lfsr = ((lfsr >> 1) | (bit << 14)) & 0x7FFF + state[i] = lfsr & 1 + waveform = np.where(state, 1.0, -1.0) + amp = velocity / 15.0 + gate = max(0.0, min(1.0, event.gate_time)) + if gate < 1.0: + gate_sample = int((end - start) * gate) + if gate_sample < len(waveform): + waveform[gate_sample:] = 0.0 + self.wave[start:end] += waveform * amp + + +def tick_to_second(tick: int, bpm: int, ticks_per_quarter: int = 192) -> float: + quarter_seconds = 60.0 / bpm + return tick * (quarter_seconds / ticks_per_quarter) + + +def synthesize( + note_sequence: NoteSequence, + mode: str, + sample_rate: int = SAMPLE_RATE_DEFAULT, + normalize: bool = True, +) -> tuple[np.ndarray, float, list[ErrorDetail]]: + errors: list[ErrorDetail] = [] + total_ticks = max(ch.total_ticks for ch in note_sequence.channels.values()) + if total_ticks <= 0: + return np.zeros(0, dtype=np.float64), 0.0, errors + channels = [] + for _name, ch in note_sequence.channels.items(): + synth = ChannelSynthesizer(ch.channel_type, sample_rate) + wave = synth.render(ch.events, total_ticks, note_sequence.bpm, mode) + channels.append(wave) + max_len = max(len(w) for w in channels) + padded = [np.pad(w, (0, max_len - len(w)), mode="constant") for w in channels] + mixed = np.mean(padded, axis=0) + peak = np.max(np.abs(mixed)) + if normalize and peak > 0: + mixed = mixed / peak + mixed = np.clip(mixed, -1.0, 1.0) + duration = max_len / sample_rate + return mixed, duration, errors + + +def write_wav(path: Path, wave_data: np.ndarray, sample_rate: int) -> list[ErrorDetail]: + errors: list[ErrorDetail] = [] + try: + path.parent.mkdir(parents=True, exist_ok=True) + pcm = np.int16(np.clip(wave_data * 32767, -32768, 32767)) + with wave.open(str(path), "wb") as wav: + wav.setnchannels(1) + wav.setsampwidth(2) + wav.setframerate(sample_rate) + wav.writeframes(pcm.tobytes()) + except Exception as exc: + errors.append( + ErrorDetail( + code=ErrorCode.SYSTEM_WAV_WRITE_FAILED, + line=0, + column=0, + message=f"WAVファイルの出力に失敗しました: {exc}", + severity="error", + hint="しばらく待ってから再度お試しください。", + ) + ) + return errors + + +def build_channel_summary(note_sequence: NoteSequence | dict) -> list[dict]: + if isinstance(note_sequence, dict): + channels = note_sequence["channels"] + else: + channels = note_sequence.channels + summary = [] + for name, ch in channels.items(): + events = ch["events"] if isinstance(ch, dict) else ch.events + total_ticks = ch["total_ticks"] if isinstance(ch, dict) else ch.total_ticks + notes = [e for e in events if isinstance(e, NoteEvent)] + if not notes: + notes = [ + e for e in events if isinstance(e, dict) and e.get("type") == "note" + ] + octaves = [e["note_number"] // 12 - 1 for e in notes] + else: + octaves = [e.note_number // 12 - 1 for e in notes] + summary.append( + { + "channel": name, + "note_count": len(notes), + "octave_range": [min(octaves, default=4), max(octaves, default=4)], + "duration_ticks": total_ticks, + } + ) + return summary diff --git a/src/mml_composemusic_mcp/templates.py b/src/mml_composemusic_mcp/templates.py new file mode 100644 index 0000000..f351583 --- /dev/null +++ b/src/mml_composemusic_mcp/templates.py @@ -0,0 +1,32 @@ +# MML template generators for both modes. + +TEMPLATES = { + "ppmck": { + "basic": '#TITLE "Basic Song"\n#COMPOSER "LLM"\n\nA t120 l8 o4 v15 q2\n c d e f | g a b > c\n\nB l8 o3 v12 q1\n e g b r | e g b r\n\nT l4 o3 v7\n c2 g2 e2 c2\n\nN l8 v10\n c r c r | c r c r\n', + "melody": '#TITLE "Melody Lead"\n#COMPOSER "LLM"\n\nA t120 l8 o5 v15 q2\n c e g e | d f a f | e g b g | c4 r4\n\nB l8 o4 v10 q1\n c e g e | c e g e | d f a f | c4 r4\n\nT l4 o3 v7\n c2 g2 | a2 f2 | g2 d2 | c2 r2\n\nN l8 v8\n c r c r | c r c r | c r c r | c r c r\n', + "chord": '#TITLE "Chord Backing"\n#COMPOSER "LLM"\n\nA t120 l4 o4 v12 q2\n c e g c | f a > c f | g b > d g | c e g c\n\nB l4 o4 v12 q1\n e g b e | f a > c f | d g b d | e g b e\n\nT l2 o3 v7\n c2 | f2 | g2 | c2\n\nN l8 v8\n r c r c | r c r c | r c r c | r c r c\n', + "drum": '#TITLE "Drum Beat"\n#COMPOSER "LLM"\n\nA t120 l8 o4 v15 q2\n c r e r | g r e r | c r e r | g r c r\n\nB l8 o3 v12 q1\n e r g r | e r g r | e r g r | e r g r\n\nT l4 o2 v7\n c2 c2 | c2 c2 | c2 c2 | c2 c2\n\nN l16 v10\n c r c r c r c r | c r c r c r c r\n c r c r c r c r | c r c r c r c r\n', + "empty": '#TITLE "Empty Template"\n#COMPOSER "LLM"\n\nA t120 l4 o4 v15 q2\n \n\nB l4 o3 v12 q1\n \n\nT l4 o3 v7\n \n\nN l4 v10\n \n', + }, + "pyxel": { + "basic": "0: T120 L8 O4 V100 @1\n C D E F | G A B >C\n\n1: L8 O3 V80 @2\n E G B R | E G B R\n\n2: L4 O3 V60\n C2 G2 E2 C2\n\n3: L8 V80\n C R C R | C R C R\n", + "melody": "0: T120 L8 O5 V100 @1\n C E G E | D F A F | E G B G | C4 R4\n\n1: L8 O4 V80 @2\n C E G E | C E G E | D F A F | C4 R4\n\n2: L4 O3 V60\n C2 G2 | A2 F2 | G2 D2 | C2 R2\n\n3: L8 V70\n C R C R | C R C R | C R C R | C R C R\n", + "chord": "0: T120 L4 O4 V90 @1\n C E G C | F A >C F | G B >D G | C E G C\n\n1: L4 O4 V90 @2\n E G B E | F A >C F | D G B D | E G B E\n\n2: L2 O3 V60\n C2 | F2 | G2 | C2\n\n3: L8 V70\n R C R C | R C R C | R C R C | R C R C\n", + "drum": "0: T120 L8 O4 V100 @1\n C R E R | G R E R | C R E R | G R C R\n\n1: L8 O3 V80 @2\n E R G R | E R G R | E R G R | E R G R\n\n2: L4 O2 V60\n C2 C2 | C2 C2 | C2 C2 | C2 C2\n\n3: L16 V80\n C R C R C R C R | C R C R C R C R\n C R C R C R C R | C R C R C R C R\n", + "empty": "0: T120 L4 O4 V100 @1\n \n\n1: L4 O3 V80 @2\n \n\n2: L4 O3 V60\n \n\n3: L4 V80\n \n", + }, +} + +DESCRIPTIONS = { + "basic": "基本的な4ch構成(メロディ+和音+ベース+リズム)", + "melody": "メロディ重視(Pulse1主旋律、他は伴奏最小限)", + "chord": "コード伴奏重視(Pulse2で和音、Triangleでベース)", + "drum": "リズム重視(Noise中心のビートパターン)", + "empty": "各チャンネルのヘッダーのみ(空のテンプレート)", +} + + +def get_template(mode: str, template: str) -> tuple[str, str]: + mml = TEMPLATES[mode][template] + desc = DESCRIPTIONS[template] + return mml, desc |
