summaryrefslogtreecommitdiff
path: root/src/mml_composemusic_mcp
diff options
context:
space:
mode:
authortanaka takashi <isazaka@gmail.com>2026-07-11 19:02:14 +0900
committertanaka takashi <isazaka@gmail.com>2026-07-11 19:02:14 +0900
commitca35520338057512a71acd42bb116ffbb82cc192 (patch)
tree9faa2428aeb71b62497f21ceec93b99ef7f2a5ac /src/mml_composemusic_mcp
parent7dca1616b8c891f5e140c7267323d1fcdde6741a (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/mml_composemusic_mcp')
-rw-r--r--src/mml_composemusic_mcp/__init__.py0
-rw-r--r--src/mml_composemusic_mcp/ir.py185
-rw-r--r--src/mml_composemusic_mcp/lexer.py314
-rw-r--r--src/mml_composemusic_mcp/parser_base.py92
-rw-r--r--src/mml_composemusic_mcp/parser_ppmck.py451
-rw-r--r--src/mml_composemusic_mcp/parser_pyxel.py652
-rw-r--r--src/mml_composemusic_mcp/server.py254
-rw-r--r--src/mml_composemusic_mcp/synthesizer.py301
-rw-r--r--src/mml_composemusic_mcp/templates.py32
9 files changed, 2281 insertions, 0 deletions
diff --git a/src/mml_composemusic_mcp/__init__.py b/src/mml_composemusic_mcp/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ 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