From 22f96694a96ed9fba169e77bc27d19d00edca083 Mon Sep 17 00:00:00 2001 From: tanaka takashi Date: Sun, 12 Jul 2026 16:03:29 +0900 Subject: MML構文解析処理を追加 MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- src/mml_composemusic_mcp/ast_nodes.py | 230 ++++++++ src/mml_composemusic_mcp/ir.py | 49 +- src/mml_composemusic_mcp/lexer.py | 62 +-- src/mml_composemusic_mcp/parser_base.py | 45 +- src/mml_composemusic_mcp/parser_ppmck.py | 618 +++++++++------------- src/mml_composemusic_mcp/parser_pyxel.py | 815 ++++++++++------------------- src/mml_composemusic_mcp/semantic_base.py | 225 ++++++++ src/mml_composemusic_mcp/semantic_ppmck.py | 283 ++++++++++ src/mml_composemusic_mcp/semantic_pyxel.py | 423 +++++++++++++++ src/mml_composemusic_mcp/server.py | 30 +- src/mml_composemusic_mcp/synthesizer.py | 4 +- 11 files changed, 1809 insertions(+), 975 deletions(-) create mode 100644 src/mml_composemusic_mcp/ast_nodes.py create mode 100644 src/mml_composemusic_mcp/semantic_base.py create mode 100644 src/mml_composemusic_mcp/semantic_ppmck.py create mode 100644 src/mml_composemusic_mcp/semantic_pyxel.py (limited to 'src') diff --git a/src/mml_composemusic_mcp/ast_nodes.py b/src/mml_composemusic_mcp/ast_nodes.py new file mode 100644 index 0000000..bc4fdc4 --- /dev/null +++ b/src/mml_composemusic_mcp/ast_nodes.py @@ -0,0 +1,230 @@ +"""Common AST node definitions for ppmck and pyxel MML.""" + +from dataclasses import dataclass, field +from typing import Any + + +@dataclass +class ASTNode: + """Base class for all AST nodes.""" + + line: int + column: int + + +@dataclass +class Program(ASTNode): + """Root program node.""" + + tracks: list[Track] = field(default_factory=list) + + +@dataclass +class Track(ASTNode): + """A single track definition.""" + + track_id: str # "A" / "B" / "T" / "N" / "L" (ppmck) or "0"-"3" (pyxel) + channel: str # "Pulse1" / "Pulse2" / "Triangle" / "Noise" / "Loop" + mode: str # "ppmck" / "pyxel" + statements: list[Statement] = field(default_factory=list) + headers: list[Header] = field(default_factory=list) + + +@dataclass +class Header(ASTNode): + """ppmck # header line.""" + + key: str + value: str + unterminated: bool = False + + +# --- Statements --- + + +@dataclass +class NoteStmt(ASTNode): + """A note command (c, d, e, ...).""" + + note_name: str # normalized to lowercase: "c"-"b" + accidental: int = 0 # +1 sharp, -1 flat, 0 none + length: int | None = None # None means use default length + dots: int = 0 + + +@dataclass +class RestStmt(ASTNode): + """A rest command (r / R).""" + + length: int | None = None + dots: int = 0 + + +@dataclass +class OctaveStmt(ASTNode): + """Octave command or relative octave shift.""" + + value: int | None = None # absolute octave number + direction: str | None = None # "up" or "down" for relative shifts + + +@dataclass +class LengthStmt(ASTNode): + """Default length command (l / L).""" + + value: int + + +@dataclass +class VolumeStmt(ASTNode): + """Volume command (v / V).""" + + value: int + + +@dataclass +class DutyStmt(ASTNode): + """Duty/tone command (q / @).""" + + value: int + + +@dataclass +class TempoStmt(ASTNode): + """Tempo command (t / T).""" + + value: int + + +@dataclass +class GateTimeStmt(ASTNode): + """Gate time command (Q, pyxel only).""" + + value: int + + +@dataclass +class TransposeStmt(ASTNode): + """Transpose command (K, pyxel only).""" + + value: int + + +@dataclass +class DetuneStmt(ASTNode): + """Detune command (Y, pyxel only).""" + + value: int + + +@dataclass +class TieStmt(ASTNode): + """Tie/slur command (&).""" + + target: NoteStmt | RestStmt | int | None = None + + +@dataclass +class RepeatStartStmt(ASTNode): + """Start of repeat section ([).""" + + pass + + +@dataclass +class RepeatEndStmt(ASTNode): + """End of repeat section (] with optional count).""" + + count: int | None = None # None means infinite + + +@dataclass +class BarStmt(ASTNode): + """Bar line (|), visual only.""" + + pass + + +@dataclass +class ExtCmdStmt(ASTNode): + """pyxel @ENV / @VIB / @GLI command.""" + + cmd: str # "ENV" / "VIB" / "GLI" + slot: int + params: list[int] = field(default_factory=list) + + +# --- Stage 2 statements (kept in AST but ignored by semantic analyzer) --- + + +@dataclass +class CountLengthStmt(ASTNode): + """ppmck count length note (c%48).""" + + note_name: str + accidental: int + count: int + + +@dataclass +class EnvelopeDefStmt(ASTNode): + """ppmck envelope definition (@v0 = { ... }).""" + + slot: int + points: list[int] = field(default_factory=list) + loop_points: list[int] = field(default_factory=list) + + +@dataclass +class SweepStmt(ASTNode): + """ppmck sweep command (s1,2).""" + + speed: int + depth: int + + +@dataclass +class NoiseModeStmt(ASTNode): + """ppmck noise mode command (m0 / m1).""" + + mode: int + + +Statement = ( + NoteStmt + | RestStmt + | OctaveStmt + | LengthStmt + | VolumeStmt + | DutyStmt + | TempoStmt + | GateTimeStmt + | TransposeStmt + | DetuneStmt + | TieStmt + | RepeatStartStmt + | RepeatEndStmt + | BarStmt + | ExtCmdStmt + | CountLengthStmt + | EnvelopeDefStmt + | SweepStmt + | NoiseModeStmt +) + + +def _node_to_dict(node: Any) -> Any: + """Recursively convert an AST node to a JSON-compatible dict.""" + if isinstance(node, list): + return [_node_to_dict(item) for item in node] + if not isinstance(node, ASTNode): + return node + data: dict[str, Any] = {"__type__": node.__class__.__name__} + for key, value in node.__dict__.items(): + data[key] = _node_to_dict(value) + return data + + +def ast_to_dict(node: ASTNode) -> dict[str, Any]: + """Convert an AST node to a JSON-compatible dict.""" + return _node_to_dict(node) diff --git a/src/mml_composemusic_mcp/ir.py b/src/mml_composemusic_mcp/ir.py index bd741a6..8d1bc25 100644 --- a/src/mml_composemusic_mcp/ir.py +++ b/src/mml_composemusic_mcp/ir.py @@ -5,30 +5,53 @@ from enum import Enum from typing import Any +class ErrorPhase(Enum): + """Phase where an error occurred.""" + + LEXER = "lexer" + SYNTAX = "syntax" + SEMANTIC = "semantic" + RUNTIME = "runtime" + API = "api" + + class ErrorCode(Enum): - """Machine-readable error codes.""" + """Machine-readable error codes. + Errors are classified by phase: + - SYNTAX_*: token sequence violates the BNF grammar. + - SEMANTIC_*: syntactically valid but semantically invalid (range, + channel mismatch, undefined reference, etc.). + - RUNTIME_*: failure during synthesis or WAV output. + - VALIDATION_*: API-level parameter validation errors. + """ + + # --- Syntax errors (Parser phase) --- 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_OUTSIDE_TRACK = "SYNTAX_OUTSIDE_TRACK" 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" - # API-level errors (parameter validation, not MML syntax) + # --- Semantic errors (SemanticAnalyzer phase) --- + SEMANTIC_VALUE_OUT_OF_RANGE = "SEMANTIC_VALUE_OUT_OF_RANGE" + SEMANTIC_NOTE_OUT_OF_RANGE = "SEMANTIC_NOTE_OUT_OF_RANGE" + SEMANTIC_CHANNEL_MISMATCH = "SEMANTIC_CHANNEL_MISMATCH" + SEMANTIC_EMPTY_TRACK = "SEMANTIC_EMPTY_TRACK" + SEMANTIC_OUTSIDE_TRACK = "SEMANTIC_OUTSIDE_TRACK" + SEMANTIC_UNDEFINED_REFERENCE = "SEMANTIC_UNDEFINED_REFERENCE" + SEMANTIC_UNSUPPORTED_FEATURE = "SEMANTIC_UNSUPPORTED_FEATURE" + + # --- Runtime errors (Synthesizer phase) --- + RUNTIME_SYNTHESIS_FAILED = "RUNTIME_SYNTHESIS_FAILED" + RUNTIME_WAV_WRITE_FAILED = "RUNTIME_WAV_WRITE_FAILED" + RUNTIME_INTERNAL_ERROR = "RUNTIME_INTERNAL_ERROR" + + # --- API-level errors (parameter validation, not MML syntax) --- VALIDATION_MISSING_PARAMETER = "VALIDATION_MISSING_PARAMETER" VALIDATION_INVALID_MODE = "VALIDATION_INVALID_MODE" VALIDATION_INVALID_ACTION = "VALIDATION_INVALID_ACTION" @@ -39,6 +62,7 @@ class ErrorDetail: """Structured error/warning detail.""" code: ErrorCode + phase: ErrorPhase line: int column: int message: str @@ -49,6 +73,7 @@ class ErrorDetail: def to_dict(self) -> dict[str, Any]: return { "code": self.code.value, + "phase": self.phase.value, "line": self.line, "column": self.column, "message": self.message, diff --git a/src/mml_composemusic_mcp/lexer.py b/src/mml_composemusic_mcp/lexer.py index 05861cf..44e577a 100644 --- a/src/mml_composemusic_mcp/lexer.py +++ b/src/mml_composemusic_mcp/lexer.py @@ -27,11 +27,11 @@ class TokenType(Enum): SHARP = auto() FLAT = auto() DOT = auto() - TEXT = auto() AT = auto() TRANSPOSE = auto() DETUNE = auto() EXT_CMD = auto() # @ENV, @VIB, @GLI + INVALID = auto() # unrecognised character EOF = auto() @@ -89,18 +89,10 @@ class Lexer: 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 _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(self) -> list[Token]: while self.pos < len(self.source): @@ -108,22 +100,14 @@ class Lexer: 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) + self._tokenize_ppmck(ch) else: - self._tokenize_pyxel(ch, start_line, start_col) + self._tokenize_pyxel(ch) 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 + def _tokenize_ppmck(self, ch: str) -> None: if ch == "\n": self._advance() return @@ -182,36 +166,36 @@ class Lexer: return if ch.isalpha(): cmd = self._advance() - if ch in "olvtq": + lower = cmd.lower() + if lower in "olvtq": num = self._read_number() raw = cmd + num - if ch == "o": + if lower == "o": self._emit(TokenType.OCTAVE, num, raw) - elif ch == "l": + elif lower == "l": self._emit(TokenType.LENGTH, num, raw) - elif ch == "v": + elif lower == "v": self._emit(TokenType.VOLUME, num, raw) - elif ch == "t": + elif lower == "t": self._emit(TokenType.TEMPO, num, raw) - elif ch == "q": + elif lower == "q": self._emit(TokenType.DUTY, num, raw) return - # Note commands - note = ch.lower() + note = 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) + # Unknown command: emit as invalid token for parser to report + self._emit(TokenType.INVALID, cmd, cmd) return # Fallback invalid char raw = self._advance() - self._emit(TokenType.TEXT, raw, raw) + self._emit(TokenType.INVALID, raw, raw) - def _tokenize_pyxel(self, ch: str, start_line: int, start_col: int) -> None: # noqa: ARG002 + def _tokenize_pyxel(self, ch: str) -> None: if ch == "\n": self._advance() return @@ -299,11 +283,11 @@ class Lexer: else: self._emit(TokenType.NOTE, note, note) return - self._emit(TokenType.TEXT, cmd, cmd) + self._emit(TokenType.INVALID, cmd, cmd) return - # Fallback + # Fallback invalid char raw = self._advance() - self._emit(TokenType.TEXT, raw, raw) + self._emit(TokenType.INVALID, raw, raw) def tokenize(source: str, mode: str = "ppmck") -> list[Token]: diff --git a/src/mml_composemusic_mcp/parser_base.py b/src/mml_composemusic_mcp/parser_base.py index 0427a4c..1cdb40f 100644 --- a/src/mml_composemusic_mcp/parser_base.py +++ b/src/mml_composemusic_mcp/parser_base.py @@ -1,8 +1,9 @@ """Base parser utilities shared between ppmck and pyxel parsers.""" -from dataclasses import dataclass +from dataclasses import dataclass, field -from .ir import ErrorCode, ErrorDetail +from .ast_nodes import ASTNode +from .ir import ErrorCode, ErrorDetail, ErrorPhase from .lexer import Token, TokenType @@ -10,11 +11,7 @@ from .lexer import Token, TokenType 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 = [] + errors: list[ErrorDetail] = field(default_factory=list) def peek(self, offset: int = 0) -> Token: idx = self.pos + offset @@ -46,9 +43,11 @@ class ParserContext: hint: str = "", context: str = "", ) -> None: + phase = _code_to_phase(code) self.errors.append( ErrorDetail( code=code, + phase=phase, line=line, column=column, message=message, @@ -58,11 +57,34 @@ class ParserContext: ) ) + def add_invalid_token_error(self, token: Token, context: str = "") -> None: + self.add_error( + code=ErrorCode.SYNTAX_INVALID_TOKEN, + line=token.line, + column=token.column, + message=f"無効な文字 '{token.raw}' が見つかりました。", + severity="error", + hint="MMLコマンド（c,d,e,f,g,a,b,r,o,l,v,t,q など）を使用してください。", + context=context, + ) + class ParserError(Exception): + """Raised by parsers to abort a single statement and recover.""" + pass +def _code_to_phase(code: ErrorCode) -> ErrorPhase: + if code.value.startswith("SYNTAX_"): + return ErrorPhase.SYNTAX + if code.value.startswith("SEMANTIC_"): + return ErrorPhase.SEMANTIC + if code.value.startswith("RUNTIME_"): + return ErrorPhase.RUNTIME + return ErrorPhase.API + + def length_value_to_ticks(length: int, dots: int, ticks_per_quarter: int = 192) -> int: base = ticks_per_quarter * 4 // length total = base @@ -90,3 +112,12 @@ def note_to_midi(note: str, octave: int, accidental: int = 0) -> int: def clamp(value: int, min_val: int, max_val: int) -> int: return max(min_val, min(value, max_val)) + + +def context_line(source: str, node: ASTNode | Token) -> str: + """Return the source line for a node or token.""" + lines = source.splitlines() + line = getattr(node, "line", 1) + if 1 <= line <= len(lines): + return lines[line - 1] + return "" diff --git a/src/mml_composemusic_mcp/parser_ppmck.py b/src/mml_composemusic_mcp/parser_ppmck.py index 67e5093..9540844 100644 --- a/src/mml_composemusic_mcp/parser_ppmck.py +++ b/src/mml_composemusic_mcp/parser_ppmck.py @@ -1,22 +1,25 @@ -"""PPMCK-style MML parser.""" - -from .ir import ( - DutyEvent, - ErrorCode, - ErrorDetail, - NoteEvent, - NoteSequence, - RepeatEvent, - RestEvent, - TempoEvent, +"""PPMCK-style MML parser: tokens -> AST.""" + +from .ast_nodes import ( + ASTNode, + BarStmt, + DutyStmt, + Header, + LengthStmt, + NoteStmt, + OctaveStmt, + Program, + RepeatEndStmt, + RepeatStartStmt, + RestStmt, + TempoStmt, + TieStmt, + Track, + VolumeStmt, ) +from .ir import ErrorCode, ErrorDetail from .lexer import Token, TokenType -from .parser_base import ( - ParserContext, - clamp, - length_value_to_ticks, - note_to_midi, -) +from .parser_base import ParserContext, context_line CHANNEL_MAP = { "A": "Pulse1", @@ -24,111 +27,94 @@ CHANNEL_MAP = { "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() - self._tie_pending: bool = False - 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[Program, list[ErrorDetail]]: + program = Program(line=1, column=1, tracks=[]) + headers: list[Header] = [] - 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() + headers.append(self._parse_header()) continue if token.type == TokenType.COMMENT: self.ctx.advance() continue if token.type == TokenType.TRACK_HEADER: - self._parse_track_header() + track = self._parse_track() + if track is not None: + track.headers = list(headers) + program.tracks.append(track) continue - if self.current_channel is None: - self._warn_outside_track(token) - self.ctx.advance() + if token.type == TokenType.INVALID: + self._handle_invalid(token) 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, - ) - ) + # Anything else before a track header is outside a track. + self._warn_outside_track(token) + self.ctx.advance() - 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, - ) + # Attach headers to the first track if no track exists yet + if headers and not program.tracks: + pass # headers are simply discarded if no tracks + elif headers and program.tracks: + # Headers already attached to first parsed track; nothing to do + pass - return self.note_sequence, self.ctx.errors + return program, self.ctx.errors - def _parse_header(self) -> None: + def _parse_header(self) -> Header: 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 _warn_outside_track(self, token: Token) -> None: - if token.type in (TokenType.EOF, TokenType.COMMENT): - return - self.ctx.add_error( - code=ErrorCode.SYNTAX_OUTSIDE_TRACK, + unterminated = False + if len(parts) > 1: + rest = parts[1] + if rest.startswith('"'): + if not rest.endswith('"') or rest == '"': + unterminated = True + value = rest[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=context_line(self.source, token), + ) + else: + value = rest[1:-1] + else: + value = rest + return Header( line=token.line, column=token.column, - message=f"'{token.raw}' はトラック外に書かれています。", - severity="warning", - hint="コマンドはトラックヘッダー（A, B, T, N）の後に記述してください。", - context=self._context_line(token), + key=key, + value=value, + unterminated=unterminated, ) - def _parse_track_header(self) -> None: + def _parse_track(self) -> Track | None: token = self.ctx.advance() ch = token.value.upper() if ch == "L": - self.loop_start = 0 - return + return Track( + line=token.line, + column=token.column, + track_id="L", + channel="Loop", + mode="ppmck", + statements=[], + ) if ch not in CHANNEL_MAP: self.ctx.add_error( code=ErrorCode.SYNTAX_INVALID_TRACK_HEADER, @@ -137,12 +123,11 @@ class PpmckParser: message=f"無効なトラックヘッダー '{ch}' です。", severity="error", hint="A, B, T, N, L のいずれかを使用してください。", - context=self._context_line(token), + context=context_line(self.source, token), ) - self.current_channel = None - return - self.current_channel = CHANNEL_MAP[ch] - if self.current_channel in self.tracks_seen: + return None + channel = CHANNEL_MAP[ch] + if channel in self.tracks_seen: self.ctx.add_error( code=ErrorCode.SYNTAX_DUPLICATE_TRACK, line=token.line, @@ -150,64 +135,87 @@ class PpmckParser: message=f"トラック '{ch}' が複数回定義されています。", severity="error", hint="各トラックは1回のみ定義できます。重複定義を削除してください。", - context=self._context_line(token), + context=context_line(self.source, 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: + self.tracks_seen.add(channel) + track = Track( + line=token.line, + column=token.column, + track_id=ch, + channel=channel, + mode="ppmck", + statements=[], + ) + while self.ctx.peek().type not in ( + TokenType.TRACK_HEADER, + TokenType.EOF, + ): + stmt = self._parse_statement() + if stmt is not None: + track.statements.append(stmt) + return track + + def _parse_statement(self) -> ASTNode | 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), - ) + if token.type == TokenType.INVALID: + self._handle_invalid(token) + return None + if token.type == TokenType.COMMENT: self.ctx.advance() + return None + if token.type == TokenType.NOTE: + return self._parse_note() + if token.type == TokenType.REST: + return self._parse_rest() + if token.type == TokenType.OCTAVE: + return self._parse_octave() + if token.type == TokenType.OCTAVE_UP: + return self._parse_octave_up() + if token.type == TokenType.OCTAVE_DOWN: + return self._parse_octave_down() + if token.type == TokenType.LENGTH: + return self._parse_length() + if token.type == TokenType.VOLUME: + return self._parse_volume() + if token.type == TokenType.DUTY: + return self._parse_duty() + if token.type == TokenType.TEMPO: + return self._parse_tempo() + if token.type == TokenType.TIE: + return self._parse_tie() + if token.type == TokenType.REPEAT_START: + return self._parse_repeat_start() + if token.type == TokenType.REPEAT_END: + return self._parse_repeat_end() + if token.type == TokenType.BAR: + self.ctx.advance() + return BarStmt(line=token.line, column=token.column) + # Unknown token in track context + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"'{token.raw}' はここでは使用できません。", + severity="error", + hint="音符、休符、コマンド、小節線を配置してください。", + context=context_line(self.source, token), + ) + self.ctx.advance() + return None - def _read_length(self) -> tuple[int, int]: - dots = 0 + def _read_length(self, token: Token) -> tuple[int | None, int]: length_token = self.ctx.match(TokenType.NUMBER) - if length_token is None: - length = self.default_length - else: + length: int | None = None + if length_token is not None: length = int(length_token.value) + dots = 0 while self.ctx.match(TokenType.DOT): dots += 1 return length, dots - def _parse_note(self) -> None: + def _parse_note(self) -> NoteStmt: token = self.ctx.advance() - note = token.value.lower() - length, dots = self._read_length() - ticks = length_value_to_ticks(length, dots) + length, dots = self._read_length(token) accidental = 0 while True: if self.ctx.match(TokenType.SHARP): @@ -216,263 +224,157 @@ class PpmckParser: accidental -= 1 else: break + return NoteStmt( + line=token.line, + column=token.column, + note_name=token.value.lower(), + accidental=accidental, + length=length, + dots=dots, + ) - if self._tie_pending: - # Tie: extend the last note/rest's duration instead of creating a new event - self._extend_last_note(ticks) - self.tick_position += ticks - self._tie_pending = False - return - - 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) + def _parse_rest(self) -> RestStmt: + token = self.ctx.advance() + length, dots = self._read_length(token) + return RestStmt( + line=token.line, + column=token.column, + length=length, + dots=dots, + ) - event = NoteEvent( - tick_position=self.tick_position, - duration=ticks, - note_number=midi, - velocity=self.velocity, - duty=self.duty, - gate_time=1.0, + def _parse_octave(self) -> OctaveStmt: + token = self.ctx.advance() + value = int(token.value) if token.value else 4 + return OctaveStmt( + line=token.line, + column=token.column, + value=value, + direction=None, ) - 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) - if self._tie_pending: - self._extend_last_note(ticks) - self.tick_position += ticks - self._tie_pending = False - return - event = RestEvent(tick_position=self.tick_position, duration=ticks) - self._add_event(event) - self.tick_position += ticks + def _parse_octave_up(self) -> OctaveStmt: + token = self.ctx.advance() + return OctaveStmt( + line=token.line, + column=token.column, + value=None, + direction="up", + ) - def _parse_octave(self) -> None: + def _parse_octave_down(self) -> OctaveStmt: 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 + return OctaveStmt( + line=token.line, + column=token.column, + value=None, + direction="down", + ) - def _parse_octave_up(self) -> None: - self.ctx.advance() - if self.octave < 7: - self.octave += 1 + def _parse_length(self) -> LengthStmt: + token = self.ctx.advance() + value = int(token.value) if token.value else 4 + return LengthStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_octave_down(self) -> None: - self.ctx.advance() - if self.octave > 0: - self.octave -= 1 + def _parse_volume(self) -> VolumeStmt: + token = self.ctx.advance() + value = int(token.value) if token.value else 0 + return VolumeStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_length(self) -> None: + def _parse_duty(self) -> DutyStmt: 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 + value = int(token.value) if token.value else 0 + return DutyStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_volume(self) -> None: + def _parse_tempo(self) -> TempoStmt: 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 + value = int(token.value) if token.value else 120 + return TempoStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_duty(self) -> None: + def _parse_tie(self) -> TieStmt: token = self.ctx.advance() - if self.current_channel not in PULSE_CHANNELS: - self.ctx.add_error( - code=ErrorCode.SYNTAX_CHANNEL_MISMATCH, + next_token = self.ctx.peek() + if next_token.type == TokenType.NUMBER: + self.ctx.advance() + return TieStmt( line=token.line, column=token.column, - message=f"チャンネル '{self.current_channel}' で 'q' は使用できません。", - severity="warning", - hint="Triangleチャンネルはデューティ比を持ちません。q コマンドを削除してください。", - context=self._context_line(token), + target=int(next_token.value), ) - return - value = int(token.value) - if not 0 <= value <= 3: - self.ctx.add_error( - code=ErrorCode.SYNTAX_VALUE_OUT_OF_RANGE, + if next_token.type == TokenType.NOTE: + return TieStmt( line=token.line, column=token.column, - message=f"'q' の値 {value} は範囲外です。有効範囲: 0〜3。", - severity="error", - hint="例: q2 のように指定してください。", - context=self._context_line(token), + target=self._parse_note(), ) - 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, + if next_token.type == TokenType.REST: + return TieStmt( line=token.line, column=token.column, - message=f"'t' の値 {value} は範囲外です。有効範囲: 1以上。", - severity="error", - hint="例: t150 のように指定してください。", - context=self._context_line(token), + target=self._parse_rest(), ) - return - self.note_sequence.bpm = value - self._add_event(TempoEvent(tick_position=self.tick_position, bpm=value)) + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_TIE, + line=token.line, + column=token.column, + message="タイ '&' の後に音符が見つかりません。", + severity="error", + hint="& の後に音符（例: c4）を続けてください。音長のみ（例: &16）も可能です。", + context=context_line(self.source, token), + ) + return TieStmt(line=token.line, column=token.column, target=None) - def _parse_tie(self) -> None: + def _parse_repeat_start(self) -> RepeatStartStmt: 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: back up tick_position so the next event starts at the - # end of the last note/rest. The next note's duration will be added - # to the previous note's duration via _tie_pending flag. - self.tick_position = self._last_note_end() - self._tie_pending = True + return RepeatStartStmt(line=token.line, column=token.column) - 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 + ev.duration - 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 + def _parse_repeat_end(self) -> RepeatEndStmt: token = self.ctx.advance() - self.ctx.add_error( - code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + count: int | None = None + if token.value: + count = int(token.value) + return RepeatEndStmt( line=token.line, column=token.column, - message="区間リピート '[' は第2段階で実装されます。", - severity="warning", - hint="現時点では無視されます。", - context=self._context_line(token), + count=count, ) - def _parse_repeat_end(self) -> None: - token = self.ctx.advance() + def _handle_invalid(self, token: Token) -> None: + self.ctx.add_invalid_token_error(token, context_line(self.source, token)) + self.ctx.advance() + + def _warn_outside_track(self, token: Token) -> None: + if token.type in (TokenType.EOF, TokenType.COMMENT): + return self.ctx.add_error( - code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + code=ErrorCode.SEMANTIC_OUTSIDE_TRACK, line=token.line, column=token.column, - message="区間リピート ']' は第2段階で実装されます。", + message=f"'{token.raw}' はトラック外に書かれています。", severity="warning", - hint="現時点では無視されます。", - context=self._context_line(token), + hint="コマンドはトラックヘッダー（A, B, T, N）の後に記述してください。", + context=context_line(self.source, 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]]: +) -> tuple[Program, 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 index f75c169..db9787d 100644 --- a/src/mml_composemusic_mcp/parser_pyxel.py +++ b/src/mml_composemusic_mcp/parser_pyxel.py @@ -1,26 +1,28 @@ -"""Pyxel MML parser.""" - -from .ir import ( - DutyEvent, - EnvelopeEvent, - ErrorCode, - ErrorDetail, - GlideEvent, - NoteEvent, - NoteSequence, - RepeatEvent, - RestEvent, - TempoEvent, - VibratoEvent, - VolumeEvent, +"""Pyxel MML parser: tokens -> AST.""" + +from .ast_nodes import ( + ASTNode, + BarStmt, + DetuneStmt, + DutyStmt, + ExtCmdStmt, + GateTimeStmt, + LengthStmt, + NoteStmt, + OctaveStmt, + Program, + RepeatEndStmt, + RepeatStartStmt, + RestStmt, + TempoStmt, + TieStmt, + Track, + TransposeStmt, + VolumeStmt, ) +from .ir import ErrorCode, ErrorDetail from .lexer import Token, TokenType -from .parser_base import ( - ParserContext, - clamp, - length_value_to_ticks, - note_to_midi, -) +from .parser_base import ParserContext, context_line CHANNEL_MAP = { "0": "Pulse1", @@ -28,95 +30,33 @@ CHANNEL_MAP = { "2": "Triangle", "3": "Noise", } -PULSE_CHANNELS = {"Pulse1", "Pulse2"} - -PYXEL_DUTY = { - "1": 2, # Square 50% - "2": 1, # Pulse 25% -} - - -class RepeatFrame: - """Stack frame for nested repeat expansion.""" - - def __init__(self, start_tick: int) -> None: - self.start_tick = start_tick - # Snapshot of event indices that existed before the repeat section - self.event_count: int = 0 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[RepeatFrame] = [] - self._tie_pending: bool = False - 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[Program, list[ErrorDetail]]: + program = Program(line=1, column=1, tracks=[]) - 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() + track = self._parse_track() + if track is not None: + program.tracks.append(track) continue - if self.current_channel is None: - self._warn_outside_track(token) - self.ctx.advance() + if token.type == TokenType.INVALID: + self._handle_invalid(token) continue - self._parse_statement() - - if self.repeat_stack: - for _frame 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, - ) + self._warn_outside_track(token) + self.ctx.advance() - return self.note_sequence, self.ctx.errors + return program, self.ctx.errors - def _warn_outside_track(self, token: Token) -> None: - if token.type == TokenType.EOF: - return - self.ctx.add_error( - code=ErrorCode.SYNTAX_OUTSIDE_TRACK, - line=token.line, - column=token.column, - message=f"'{token.raw}' はトラック外に書かれています。", - severity="warning", - hint="コマンドはトラックヘッダー（0:, 1:, 2:, 3:）の後に記述してください。", - context=self._context_line(token), - ) - - def _parse_track_header(self) -> None: + def _parse_track(self) -> Track | None: token = self.ctx.advance() ch = token.value if ch not in CHANNEL_MAP: @@ -127,12 +67,11 @@ class PyxelParser: message=f"無効なトラックヘッダー '{ch}:' です。", severity="error", hint="0:, 1:, 2:, 3: のいずれかを使用してください。", - context=self._context_line(token), + context=context_line(self.source, token), ) - self.current_channel = None - return - self.current_channel = CHANNEL_MAP[ch] - if self.current_channel in self.tracks_seen: + return None + channel = CHANNEL_MAP[ch] + if channel in self.tracks_seen: self.ctx.add_error( code=ErrorCode.SYNTAX_DUPLICATE_TRACK, line=token.line, @@ -140,70 +79,91 @@ class PyxelParser: message=f"トラック '{ch}:' が複数回定義されています。", severity="error", hint="各トラックは1回のみ定義できます。重複定義を削除してください。", - context=self._context_line(token), + context=context_line(self.source, 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: + self.tracks_seen.add(channel) + track = Track( + line=token.line, + column=token.column, + track_id=ch, + channel=channel, + mode="pyxel", + statements=[], + ) + while self.ctx.peek().type not in ( + TokenType.TRACK_HEADER, + TokenType.EOF, + ): + stmt = self._parse_statement() + if stmt is not None: + track.statements.append(stmt) + return track + + def _parse_statement(self) -> ASTNode | 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.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), - ) + if token.type == TokenType.INVALID: + self._handle_invalid(token) + return None + if token.type == TokenType.NOTE: + return self._parse_note() + if token.type == TokenType.REST: + return self._parse_rest() + if token.type == TokenType.OCTAVE: + return self._parse_octave() + if token.type == TokenType.OCTAVE_UP: + return self._parse_octave_up() + if token.type == TokenType.OCTAVE_DOWN: + return self._parse_octave_down() + if token.type == TokenType.LENGTH: + return self._parse_length() + if token.type == TokenType.VOLUME: + return self._parse_volume() + if token.type == TokenType.GATE_TIME: + return self._parse_gate_time() + if token.type == TokenType.AT: + return self._parse_at() + if token.type == TokenType.TEMPO: + return self._parse_tempo() + if token.type == TokenType.TRANSPOSE: + return self._parse_transpose() + if token.type == TokenType.DETUNE: + return self._parse_detune() + if token.type == TokenType.TIE: + return self._parse_tie() + if token.type == TokenType.REPEAT_START: + return self._parse_repeat_start() + if token.type == TokenType.REPEAT_END: + return self._parse_repeat_end() + if token.type == TokenType.EXT_CMD: + return self._parse_ext_cmd() + if token.type == TokenType.BAR: self.ctx.advance() + return BarStmt(line=token.line, column=token.column) + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + line=token.line, + column=token.column, + message=f"'{token.raw}' はここでは使用できません。", + severity="error", + hint="音符、休符、コマンド、小節線を配置してください。", + context=context_line(self.source, token), + ) + self.ctx.advance() + return None - def _read_length(self) -> tuple[int, int]: - dots = 0 + def _read_length(self, token: Token) -> tuple[int | None, int]: length_token = self.ctx.match(TokenType.NUMBER) - if length_token is None: - length = self.default_length - else: + length: int | None = None + if length_token is not None: length = int(length_token.value) + dots = 0 while self.ctx.match(TokenType.DOT): dots += 1 return length, dots - def _parse_note(self) -> None: + def _parse_note(self) -> NoteStmt: token = self.ctx.advance() - note = token.value.lower() - length, dots = self._read_length() - ticks = length_value_to_ticks(length, dots) + length, dots = self._read_length(token) accidental = 0 while True: if self.ctx.match(TokenType.SHARP): @@ -212,454 +172,205 @@ class PyxelParser: accidental -= 1 else: break - - if self._tie_pending: - # Tie: extend the last note/rest's duration instead of creating a new event - self._extend_last_note(ticks) - self.tick_position += ticks - self._tie_pending = False - return - - 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, - detune_cents=float(self.detune_cents), + return NoteStmt( + line=token.line, + column=token.column, + note_name=token.value.lower(), + accidental=accidental, + length=length, + dots=dots, ) - 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) - if self._tie_pending: - self._extend_last_note(ticks) - self.tick_position += ticks - self._tie_pending = False - return - event = RestEvent(tick_position=self.tick_position, duration=ticks) - self._add_event(event) - self.tick_position += ticks - - def _parse_octave(self) -> None: + def _parse_rest(self) -> RestStmt: 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 + length, dots = self._read_length(token) + return RestStmt( + line=token.line, + column=token.column, + length=length, + dots=dots, + ) - def _parse_octave_up(self) -> None: - self.ctx.advance() - if self.octave < 7: - self.octave += 1 + def _parse_octave(self) -> OctaveStmt: + token = self.ctx.advance() + value = int(token.value) if token.value else 4 + return OctaveStmt( + line=token.line, + column=token.column, + value=value, + direction=None, + ) - def _parse_octave_down(self) -> None: - self.ctx.advance() - if self.octave > 0: - self.octave -= 1 + def _parse_octave_up(self) -> OctaveStmt: + token = self.ctx.advance() + return OctaveStmt( + line=token.line, + column=token.column, + value=None, + direction="up", + ) - def _parse_length(self) -> None: + def _parse_octave_down(self) -> OctaveStmt: 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 + return OctaveStmt( + line=token.line, + column=token.column, + value=None, + direction="down", + ) - def _parse_volume(self) -> None: + def _parse_length(self) -> LengthStmt: 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), - ) + value = int(token.value) if token.value else 4 + return LengthStmt( + line=token.line, + column=token.column, + value=value, ) - def _normalize_velocity(self, value: int) -> int: - return round(value / 127 * 15) + def _parse_volume(self) -> VolumeStmt: + token = self.ctx.advance() + value = int(token.value) if token.value else 0 + return VolumeStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_gate_time(self) -> None: + def _parse_gate_time(self) -> GateTimeStmt: 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 + value = int(token.value) if token.value else 80 + return GateTimeStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_at(self) -> None: + def _parse_at(self) -> DutyStmt: 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), - ) + value = int(token.value) if token.value else 0 + return DutyStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_tempo(self) -> None: + def _parse_tempo(self) -> TempoStmt: 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)) + value = int(token.value) if token.value else 120 + return TempoStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_transpose(self) -> None: + def _parse_transpose(self) -> TransposeStmt: token = self.ctx.advance() - self.transpose = int(token.value) + value = int(token.value) if token.value else 0 + return TransposeStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_detune(self) -> None: + def _parse_detune(self) -> DetuneStmt: token = self.ctx.advance() - self.detune_cents = int(token.value) + value = int(token.value) if token.value else 0 + return DetuneStmt( + line=token.line, + column=token.column, + value=value, + ) - def _parse_tie(self) -> None: + def _parse_tie(self) -> TieStmt: 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 - # For note/rest: back up tick_position so the next event starts at the - # end of the last note/rest. The next note's duration will be added - # to the previous note's duration via _tie_pending flag. - self.tick_position = self._last_note_end() - self._tie_pending = True - - 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 + ev.duration - 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() - ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] - frame = RepeatFrame(start_tick=self.tick_position) - frame.event_count = len(ch.events) - self.repeat_stack.append(frame) - - 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, + return TieStmt( line=token.line, column=token.column, - message="']' に対応する '[' が見つかりません。", - severity="error", - hint="直前の [ を確認するか、余分な ] を削除してください。", - context=self._context_line(token), + target=int(next_token.value), ) - return - frame = self.repeat_stack.pop() - start_tick = frame.start_tick - ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] - - count_str = token.value - if count_str == "": - count = 0 # infinite - self.ctx.add_error( - code=ErrorCode.SYNTAX_UNEXPECTED_TOKEN, + if next_token.type == TokenType.NOTE: + return TieStmt( line=token.line, column=token.column, - message="無限リピートはWAV生成時に2回で打ち切られます。", - severity="warning", - hint="回数を指定するか、2回の再生で十分であることを確認してください。", - context=self._context_line(token), + target=self._parse_note(), ) - else: - count = int(count_str) - - # Collect events that belong to this repeat section (created after the frame snapshot) - segment_events = ch.events[frame.event_count :] - segment_length = self.tick_position - start_tick - - repeat_count = 2 if count == 0 else count - for i in range(1, repeat_count): - offset = i * segment_length - for ev in segment_events: - 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 - - ch.events.sort(key=lambda e: e.tick_position) - self._add_event( - RepeatEvent( - start_tick=start_tick, - end_tick=self.tick_position, - repeat_count=count, + if next_token.type == TokenType.REST: + return TieStmt( + line=token.line, + column=token.column, + target=self._parse_rest(), ) + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_TIE, + line=token.line, + column=token.column, + message="タイ '&' の後に音符が見つかりません。", + severity="error", + hint="& の後に音符（例: C4）を続けてください。音長のみ（例: &16）も可能です。", + context=context_line(self.source, token), ) + return TieStmt(line=token.line, column=token.column, target=None) - def _clone_event(self, event, tick_position: int): # type: ignore[no-untyped-def] - import copy + def _parse_repeat_start(self) -> RepeatStartStmt: + token = self.ctx.advance() + return RepeatStartStmt(line=token.line, column=token.column) - new = copy.copy(event) - new.tick_position = tick_position - return new + def _parse_repeat_end(self) -> RepeatEndStmt: + token = self.ctx.advance() + count: int | None = None + if token.value: + count = int(token.value) + return RepeatEndStmt( + line=token.line, + column=token.column, + count=count, + ) - def _parse_ext_cmd(self) -> None: + def _parse_ext_cmd(self) -> ExtCmdStmt: token = self.ctx.advance() - cmd = token.value # ENV, VIB, GLI + cmd = token.value 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, + slot = 0 + if slot_token is not None: + slot = int(slot_token.value) + params: list[int] = [] + while True: + num = self.ctx.match(TokenType.NUMBER) + if num is None: + break + params.append(int(num.value)) + return ExtCmdStmt( line=token.line, column=token.column, - message=f"@{cmd} は第1段階では未サポート、無視されます。", - severity="warning", - hint="合成には反映されませんが、IRとして保持されます。", - context=self._context_line(token), + cmd=cmd, + slot=slot, + params=params, ) - def _parse_bar(self) -> None: + def _handle_invalid(self, token: Token) -> None: + self.ctx.add_invalid_token_error(token, context_line(self.source, token)) 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 _warn_outside_track(self, token: Token) -> None: + if token.type == TokenType.EOF: + return + self.ctx.add_error( + code=ErrorCode.SEMANTIC_OUTSIDE_TRACK, + line=token.line, + column=token.column, + message=f"'{token.raw}' はトラック外に書かれています。", + severity="warning", + hint="コマンドはトラックヘッダー（0:, 1:, 2:, 3:）の後に記述してください。", + context=context_line(self.source, token), + ) def parse_pyxel( source: str, tokens: list[Token] -) -> tuple[NoteSequence, list[ErrorDetail]]: +) -> tuple[Program, list[ErrorDetail]]: parser = PyxelParser(source, tokens) return parser.parse() diff --git a/src/mml_composemusic_mcp/semantic_base.py b/src/mml_composemusic_mcp/semantic_base.py new file mode 100644 index 0000000..0318a84 --- /dev/null +++ b/src/mml_composemusic_mcp/semantic_base.py @@ -0,0 +1,225 @@ +"""Base semantic analyzer: AST -> NoteSequence IR.""" + +from dataclasses import dataclass, field + +from .ast_nodes import ( + ASTNode, + BarStmt, + DetuneStmt, + DutyStmt, + EnvelopeDefStmt, + ExtCmdStmt, + GateTimeStmt, + LengthStmt, + NoteStmt, + OctaveStmt, + Program, + RepeatEndStmt, + RepeatStartStmt, + RestStmt, + Statement, + TempoStmt, + TieStmt, + Track, + TransposeStmt, + VolumeStmt, +) +from .ir import ( + ErrorCode, + ErrorDetail, + ErrorPhase, + NoteEvent, + NoteSequence, + RestEvent, +) +from .parser_base import length_value_to_ticks + + +@dataclass +class SemanticContext: + errors: list[ErrorDetail] = field(default_factory=list) + + 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, + phase=ErrorPhase.SEMANTIC, + line=line, + column=column, + message=message, + severity=severity, + hint=hint, + context=context, + ) + ) + + +class SemanticAnalyzer: + """Base class for converting AST to NoteSequence IR.""" + + def __init__(self, source: str, program: Program) -> None: + self.source = source + self.program = program + self.ctx = SemanticContext() + self.note_sequence = NoteSequence() + self.current_channel: str | None = None + self.octave = 4 + self.default_length = 4 + self.velocity = 15 + self.gate_time = 1.0 + self.duty = 2 + self.tick_position = 0 + + def analyze(self) -> tuple[NoteSequence, list[ErrorDetail]]: + for track in self.program.tracks: + self._analyze_track(track) + self._finalize() + return self.note_sequence, self.ctx.errors + + def _analyze_track(self, track: Track) -> None: + self.current_channel = track.channel + self._reset_state(track) + if track.channel == "Loop": + self._handle_loop_track(track) + return + for stmt in track.statements: + self._analyze_statement(stmt) + + def _reset_state(self, track: Track) -> None: + self.octave = 4 + self.default_length = 4 + self.velocity = 15 + self.gate_time = 1.0 + self.duty = 2 + self.tick_position = 0 + + def _handle_loop_track(self, track: Track) -> None: + # Loop track is handled after all tracks are processed. + pass + + def _finalize(self) -> None: + 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, + ) + + def _analyze_statement(self, stmt: Statement) -> None: + if isinstance(stmt, NoteStmt): + self._analyze_note(stmt) + elif isinstance(stmt, RestStmt): + self._analyze_rest(stmt) + elif isinstance(stmt, OctaveStmt): + self._analyze_octave(stmt) + elif isinstance(stmt, LengthStmt): + self._analyze_length(stmt) + elif isinstance(stmt, VolumeStmt): + self._analyze_volume(stmt) + elif isinstance(stmt, DutyStmt): + self._analyze_duty(stmt) + elif isinstance(stmt, TempoStmt): + self._analyze_tempo(stmt) + elif isinstance(stmt, GateTimeStmt): + self._analyze_gate_time(stmt) + elif isinstance(stmt, TransposeStmt): + self._analyze_transpose(stmt) + elif isinstance(stmt, DetuneStmt): + self._analyze_detune(stmt) + elif isinstance(stmt, TieStmt): + self._analyze_tie(stmt) + elif isinstance(stmt, RepeatStartStmt): + self._analyze_repeat_start(stmt) + elif isinstance(stmt, RepeatEndStmt): + self._analyze_repeat_end(stmt) + elif isinstance(stmt, BarStmt): + pass + elif isinstance(stmt, ExtCmdStmt): + self._analyze_ext_cmd(stmt) + elif isinstance(stmt, EnvelopeDefStmt): + self._analyze_stage2(stmt) + else: + self._analyze_stage2(stmt) + + def _ticks_for(self, stmt: NoteStmt | RestStmt) -> int: + length = stmt.length if stmt.length is not None else self.default_length + return length_value_to_ticks(length, stmt.dots) + + def _analyze_note(self, stmt: NoteStmt) -> None: + raise NotImplementedError + + def _analyze_rest(self, stmt: RestStmt) -> None: + raise NotImplementedError + + def _analyze_octave(self, stmt: OctaveStmt) -> None: + raise NotImplementedError + + def _analyze_length(self, stmt: LengthStmt) -> None: + raise NotImplementedError + + def _analyze_volume(self, stmt: VolumeStmt) -> None: + raise NotImplementedError + + def _analyze_duty(self, stmt: DutyStmt) -> None: + raise NotImplementedError + + def _analyze_tempo(self, stmt: TempoStmt) -> None: + raise NotImplementedError + + def _analyze_gate_time(self, stmt: GateTimeStmt) -> None: + raise NotImplementedError + + def _analyze_tie(self, stmt: TieStmt) -> None: + raise NotImplementedError + + def _analyze_repeat_start(self, stmt: RepeatStartStmt) -> None: + raise NotImplementedError + + def _analyze_repeat_end(self, stmt: RepeatEndStmt) -> None: + raise NotImplementedError + + def _analyze_ext_cmd(self, stmt: ExtCmdStmt) -> None: + raise NotImplementedError + + def _analyze_stage2(self, stmt: ASTNode) -> None: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE, + line=stmt.line, + column=stmt.column, + message=f"'{stmt.__class__.__name__}' は第2段階で実装されます。", + severity="warning", + hint="現時点では無視されます。", + context=self._context_line(stmt), + ) + + 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 _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 + ev.duration + 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 _context_line(self, node: ASTNode) -> str: + lines = self.source.splitlines() + if 1 <= node.line <= len(lines): + return lines[node.line - 1] + return "" diff --git a/src/mml_composemusic_mcp/semantic_ppmck.py b/src/mml_composemusic_mcp/semantic_ppmck.py new file mode 100644 index 0000000..b4393c0 --- /dev/null +++ b/src/mml_composemusic_mcp/semantic_ppmck.py @@ -0,0 +1,283 @@ +"""PPMCK semantic analyzer: AST -> NoteSequence IR.""" + + +from .ast_nodes import ( + DutyStmt, + EnvelopeDefStmt, + ExtCmdStmt, + GateTimeStmt, + LengthStmt, + NoteStmt, + OctaveStmt, + Program, + RepeatEndStmt, + RepeatStartStmt, + RestStmt, + SweepStmt, + TempoStmt, + TieStmt, + Track, + VolumeStmt, +) +from .ir import ( + DutyEvent, + ErrorCode, + ErrorDetail, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, + VolumeEvent, +) +from .parser_base import clamp, length_value_to_ticks, note_to_midi +from .semantic_base import SemanticAnalyzer + +PULSE_CHANNELS = {"Pulse1", "Pulse2"} + + +class PpmckSemanticAnalyzer(SemanticAnalyzer): + def __init__(self, source: str, program: Program) -> None: + super().__init__(source, program) + self.loop_start: int | None = None + + def analyze(self) -> tuple[NoteSequence, list[ErrorDetail]]: + ns, errors = super().analyze() + if self.loop_start is not None: + max_tick = max( + ch.total_ticks for ch in ns.channels.values() + ) + for ch in ns.channels.values(): + ch.events.append( + RepeatEvent( + start_tick=self.loop_start, + end_tick=max_tick, + repeat_count=0, + ) + ) + # Recompute total_ticks after adding loop events + for ch in ns.channels.values(): + ch.total_ticks = max( + (e.tick_position + getattr(e, "duration", 0) for e in ch.events), + default=0, + ) + return ns, errors + + def _reset_state(self, track: Track) -> None: + self.octave = 4 + self.default_length = 4 + self.velocity = 15 + self.gate_time = 1.0 + self.duty = 2 + self.tick_position = 0 + + def _analyze_track(self, track: Track) -> None: + if track.channel == "Loop": + self.loop_start = 0 + return + super()._analyze_track(track) + + def _analyze_note(self, stmt: NoteStmt) -> None: + ticks = self._ticks_for(stmt) + midi = note_to_midi(stmt.note_name, self.octave, stmt.accidental) + if self.current_channel == "Noise": + self.ctx.add_error( + code=ErrorCode.SEMANTIC_CHANNEL_MISMATCH, + line=stmt.line, + column=stmt.column, + message="チャンネル 'Noise' で音高指定は使用できません。", + severity="warning", + hint="Noiseチャンネルでは音高ではなく音長と音量のみ指定してください。", + context=self._context_line(stmt), + ) + midi = 0 + if not 0 <= midi <= 127: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_NOTE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"音高が範囲外です（MIDI音番号 {midi}）。", + severity="warning", + hint="オクターブを調整してください。有効範囲: o0〜o7。", + context=self._context_line(stmt), + ) + 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=self.gate_time, + ) + self._add_event(event) + self.tick_position += ticks + + def _analyze_rest(self, stmt: RestStmt) -> None: + ticks = self._ticks_for(stmt) + event = RestEvent(tick_position=self.tick_position, duration=ticks) + self._add_event(event) + self.tick_position += ticks + + def _analyze_octave(self, stmt: OctaveStmt) -> None: + if stmt.direction == "up": + if self.octave < 7: + self.octave += 1 + return + if stmt.direction == "down": + if self.octave > 0: + self.octave -= 1 + return + if stmt.value is None: + return + if not 0 <= stmt.value <= 7: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'o' の値 {stmt.value} は範囲外です。有効範囲: 0〜7。", + severity="error", + hint="例: o4 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.octave = stmt.value + + def _analyze_length(self, stmt: LengthStmt) -> None: + if not 1 <= stmt.value <= 192: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'l' の値 {stmt.value} は範囲外です。有効範囲: 1〜192。", + severity="error", + hint="例: l8 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.default_length = stmt.value + + def _analyze_volume(self, stmt: VolumeStmt) -> None: + if self.current_channel == "Triangle": + # Silently ignored per spec + return + if not 0 <= stmt.value <= 15: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'v' の値 {stmt.value} は範囲外です。有効範囲: 0〜15。", + severity="error", + hint="例: v15 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.velocity = stmt.value + self._add_event(VolumeEvent(tick_position=self.tick_position, value=stmt.value)) + + def _analyze_duty(self, stmt: DutyStmt) -> None: + if self.current_channel not in PULSE_CHANNELS: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_CHANNEL_MISMATCH, + line=stmt.line, + column=stmt.column, + message=f"チャンネル '{self.current_channel}' で 'q' は使用できません。", + severity="warning", + hint="Triangleチャンネルはデューティ比を持ちません。q コマンドを削除してください。", + context=self._context_line(stmt), + ) + return + if not 0 <= stmt.value <= 3: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'q' の値 {stmt.value} は範囲外です。有効範囲: 0〜3。", + severity="error", + hint="例: q2 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.duty = stmt.value + self._add_event(DutyEvent(tick_position=self.tick_position, value=stmt.value)) + + def _analyze_tempo(self, stmt: TempoStmt) -> None: + if stmt.value < 1: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'t' の値 {stmt.value} は範囲外です。有効範囲: 1以上。", + severity="error", + hint="例: t150 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.note_sequence.bpm = stmt.value + self._add_event(TempoEvent(tick_position=self.tick_position, bpm=stmt.value)) + + def _analyze_gate_time(self, stmt: GateTimeStmt) -> None: + # ppmck does not support gate time + self._analyze_stage2(stmt) + + def _analyze_tie(self, stmt: TieStmt) -> None: + if stmt.target is None: + return + if isinstance(stmt.target, int): + ticks = length_value_to_ticks(stmt.target, 0) + self._extend_last_note(ticks) + self.tick_position += ticks + return + if isinstance(stmt.target, NoteStmt): + target_ticks = self._ticks_for(stmt.target) + self._extend_last_note(target_ticks) + self.tick_position += target_ticks + return + if isinstance(stmt.target, RestStmt): + target_ticks = self._ticks_for(stmt.target) + self._extend_last_note(target_ticks) + self.tick_position += target_ticks + return + + def _analyze_repeat_start(self, stmt: RepeatStartStmt) -> None: + # Stage 2 feature + self._analyze_stage2(stmt) + + def _analyze_repeat_end(self, stmt: RepeatEndStmt) -> None: + # Stage 2 feature + self._analyze_stage2(stmt) + + def _analyze_ext_cmd(self, stmt: ExtCmdStmt) -> None: + # ppmck does not use pyxel ext commands + self._analyze_stage2(stmt) + + def _analyze_stage2(self, stmt) -> None: # type: ignore[no-untyped-def] + if isinstance(stmt, EnvelopeDefStmt): + self.ctx.add_error( + code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE, + line=stmt.line, + column=stmt.column, + message="エンベロープ定義 '@v' は第2段階で実装されます。", + severity="warning", + hint="現時点では無視されます。", + context=self._context_line(stmt), + ) + elif isinstance(stmt, SweepStmt): + self.ctx.add_error( + code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE, + line=stmt.line, + column=stmt.column, + message="スイープ 's' は第2段階で実装されます。", + severity="warning", + hint="現時点では無視されます。", + context=self._context_line(stmt), + ) + else: + super()._analyze_stage2(stmt) + + +def analyze_ppmck( + source: str, program: Program +) -> tuple[NoteSequence, list[ErrorDetail]]: + analyzer = PpmckSemanticAnalyzer(source, program) + return analyzer.analyze() diff --git a/src/mml_composemusic_mcp/semantic_pyxel.py b/src/mml_composemusic_mcp/semantic_pyxel.py new file mode 100644 index 0000000..32808d1 --- /dev/null +++ b/src/mml_composemusic_mcp/semantic_pyxel.py @@ -0,0 +1,423 @@ +"""Pyxel semantic analyzer: AST -> NoteSequence IR.""" + +import copy + +from .ast_nodes import ( + DutyStmt, + ExtCmdStmt, + GateTimeStmt, + LengthStmt, + NoteStmt, + OctaveStmt, + Program, + RepeatEndStmt, + RepeatStartStmt, + RestStmt, + TempoStmt, + TieStmt, + Track, + TransposeStmt, + VolumeStmt, +) +from .ir import ( + DutyEvent, + EnvelopeEvent, + ErrorCode, + ErrorDetail, + GlideEvent, + NoteEvent, + NoteSequence, + RepeatEvent, + RestEvent, + TempoEvent, + VibratoEvent, + VolumeEvent, +) +from .parser_base import clamp, length_value_to_ticks, note_to_midi +from .semantic_base import SemanticAnalyzer + +PULSE_CHANNELS = {"Pulse1", "Pulse2"} +PYXEL_DUTY = { + 1: 2, # Square 50% + 2: 1, # Pulse 25% +} + + +class RepeatFrame: + """Stack frame for nested repeat expansion.""" + + def __init__(self, start_tick: int) -> None: + self.start_tick = start_tick + self.event_count: int = 0 + + +class PyxelSemanticAnalyzer(SemanticAnalyzer): + def __init__(self, source: str, program: Program) -> None: + super().__init__(source, program) + self.transpose = 0 + self.detune_cents = 0.0 + self.repeat_stack: list[RepeatFrame] = [] + + def _reset_state(self, track: Track) -> 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.0 + self.tick_position = 0 + self.repeat_stack = [] + + def _analyze_track(self, track: Track) -> None: + self.current_channel = track.channel + self._reset_state(track) + for stmt in track.statements: + self._analyze_statement(stmt) + if self.repeat_stack: + for _frame in self.repeat_stack: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNTERMINATED_REPEAT, + line=1, + column=1, + message="リピート '[' に対応する ']' が見つかりません。", + severity="error", + hint="] を追加してリピートを閉じてください。回数指定（例: ]2）も可能です。", + context="", + ) + + def _analyze_note(self, stmt: NoteStmt) -> None: + ticks = self._ticks_for(stmt) + midi = note_to_midi(stmt.note_name, self.octave, stmt.accidental) + if self.current_channel != "Noise": + midi += self.transpose + if not 0 <= midi <= 127: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_NOTE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"音高が範囲外です（MIDI音番号 {midi}）。", + severity="warning", + hint="オクターブ/トランスポーズを調整してください。有効範囲: O0〜O7。", + context=self._context_line(stmt), + ) + 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, + detune_cents=self.detune_cents, + ) + self._add_event(event) + self.tick_position += ticks + + def _analyze_rest(self, stmt: RestStmt) -> None: + ticks = self._ticks_for(stmt) + event = RestEvent(tick_position=self.tick_position, duration=ticks) + self._add_event(event) + self.tick_position += ticks + + def _analyze_octave(self, stmt: OctaveStmt) -> None: + if stmt.direction == "up": + if self.octave < 7: + self.octave += 1 + return + if stmt.direction == "down": + if self.octave > 0: + self.octave -= 1 + return + if stmt.value is None: + return + if not 0 <= stmt.value <= 7: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'O' の値 {stmt.value} は範囲外です。有効範囲: 0〜7。", + severity="error", + hint="例: O4 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.octave = stmt.value + + def _analyze_length(self, stmt: LengthStmt) -> None: + if not 1 <= stmt.value <= 192: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'L' の値 {stmt.value} は範囲外です。有効範囲: 1〜192。", + severity="error", + hint="例: L8 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.default_length = stmt.value + + def _analyze_volume(self, stmt: VolumeStmt) -> None: + if not 0 <= stmt.value <= 127: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'V' の値 {stmt.value} は範囲外です。有効範囲: 0〜127。", + severity="error", + hint="例: V100 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.velocity = stmt.value + self._add_event( + VolumeEvent( + tick_position=self.tick_position, + value=self._normalize_velocity(stmt.value), + ) + ) + + def _normalize_velocity(self, value: int) -> int: + return round(value / 127 * 15) + + def _analyze_gate_time(self, stmt: GateTimeStmt) -> None: + if not 0 <= stmt.value <= 100: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'Q' の値 {stmt.value} は範囲外です。有効範囲: 0〜100。", + severity="error", + hint="例: Q80 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.gate_time = stmt.value / 100.0 + + def _analyze_duty(self, stmt: DutyStmt) -> None: + if stmt.value in PYXEL_DUTY: + if self.current_channel not in PULSE_CHANNELS: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_CHANNEL_MISMATCH, + line=stmt.line, + column=stmt.column, + message=f"チャンネル '{self.current_channel}' で '@{stmt.value}' は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、Pulseチャンネル（0: または 1:）に移動してください。", + context=self._context_line(stmt), + ) + return + self.duty = PYXEL_DUTY[stmt.value] + self._add_event( + DutyEvent(tick_position=self.tick_position, value=self.duty) + ) + return + if stmt.value == 0: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_CHANNEL_MISMATCH, + line=stmt.line, + column=stmt.column, + message=f"チャンネル '{self.current_channel}' で '@0' (Triangle) は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、適切なチャンネルに移動してください。", + context=self._context_line(stmt), + ) + elif stmt.value == 3: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_CHANNEL_MISMATCH, + line=stmt.line, + column=stmt.column, + message=f"チャンネル '{self.current_channel}' で '@3' (Noise) は使用できません。", + severity="warning", + hint="トラック番号がチャンネルを決定します。@ コマンドを削除するか、適切なチャンネルに移動してください。", + context=self._context_line(stmt), + ) + else: + self.ctx.add_error( + code=ErrorCode.SYNTAX_INVALID_NUMBER, + line=stmt.line, + column=stmt.column, + message=f"'@' の後に無効な値 '{stmt.value}' が見つかりました。", + severity="error", + hint="例: @1 のように指定してください。", + context=self._context_line(stmt), + ) + + def _analyze_tempo(self, stmt: TempoStmt) -> None: + if stmt.value < 1: + self.ctx.add_error( + code=ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE, + line=stmt.line, + column=stmt.column, + message=f"'T' の値 {stmt.value} は範囲外です。有効範囲: 1以上。", + severity="error", + hint="例: T150 のように指定してください。", + context=self._context_line(stmt), + ) + return + self.note_sequence.bpm = stmt.value + self._add_event(TempoEvent(tick_position=self.tick_position, bpm=stmt.value)) + + def _analyze_transpose(self, stmt: TransposeStmt) -> None: + self.transpose = stmt.value + + def _analyze_detune(self, stmt) -> None: # type: ignore[no-untyped-def] + self.detune_cents = float(stmt.value) + + def _analyze_tie(self, stmt: TieStmt) -> None: + if stmt.target is None: + return + if isinstance(stmt.target, int): + ticks = length_value_to_ticks(stmt.target, 0) + self._extend_last_note(ticks) + self.tick_position += ticks + return + if isinstance(stmt.target, NoteStmt): + target_ticks = self._ticks_for(stmt.target) + self._extend_last_note(target_ticks) + self.tick_position += target_ticks + return + if isinstance(stmt.target, RestStmt): + target_ticks = self._ticks_for(stmt.target) + self._extend_last_note(target_ticks) + self.tick_position += target_ticks + return + + def _analyze_repeat_start(self, stmt: RepeatStartStmt) -> None: + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + frame = RepeatFrame(start_tick=self.tick_position) + frame.event_count = len(ch.events) + self.repeat_stack.append(frame) + + def _analyze_repeat_end(self, stmt: RepeatEndStmt) -> None: + if not self.repeat_stack: + self.ctx.add_error( + code=ErrorCode.SYNTAX_UNMATCHED_REPEAT_END, + line=stmt.line, + column=stmt.column, + message="']' に対応する '[' が見つかりません。", + severity="error", + hint="直前の [ を確認するか、余分な ] を削除してください。", + context=self._context_line(stmt), + ) + return + frame = self.repeat_stack.pop() + start_tick = frame.start_tick + ch = self.note_sequence.channels[self.current_channel] # type: ignore[index] + + count = stmt.count + if count is None: + count = 0 # infinite + self.ctx.add_error( + code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE, + line=stmt.line, + column=stmt.column, + message="無限リピートはWAV生成時に2回で打ち切られます。", + severity="warning", + hint="回数を指定するか、2回の再生で十分であることを確認してください。", + context=self._context_line(stmt), + ) + + segment_events = ch.events[frame.event_count :] + segment_length = self.tick_position - start_tick + + repeat_count = 2 if count == 0 else count + for i in range(1, repeat_count): + offset = i * segment_length + for ev in segment_events: + 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 + + ch.events.sort(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] + new = copy.copy(event) + new.tick_position = tick_position + return new + + def _analyze_ext_cmd(self, stmt: ExtCmdStmt) -> None: + if stmt.slot == 0: + if stmt.cmd == "ENV": + ev = EnvelopeEvent(tick_position=self.tick_position, slot=0) + elif stmt.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 stmt.params: + if stmt.cmd == "ENV": + points = [] + for i in range(1, len(stmt.params), 2): + duration = stmt.params[i] + target = stmt.params[i - 1] if i - 1 < len(stmt.params) else stmt.params[-1] + points.append({"target_volume": target, "duration_ticks": duration}) + ev = EnvelopeEvent( + tick_position=self.tick_position, + slot=stmt.slot, + points=points, + ) + elif stmt.cmd == "VIB": + if len(stmt.params) >= 3: + params = { + "delay_ticks": stmt.params[0], + "period_ticks": stmt.params[1], + "depth_cents": stmt.params[2], + } + else: + params = {} + ev = VibratoEvent( + tick_position=self.tick_position, + slot=stmt.slot, + params=params, + ) + else: + if len(stmt.params) >= 2: + params = { + "initial_offset_cents": stmt.params[0], + "duration_ticks": stmt.params[1], + } + else: + params = {} + ev = GlideEvent( + tick_position=self.tick_position, + slot=stmt.slot, + params=params, + ) + self._add_event(ev) + else: + if stmt.cmd == "ENV": + ev = EnvelopeEvent(tick_position=self.tick_position, slot=stmt.slot) + elif stmt.cmd == "VIB": + ev = VibratoEvent(tick_position=self.tick_position, slot=stmt.slot) + else: + ev = GlideEvent(tick_position=self.tick_position, slot=stmt.slot) + self._add_event(ev) + self.ctx.add_error( + code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE, + line=stmt.line, + column=stmt.column, + message=f"@{stmt.cmd} は第1段階では未サポート、無視されます。", + severity="warning", + hint="合成には反映されませんが、IRとして保持されます。", + context=self._context_line(stmt), + ) + + +def analyze_pyxel( + source: str, program: Program +) -> tuple[NoteSequence, list[ErrorDetail]]: + analyzer = PyxelSemanticAnalyzer(source, program) + return analyzer.analyze() diff --git a/src/mml_composemusic_mcp/server.py b/src/mml_composemusic_mcp/server.py index 0a09131..de0ea82 100644 --- a/src/mml_composemusic_mcp/server.py +++ b/src/mml_composemusic_mcp/server.py @@ -6,10 +6,12 @@ from pathlib import Path from fastmcp import FastMCP -from .ir import ErrorCode, ErrorDetail, NoteSequence +from .ir import ErrorCode, ErrorDetail, ErrorPhase, NoteSequence from .lexer import tokenize from .parser_ppmck import parse_ppmck from .parser_pyxel import parse_pyxel +from .semantic_ppmck import analyze_ppmck +from .semantic_pyxel import analyze_pyxel from .synthesizer import build_channel_summary, synthesize, write_wav from .templates import get_template @@ -40,16 +42,25 @@ def _split_errors(errors: list[ErrorDetail]) -> tuple[list[dict], list[dict]]: return errs, warns -def _parse_mml(mml: str, mode: str) -> tuple[dict, list[ErrorDetail]]: +def _parse_mml(mml: str, mode: str) -> tuple[dict | None, list[ErrorDetail]]: + """Run lexer, parser, and semantic analyzer for the given mode.""" tokens = tokenize(mml, mode) + if mode == "ppmck": - note_sequence, errors = parse_ppmck(mml, tokens) + program, errors = parse_ppmck(mml, tokens) + if not _is_error(errors): + note_sequence, semantic_errors = analyze_ppmck(mml, program) + errors.extend(semantic_errors) elif mode == "pyxel": - note_sequence, errors = parse_pyxel(mml, tokens) + program, errors = parse_pyxel(mml, tokens) + if not _is_error(errors): + note_sequence, semantic_errors = analyze_pyxel(mml, program) + errors.extend(semantic_errors) else: errors = [ ErrorDetail( code=ErrorCode.VALIDATION_INVALID_MODE, + phase=ErrorPhase.API, line=0, column=0, message=f"未知のモード '{mode}' です。", @@ -58,6 +69,9 @@ def _parse_mml(mml: str, mode: str) -> tuple[dict, list[ErrorDetail]]: ) ] note_sequence = None + + if _is_error(errors): + note_sequence = None return (note_sequence.to_dict() if note_sequence else None), errors @@ -89,6 +103,7 @@ def compose_mml( "errors": [ { "code": ErrorCode.VALIDATION_MISSING_PARAMETER.value, + "phase": ErrorPhase.API.value, "line": 0, "column": 0, "message": "mml と mode は compose/validate の必須パラメータです。", @@ -136,7 +151,8 @@ def compose_mml( "validation": { "errors": [ { - "code": ErrorCode.SYSTEM_SYNTHESIS_FAILED.value, + "code": ErrorCode.RUNTIME_SYNTHESIS_FAILED.value, + "phase": ErrorPhase.RUNTIME.value, "line": 0, "column": 0, "message": f"音声合成中にエラーが発生しました: {exc}", @@ -177,7 +193,8 @@ def compose_mml( "validation": { "errors": [ { - "code": ErrorCode.SYSTEM_INTERNAL_ERROR.value, + "code": ErrorCode.RUNTIME_INTERNAL_ERROR.value, + "phase": ErrorPhase.RUNTIME.value, "line": 0, "column": 0, "message": f"内部エラーが発生しました: {exc}", @@ -195,6 +212,7 @@ def compose_mml( "errors": [ { "code": ErrorCode.VALIDATION_INVALID_ACTION.value, + "phase": ErrorPhase.API.value, "line": 0, "column": 0, "message": f"未知の action '{action}' です。", diff --git a/src/mml_composemusic_mcp/synthesizer.py b/src/mml_composemusic_mcp/synthesizer.py index 6d9ffe2..b314aa5 100644 --- a/src/mml_composemusic_mcp/synthesizer.py +++ b/src/mml_composemusic_mcp/synthesizer.py @@ -10,6 +10,7 @@ from .ir import ( DutyEvent, ErrorCode, ErrorDetail, + ErrorPhase, NoteEvent, NoteSequence, RestEvent, @@ -262,7 +263,8 @@ def write_wav(path: Path, wave_data: np.ndarray, sample_rate: int) -> list[Error except Exception as exc: errors.append( ErrorDetail( - code=ErrorCode.SYSTEM_WAV_WRITE_FAILED, + code=ErrorCode.RUNTIME_WAV_WRITE_FAILED, + phase=ErrorPhase.RUNTIME, line=0, column=0, message=f"WAVファイルの出力に失敗しました: {exc}", -- cgit v1.2.3