summaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/mml_composemusic_mcp/ast_nodes.py1
-rw-r--r--src/mml_composemusic_mcp/ir.py18
-rw-r--r--src/mml_composemusic_mcp/lexer.py2
-rw-r--r--src/mml_composemusic_mcp/parser_ppmck.py30
-rw-r--r--src/mml_composemusic_mcp/semantic_base.py3
-rw-r--r--src/mml_composemusic_mcp/semantic_ppmck.py347
-rw-r--r--src/mml_composemusic_mcp/semantic_pyxel.py69
-rw-r--r--src/mml_composemusic_mcp/server.py2
-rw-r--r--src/mml_composemusic_mcp/synthesizer.py693
9 files changed, 759 insertions, 406 deletions
diff --git a/src/mml_composemusic_mcp/ast_nodes.py b/src/mml_composemusic_mcp/ast_nodes.py
index 3341510..dc85e79 100644
--- a/src/mml_composemusic_mcp/ast_nodes.py
+++ b/src/mml_composemusic_mcp/ast_nodes.py
@@ -17,6 +17,7 @@ class Program(ASTNode):
"""Root program node."""
tracks: list[Track] = field(default_factory=list)
+ global_statements: list[Statement] = field(default_factory=list)
@dataclass
diff --git a/src/mml_composemusic_mcp/ir.py b/src/mml_composemusic_mcp/ir.py
index 4dd7978..d9a2f93 100644
--- a/src/mml_composemusic_mcp/ir.py
+++ b/src/mml_composemusic_mcp/ir.py
@@ -44,6 +44,7 @@ class ErrorCode(Enum):
SEMANTIC_EMPTY_TRACK = "SEMANTIC_EMPTY_TRACK"
SEMANTIC_OUTSIDE_TRACK = "SEMANTIC_OUTSIDE_TRACK"
SEMANTIC_UNDEFINED_REFERENCE = "SEMANTIC_UNDEFINED_REFERENCE"
+ SEMANTIC_DUPLICATE_DEFINITION = "SEMANTIC_DUPLICATE_DEFINITION"
SEMANTIC_UNSUPPORTED_FEATURE = "SEMANTIC_UNSUPPORTED_FEATURE"
# --- Runtime errors (Synthesizer phase) ---
@@ -279,9 +280,22 @@ class ChannelSequence:
@dataclass
class NoteSequence:
- version: str = "1.0"
+ version: str = "2.0"
bpm: int = 120
ticks_per_quarter: int = 192
+ mode: str = "ppmck"
+ definitions: dict[str, dict[int, Any]] = field(
+ default_factory=lambda: {
+ "volume_envelopes": {},
+ "duty_envelopes": {},
+ "pitch_envelopes": {},
+ "note_envelopes": {},
+ "lfos": {},
+ "envelopes": {},
+ "vibratos": {},
+ "glides": {},
+ }
+ )
channels: dict[str, ChannelSequence] = field(
default_factory=lambda: {
"Pulse1": ChannelSequence("pulse"),
@@ -296,6 +310,8 @@ class NoteSequence:
"version": self.version,
"bpm": self.bpm,
"ticks_per_quarter": self.ticks_per_quarter,
+ "mode": self.mode,
+ "definitions": self.definitions,
"channels": {k: v.to_dict() for k, v in self.channels.items()},
}
diff --git a/src/mml_composemusic_mcp/lexer.py b/src/mml_composemusic_mcp/lexer.py
index e5f08a8..ed0a26e 100644
--- a/src/mml_composemusic_mcp/lexer.py
+++ b/src/mml_composemusic_mcp/lexer.py
@@ -329,7 +329,7 @@ class Lexer:
num0 = self._read_number()
if self._peek() == ",":
self._advance()
- num1 = self._read_number()
+ num1 = self._read_signed_number()
raw = f"s{num0},{num1}"
self._emit(TokenType.SWEEP, f"{num0},{num1}", raw)
return
diff --git a/src/mml_composemusic_mcp/parser_ppmck.py b/src/mml_composemusic_mcp/parser_ppmck.py
index 47a0412..bd3b06b 100644
--- a/src/mml_composemusic_mcp/parser_ppmck.py
+++ b/src/mml_composemusic_mcp/parser_ppmck.py
@@ -97,6 +97,36 @@ class PpmckParser:
if token.type == TokenType.INVALID:
self._handle_invalid(token)
continue
+ if token.type in {
+ TokenType.VOL_ENV,
+ TokenType.DUTY,
+ TokenType.LFO_DEF,
+ TokenType.PITCH_ENV_DEF,
+ TokenType.NOTE_ENV_DEF,
+ }:
+ statement = self._parse_statement()
+ if isinstance(
+ statement,
+ (
+ VolumeEnvelopeDefStmt,
+ DutyEnvelopeDefStmt,
+ LfoDefStmt,
+ PitchEnvDefStmt,
+ NoteEnvDefStmt,
+ ),
+ ):
+ program.global_statements.append(statement)
+ else:
+ self.ctx.add_error(
+ code=ErrorCode.SEMANTIC_OUTSIDE_TRACK,
+ line=token.line,
+ column=token.column,
+ message="トラック外ではエフェクト定義だけを使用できます。",
+ severity="error",
+ hint="選択コマンドはトラック内へ移動してください。",
+ context=context_line(self.source, token),
+ )
+ continue
# Anything else before a track header is outside a track.
self._warn_outside_track(token)
self.ctx.advance()
diff --git a/src/mml_composemusic_mcp/semantic_base.py b/src/mml_composemusic_mcp/semantic_base.py
index 8b27e89..0a6acdb 100644
--- a/src/mml_composemusic_mcp/semantic_base.py
+++ b/src/mml_composemusic_mcp/semantic_base.py
@@ -97,6 +97,9 @@ class SemanticAnalyzer:
self.tick_position = 0
def analyze(self) -> tuple[NoteSequence, list[ErrorDetail]]:
+ self.current_channel = None
+ for stmt in self.program.global_statements:
+ self._analyze_statement(stmt)
for track in self.program.tracks:
self._analyze_track(track)
self._finalize()
diff --git a/src/mml_composemusic_mcp/semantic_ppmck.py b/src/mml_composemusic_mcp/semantic_ppmck.py
index e2cabf9..4171a4b 100644
--- a/src/mml_composemusic_mcp/semantic_ppmck.py
+++ b/src/mml_composemusic_mcp/semantic_ppmck.py
@@ -63,10 +63,50 @@ PULSE_CHANNELS = {"Pulse1", "Pulse2"}
class PpmckSemanticAnalyzer(SemanticAnalyzer):
def __init__(self, source: str, program: Program) -> None:
super().__init__(source, program)
+ self.note_sequence.mode = "ppmck"
self.loop_start: int | None = None
+ self.detune_cents = 0.0
+
+ def _define(self, kind: str, slot: int, value: dict) -> None:
+ table = self.note_sequence.definitions[kind]
+ if slot in table:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_DUPLICATE_DEFINITION,
+ 0,
+ 0,
+ f"エフェクトスロット {slot} は既に定義されています。",
+ "error",
+ "別のスロット番号を使用してください。",
+ )
+ return
+ table[slot] = value
def analyze(self) -> tuple[NoteSequence, list[ErrorDetail]]:
ns, errors = super().analyze()
+ references = (
+ (VolumeEnvelopeEvent, "volume_envelopes"),
+ (DutyEnvelopeEvent, "duty_envelopes"),
+ (LfoEvent, "lfos"),
+ (PitchEnvEvent, "pitch_envelopes"),
+ (NoteEnvEvent, "note_envelopes"),
+ )
+ for channel in ns.channels.values():
+ for event in channel.events:
+ for event_type, table_name in references:
+ if (
+ isinstance(event, event_type)
+ and not event.is_definition
+ and not getattr(event, "is_off", False)
+ and event.slot not in ns.definitions[table_name]
+ ):
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_UNDEFINED_REFERENCE,
+ 0,
+ 0,
+ f"未定義のエフェクトスロット {event.slot} です。",
+ "error",
+ "先に同じ番号の定義を追加してください。",
+ )
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():
@@ -92,6 +132,7 @@ class PpmckSemanticAnalyzer(SemanticAnalyzer):
self.gate_time = 1.0
self.duty = 2
self.tick_position = 0
+ self.detune_cents = 0.0
def _analyze_track(self, track: Track) -> None:
if track.channel == "Loop":
@@ -131,6 +172,7 @@ class PpmckSemanticAnalyzer(SemanticAnalyzer):
velocity=self.velocity,
duty=self.duty,
gate_time=self.gate_time,
+ detune_cents=self.detune_cents,
)
self._add_event(event)
self.tick_position += ticks
@@ -255,33 +297,37 @@ class PpmckSemanticAnalyzer(SemanticAnalyzer):
return
def _analyze_detune_cmd(self, stmt: DetuneCmdStmt) -> None:
+ self.detune_cents = float(stmt.value)
self._add_event(DetuneEvent(tick_position=self.tick_position, value=stmt.value))
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"ディチューン 'D{stmt.value}' はIR保持のみで合成に反映されません。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_relative_volume(self, stmt: RelativeVolumeStmt) -> None:
+ self.velocity = clamp(self.velocity + stmt.delta, 0, 15)
self._add_event(
RelativeVolumeEvent(tick_position=self.tick_position, delta=stmt.delta)
)
- sign = "+" if stmt.delta > 0 else "-"
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"相対音量 'v{sign}{abs(stmt.delta)}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_sweep(self, stmt: SweepStmt) -> None:
+ if self.current_channel not in PULSE_CHANNELS:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_CHANNEL_MISMATCH,
+ stmt.line,
+ stmt.column,
+ "スイープはPulseチャンネル専用です。",
+ "error",
+ "AまたはBトラックで使用してください。",
+ self._context_line(stmt),
+ )
+ return
+ if not 0 <= stmt.speed <= 7 or not 1 <= abs(stmt.depth) <= 7:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE,
+ stmt.line,
+ stmt.column,
+ "スイープ値はspeed=0〜7、depth=±1〜±7で指定してください。",
+ "error",
+ context=self._context_line(stmt),
+ )
+ return
self._add_event(
SweepEvent(
tick_position=self.tick_position,
@@ -289,221 +335,141 @@ class PpmckSemanticAnalyzer(SemanticAnalyzer):
depth=stmt.depth,
)
)
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"スイープ 's{stmt.speed},{stmt.depth}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_vol_env_use(self, stmt: VolumeEnvelopeUseStmt) -> None:
self._add_event(
VolumeEnvelopeEvent(tick_position=self.tick_position, slot=stmt.slot)
)
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"音量エンベロープ '@v{stmt.slot}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_vol_env_def(self, stmt: VolumeEnvelopeDefStmt) -> None:
- self._add_event(
- VolumeEnvelopeEvent(
- tick_position=self.tick_position,
- slot=stmt.slot,
- points=stmt.points,
- loop_points=stmt.loop_points,
- is_definition=True,
- )
- )
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"音量エンベロープ定義 '@v{stmt.slot}={{...}}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
+ self._define(
+ "volume_envelopes",
+ stmt.slot,
+ {"points": stmt.points, "loop_points": stmt.loop_points},
)
+ if self.current_channel is not None:
+ self._add_event(
+ VolumeEnvelopeEvent(
+ tick_position=self.tick_position,
+ slot=stmt.slot,
+ points=stmt.points,
+ loop_points=stmt.loop_points,
+ is_definition=True,
+ )
+ )
def _analyze_duty_env_def(self, stmt: DutyEnvelopeDefStmt) -> None:
- self._add_event(
- DutyEnvelopeEvent(
- tick_position=self.tick_position,
- slot=stmt.slot,
- points=stmt.points,
- loop_points=stmt.loop_points,
- is_definition=True,
- )
- )
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"デューティエンベロープ定義 '@{stmt.slot}={{...}}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
+ self._define(
+ "duty_envelopes",
+ stmt.slot,
+ {"points": stmt.points, "loop_points": stmt.loop_points},
)
+ if self.current_channel is not None:
+ self._add_event(
+ DutyEnvelopeEvent(
+ tick_position=self.tick_position,
+ slot=stmt.slot,
+ points=stmt.points,
+ loop_points=stmt.loop_points,
+ is_definition=True,
+ )
+ )
def _analyze_duty_env_use(self, stmt: DutyEnvelopeUseStmt) -> None:
- self._add_event(
- DutyEnvelopeEvent(tick_position=self.tick_position, slot=stmt.slot)
- )
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"デューティエンベロープ使用 '@@{stmt.slot}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
+ if self.current_channel is not None:
+ self._add_event(
+ DutyEnvelopeEvent(tick_position=self.tick_position, slot=stmt.slot)
+ )
def _analyze_lfo_def(self, stmt: LfoDefStmt) -> None:
- self._add_event(
- LfoEvent(
- tick_position=self.tick_position,
- slot=stmt.slot,
- delay=stmt.delay,
- speed=stmt.speed,
- depth=stmt.depth,
- transition=stmt.transition,
- is_definition=True,
+ if stmt.transition != 0 or stmt.speed <= 0:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE,
+ stmt.line,
+ stmt.column,
+ "@MPはspeed>0、transition=0で指定してください。",
+ "error",
+ context=self._context_line(stmt),
)
+ return
+ self._define(
+ "lfos",
+ stmt.slot,
+ {
+ "delay": stmt.delay,
+ "speed": stmt.speed,
+ "depth": stmt.depth,
+ "transition": stmt.transition,
+ },
)
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"LFO定義 '@MP{stmt.slot}={{...}}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
+ if self.current_channel is not None:
+ self._add_event(
+ LfoEvent(
+ tick_position=self.tick_position,
+ slot=stmt.slot,
+ delay=stmt.delay,
+ speed=stmt.speed,
+ depth=stmt.depth,
+ transition=stmt.transition,
+ is_definition=True,
+ )
+ )
def _analyze_lfo_use(self, stmt: LfoUseStmt) -> None:
self._add_event(LfoEvent(tick_position=self.tick_position, slot=stmt.slot))
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"LFO使用 'MP{stmt.slot}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_lfo_off(self, stmt: LfoOffStmt) -> None:
self._add_event(LfoEvent(tick_position=self.tick_position, slot=0, is_off=True))
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message="LFO解除 'MPOF' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_pitch_env_def(self, stmt: PitchEnvDefStmt) -> None:
- self._add_event(
- PitchEnvEvent(
- tick_position=self.tick_position,
- slot=stmt.slot,
- points=stmt.points,
- loop_points=stmt.loop_points,
- is_definition=True,
- )
- )
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"ピッチエンベロープ定義 '@EP{stmt.slot}={{...}}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
+ self._define(
+ "pitch_envelopes",
+ stmt.slot,
+ {"points": stmt.points, "loop_points": stmt.loop_points},
)
+ if self.current_channel is not None:
+ self._add_event(
+ PitchEnvEvent(
+ tick_position=self.tick_position,
+ slot=stmt.slot,
+ points=stmt.points,
+ loop_points=stmt.loop_points,
+ is_definition=True,
+ )
+ )
def _analyze_pitch_env_use(self, stmt: PitchEnvUseStmt) -> None:
self._add_event(PitchEnvEvent(tick_position=self.tick_position, slot=stmt.slot))
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"ピッチエンベロープ使用 'EP{stmt.slot}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_pitch_env_off(self, stmt: PitchEnvOffStmt) -> None:
self._add_event(
PitchEnvEvent(tick_position=self.tick_position, slot=0, is_off=True)
)
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message="ピッチエンベロープ解除 'EPOF' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_note_env_def(self, stmt: NoteEnvDefStmt) -> None:
- self._add_event(
- NoteEnvEvent(
- tick_position=self.tick_position,
- slot=stmt.slot,
- points=stmt.points,
- loop_points=stmt.loop_points,
- is_definition=True,
- )
- )
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"ノートエンベロープ定義 '@EN{stmt.slot}={{...}}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
+ self._define(
+ "note_envelopes",
+ stmt.slot,
+ {"points": stmt.points, "loop_points": stmt.loop_points},
)
+ if self.current_channel is not None:
+ self._add_event(
+ NoteEnvEvent(
+ tick_position=self.tick_position,
+ slot=stmt.slot,
+ points=stmt.points,
+ loop_points=stmt.loop_points,
+ is_definition=True,
+ )
+ )
def _analyze_note_env_use(self, stmt: NoteEnvUseStmt) -> None:
self._add_event(NoteEnvEvent(tick_position=self.tick_position, slot=stmt.slot))
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message=f"ノートエンベロープ使用 'EN{stmt.slot}' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_note_env_off(self, stmt: NoteEnvOffStmt) -> None:
self._add_event(
NoteEnvEvent(tick_position=self.tick_position, slot=0, is_off=True)
)
- self.ctx.add_error(
- code=ErrorCode.SEMANTIC_UNSUPPORTED_FEATURE,
- line=stmt.line,
- column=stmt.column,
- message="ノートエンベロープ解除 'ENOF' はIR保持のみです。",
- severity="warning",
- hint="現時点では無視されます。",
- context=self._context_line(stmt),
- )
def _analyze_tempo(self, stmt: TempoStmt) -> None:
if stmt.value < 1:
@@ -517,7 +483,8 @@ class PpmckSemanticAnalyzer(SemanticAnalyzer):
context=self._context_line(stmt),
)
return
- self.note_sequence.bpm = stmt.value
+ if self.tick_position == 0:
+ 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:
diff --git a/src/mml_composemusic_mcp/semantic_pyxel.py b/src/mml_composemusic_mcp/semantic_pyxel.py
index 7bf94ba..74a6c29 100644
--- a/src/mml_composemusic_mcp/semantic_pyxel.py
+++ b/src/mml_composemusic_mcp/semantic_pyxel.py
@@ -54,10 +54,36 @@ class RepeatFrame:
class PyxelSemanticAnalyzer(SemanticAnalyzer):
def __init__(self, source: str, program: Program) -> None:
super().__init__(source, program)
+ self.note_sequence.mode = "pyxel"
self.transpose = 0
self.detune_cents = 0.0
self.repeat_stack: list[RepeatFrame] = []
+ def analyze(self) -> tuple[NoteSequence, list[ErrorDetail]]:
+ ns, errors = super().analyze()
+ references = (
+ (EnvelopeEvent, "envelopes"),
+ (VibratoEvent, "vibratos"),
+ (GlideEvent, "glides"),
+ )
+ for channel in ns.channels.values():
+ for event in channel.events:
+ for event_type, table_name in references:
+ if (
+ isinstance(event, event_type)
+ and event.slot != 0
+ and event.slot not in ns.definitions[table_name]
+ ):
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_UNDEFINED_REFERENCE,
+ 0,
+ 0,
+ f"未定義の@{table_name}スロット {event.slot} です。",
+ "error",
+ "パラメータ付きで定義してから選択してください。",
+ )
+ return ns, errors
+
def _reset_state(self, track: Track) -> None:
self.octave = 4
self.default_length = 4
@@ -256,7 +282,8 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
context=self._context_line(stmt),
)
return
- self.note_sequence.bpm = stmt.value
+ if self.tick_position == 0:
+ 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:
@@ -358,7 +385,33 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
self._add_event(ev)
return
if stmt.params:
+ table_name = {
+ "ENV": "envelopes",
+ "VIB": "vibratos",
+ "GLI": "glides",
+ }[stmt.cmd]
+ if stmt.slot in self.note_sequence.definitions[table_name]:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_DUPLICATE_DEFINITION,
+ stmt.line,
+ stmt.column,
+ f"@{stmt.cmd}スロット {stmt.slot} は既に定義されています。",
+ "error",
+ "別のスロット番号を使用してください。",
+ self._context_line(stmt),
+ )
+ return
if stmt.cmd == "ENV":
+ if len(stmt.params) % 2:
+ self.ctx.add_error(
+ ErrorCode.SEMANTIC_VALUE_OUT_OF_RANGE,
+ stmt.line,
+ stmt.column,
+ "@ENVのパラメータはtarget,durationの組で指定してください。",
+ "error",
+ context=self._context_line(stmt),
+ )
+ return
points = []
for i in range(1, len(stmt.params), 2):
duration = stmt.params[i]
@@ -373,6 +426,9 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
slot=stmt.slot,
points=points,
)
+ self.note_sequence.definitions["envelopes"][stmt.slot] = {
+ "points": points
+ }
elif stmt.cmd == "VIB":
if len(stmt.params) >= 3:
params = {
@@ -387,6 +443,7 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
slot=stmt.slot,
params=params,
)
+ self.note_sequence.definitions["vibratos"][stmt.slot] = params
else:
if len(stmt.params) >= 2:
params = {
@@ -400,6 +457,7 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
slot=stmt.slot,
params=params,
)
+ self.note_sequence.definitions["glides"][stmt.slot] = params
self._add_event(ev)
else:
if stmt.cmd == "ENV":
@@ -409,15 +467,6 @@ class PyxelSemanticAnalyzer(SemanticAnalyzer):
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(
diff --git a/src/mml_composemusic_mcp/server.py b/src/mml_composemusic_mcp/server.py
index 9e9ac67..06520a6 100644
--- a/src/mml_composemusic_mcp/server.py
+++ b/src/mml_composemusic_mcp/server.py
@@ -253,6 +253,8 @@ def _dict_to_note_sequence(data: dict) -> NoteSequence:
version=data["version"],
bpm=data["bpm"],
ticks_per_quarter=data["ticks_per_quarter"],
+ mode=data.get("mode", "ppmck"),
+ definitions=data.get("definitions", {}),
channels={},
)
for name, ch_data in data["channels"].items():
diff --git a/src/mml_composemusic_mcp/synthesizer.py b/src/mml_composemusic_mcp/synthesizer.py
index b314aa5..ae98267 100644
--- a/src/mml_composemusic_mcp/synthesizer.py
+++ b/src/mml_composemusic_mcp/synthesizer.py
@@ -1,64 +1,255 @@
-"""Retro APU-style synthesizer: NoteSequence -> WAV."""
+"""Deterministic, minimal NTSC 2A03 synthesizer: NoteSequence -> WAV."""
import wave
-from math import floor
+from dataclasses import dataclass
from pathlib import Path
import numpy as np
from .ir import (
+ DetuneEvent,
+ DutyEnvelopeEvent,
DutyEvent,
+ EnvelopeEvent,
ErrorCode,
ErrorDetail,
ErrorPhase,
+ GlideEvent,
+ LfoEvent,
+ NoteEnvEvent,
NoteEvent,
NoteSequence,
+ PitchEnvEvent,
+ RelativeVolumeEvent,
RestEvent,
+ SweepEvent,
TempoEvent,
+ VibratoEvent,
+ VolumeEnvelopeEvent,
VolumeEvent,
)
CPU_CLOCK = 1_789_772.5
SAMPLE_RATE_DEFAULT = 44100
+ENVELOPE_STEP_TICKS = 48
+NOISE_PERIODS = (
+ 4,
+ 8,
+ 16,
+ 32,
+ 64,
+ 96,
+ 128,
+ 160,
+ 202,
+ 254,
+ 380,
+ 508,
+ 762,
+ 1016,
+ 2034,
+ 4068,
+)
+PULSE_SEQUENCES = (
+ (0, 1, 0, 0, 0, 0, 0, 0),
+ (0, 1, 1, 0, 0, 0, 0, 0),
+ (0, 1, 1, 1, 1, 0, 0, 0),
+ (1, 0, 0, 1, 1, 1, 1, 1),
+)
+TRIANGLE_SEQUENCE = (
+ 15,
+ 14,
+ 13,
+ 12,
+ 11,
+ 10,
+ 9,
+ 8,
+ 7,
+ 6,
+ 5,
+ 4,
+ 3,
+ 2,
+ 1,
+ 0,
+ 0,
+ 1,
+ 2,
+ 3,
+ 4,
+ 5,
+ 6,
+ 7,
+ 8,
+ 9,
+ 10,
+ 11,
+ 12,
+ 13,
+ 14,
+ 15,
+)
+
+
+def midi_to_freq(note_number: float) -> float:
+ return 440.0 * 2.0 ** ((note_number - 69.0) / 12.0)
+
+
+def frequency_to_timer(frequency: float, channel_type: str) -> int:
+ divider = 16.0 if channel_type == "pulse" else 32.0
+ if frequency <= 0:
+ return 0x7FF
+ return max(0, min(0x7FF, round(CPU_CLOCK / (divider * frequency) - 1.0)))
+
+
+def timer_to_frequency(timer: int, channel_type: str) -> float:
+ divider = 16.0 if channel_type == "pulse" else 32.0
+ return CPU_CLOCK / (divider * (timer + 1))
+
+
+def noise_period_to_rate(
+ period: int, mode: int = 0, sample_rate: int = SAMPLE_RATE_DEFAULT
+) -> float:
+ del mode, sample_rate
+ return CPU_CLOCK / NOISE_PERIODS[period & 0x0F]
+
+
+def apu_mix(
+ pulse1: np.ndarray, pulse2: np.ndarray, triangle: np.ndarray, noise: np.ndarray
+) -> np.ndarray:
+ pulse_sum = pulse1 + pulse2
+ pulse_out = np.zeros_like(pulse_sum, dtype=np.float64)
+ mask = pulse_sum > 0
+ pulse_out[mask] = 95.88 / (8128.0 / pulse_sum[mask] + 100.0)
+ tnd_input = triangle / 8227.0 + noise / 12241.0
+ tnd_out = np.zeros_like(tnd_input, dtype=np.float64)
+ mask = tnd_input > 0
+ tnd_out[mask] = 159.79 / (1.0 / tnd_input[mask] + 100.0)
+ return pulse_out + tnd_out
+
+
+@dataclass
+class TempoMap:
+ events: list[tuple[int, int]]
+ ticks_per_quarter: int = 192
+
+ def tick_to_second(self, tick: float) -> float:
+ elapsed = 0.0
+ previous_tick = 0
+ bpm = self.events[0][1]
+ for event_tick, event_bpm in self.events[1:]:
+ if tick <= event_tick:
+ break
+ elapsed += (
+ (event_tick - previous_tick) * 60.0 / (bpm * self.ticks_per_quarter)
+ )
+ previous_tick, bpm = event_tick, event_bpm
+ return elapsed + (tick - previous_tick) * 60.0 / (bpm * self.ticks_per_quarter)
+
+ def seconds_to_ticks(self, seconds: np.ndarray) -> np.ndarray:
+ result = np.empty_like(seconds)
+ segment_seconds = 0.0
+ for index, (start_tick, bpm) in enumerate(self.events):
+ end_tick = (
+ self.events[index + 1][0] if index + 1 < len(self.events) else None
+ )
+ end_seconds = (
+ (
+ segment_seconds
+ + (end_tick - start_tick) * 60.0 / (bpm * self.ticks_per_quarter)
+ )
+ if end_tick is not None
+ else np.inf
+ )
+ mask = (seconds >= segment_seconds) & (seconds < end_seconds)
+ result[mask] = (
+ start_tick
+ + (seconds[mask] - segment_seconds)
+ * bpm
+ * self.ticks_per_quarter
+ / 60.0
+ )
+ segment_seconds = end_seconds
+ return result
+
+
+def build_tempo_map(note_sequence: NoteSequence) -> TempoMap:
+ changes = {0: note_sequence.bpm}
+ for channel in note_sequence.channels.values():
+ for event in channel.events:
+ if isinstance(event, TempoEvent):
+ changes[event.tick_position] = event.bpm
+ return TempoMap(sorted(changes.items()), note_sequence.ticks_per_quarter)
+
+
+def tick_to_second(tick: int, bpm: int, ticks_per_quarter: int = 192) -> float:
+ return tick * 60.0 / (bpm * ticks_per_quarter)
-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:
- # Retro 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
+def _sequence_value(
+ definition: dict | None, elapsed_ticks: float, default: float
+) -> float:
+ if not definition:
+ return default
+ head = list(definition.get("points", []))
+ loop = list(definition.get("loop_points", []))
+ values = head + loop
+ if not values:
+ return default
+ index = max(0, int(elapsed_ticks // ENVELOPE_STEP_TICKS))
+ if index < len(values):
+ return values[index]
+ if loop:
+ return loop[(index - len(head)) % len(loop)]
+ return values[-1]
+
+
+def _triangle_lfo(elapsed: float, delay: float, period: float, depth: float) -> float:
+ if elapsed < delay or period <= 0:
+ return 0.0
+ phase = ((elapsed - delay) / period) % 1.0
+ return (1.0 - 4.0 * abs(phase - 0.5)) * depth
+
+
+def _definition(table: dict, slot: object) -> dict | None:
+ return table.get(slot) or table.get(str(slot))
+
+
+def _pyxel_envelope(points: list[dict], elapsed: float, initial: float) -> float:
+ value = initial * 127.0 / 15.0
+ remaining = elapsed
+ for point in points:
+ target = point["target_volume"]
+ duration = point["duration_ticks"]
+ if duration <= 0:
+ value = target
+ continue
+ if remaining <= duration:
+ return round(
+ (value + (target - value) * remaining / duration) * 15.0 / 127.0
+ )
+ value = target
+ remaining -= duration
+ return round(value * 15.0 / 127.0)
class ChannelSynthesizer:
def __init__(
- self, channel_type: str, sample_rate: int = SAMPLE_RATE_DEFAULT
+ self,
+ channel_type: str,
+ sample_rate: int = SAMPLE_RATE_DEFAULT,
+ channel_name: str = "Pulse1",
+ noise_mode: int = 0,
) -> None:
self.channel_type = channel_type
+ self.channel_name = channel_name
+ self.noise_mode = 1 if noise_mode else 0
self.sample_rate = sample_rate
self.wave = np.zeros(0, dtype=np.float64)
+ self._triangle_phase = 0.0
+ self._noise_lfsr = 1
+ self._noise_accumulator = 0.0
def render(
self,
@@ -66,162 +257,236 @@ class ChannelSynthesizer:
total_ticks: int,
bpm: int,
mode: str,
+ definitions: dict | None = None,
+ tempo_map: TempoMap | None = None,
) -> np.ndarray:
- seconds = tick_to_second(total_ticks, bpm)
- length = int(seconds * self.sample_rate)
+ definitions = definitions or {}
+ tempo_map = tempo_map or TempoMap([(0, bpm)])
+ length = int(tempo_map.tick_to_second(total_ticks) * 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 length == 0:
+ return self.wave
+ seconds = np.arange(length, dtype=np.float64) / self.sample_rate
+ ticks = tempo_map.seconds_to_ticks(seconds)
+ notes = [e for e in events if isinstance(e, NoteEvent)]
+ controls = sorted(
+ (
+ e
+ for e in events
+ if not isinstance(e, (NoteEvent, RestEvent, TempoEvent))
+ ),
+ key=lambda e: e.tick_position,
+ )
+ for note in notes:
+ mask = (ticks >= note.tick_position) & (
+ ticks < note.tick_position + note.duration
)
- if start_sample >= length or end_sample <= start_sample:
- continue
- if isinstance(event, RestEvent):
+ if not np.any(mask):
continue
- self._render_note(
- start_sample,
- end_sample,
- event,
- last_velocity,
- last_duty,
- mode,
- )
+ local_ticks = ticks[mask] - note.tick_position
+ indices = np.flatnonzero(mask)
+ self._render_note(indices, local_ticks, note, controls, definitions, mode)
return self.wave
def _render_note(
self,
- start: int,
- end: int,
- event: NoteEvent,
- velocity: int,
- duty: int,
+ indices: np.ndarray,
+ local_ticks: np.ndarray,
+ note: NoteEvent,
+ controls: list,
+ definitions: dict,
mode: str,
) -> None:
- duration_samples = end - start
- if duration_samples <= 0:
- return
+ active = [e for e in controls if e.tick_position <= note.tick_position]
+ volume = (
+ note.velocity
+ if mode == "ppmck"
+ else round(note.velocity * 15 / 127)
+ if note.velocity > 15
+ else note.velocity
+ )
+ duty = note.duty
+ detune = note.detune_cents
+ selections: dict[str, object | None] = {
+ "vol": None,
+ "duty": None,
+ "pitch": None,
+ "note": None,
+ "lfo": None,
+ "env": None,
+ "vib": None,
+ "glide": None,
+ }
+ sweep: SweepEvent | None = None
+ for event in active:
+ if isinstance(event, VolumeEvent):
+ volume = event.value
+ elif isinstance(event, RelativeVolumeEvent):
+ # Semantic analysis already stores the resulting velocity on
+ # each note; this event additionally cancels @v selection.
+ selections["vol"] = None
+ elif isinstance(event, DutyEvent):
+ duty = event.value
+ selections["duty"] = None
+ elif isinstance(event, DetuneEvent):
+ detune = event.value
+ elif isinstance(event, SweepEvent):
+ sweep = event
+ elif isinstance(event, VolumeEnvelopeEvent) and not event.is_definition:
+ selections["vol"] = event.slot
+ elif isinstance(event, DutyEnvelopeEvent) and not event.is_definition:
+ selections["duty"] = event.slot
+ elif isinstance(event, PitchEnvEvent) and not event.is_definition:
+ selections["pitch"] = None if event.is_off else event.slot
+ elif isinstance(event, NoteEnvEvent) and not event.is_definition:
+ selections["note"] = None if event.is_off else event.slot
+ elif isinstance(event, LfoEvent) and not event.is_definition:
+ selections["lfo"] = None if event.is_off else event.slot
+ elif isinstance(event, EnvelopeEvent):
+ selections["env"] = None if event.slot == 0 else event.slot
+ elif isinstance(event, VibratoEvent):
+ selections["vib"] = None if event.slot == 0 else event.slot
+ elif isinstance(event, GlideEvent):
+ selections["glide"] = None if event.slot == 0 else event.slot
+ vol_values = np.full(len(indices), float(volume))
+ duty_values = np.full(len(indices), int(duty))
+ cents = np.full(len(indices), float(detune))
+ for i, tick in enumerate(local_ticks):
+ if selections["vol"] is not None:
+ vol_values[i] = _sequence_value(
+ _definition(
+ definitions.get("volume_envelopes", {}), selections["vol"]
+ ),
+ tick,
+ volume,
+ )
+ if selections["duty"] is not None:
+ duty_values[i] = int(
+ _sequence_value(
+ _definition(
+ definitions.get("duty_envelopes", {}), selections["duty"]
+ ),
+ tick,
+ duty,
+ )
+ )
+ cents[i] += 100.0 * _sequence_value(
+ _definition(definitions.get("note_envelopes", {}), selections["note"]),
+ tick,
+ 0,
+ )
+ cents[i] += _sequence_value(
+ _definition(
+ definitions.get("pitch_envelopes", {}), selections["pitch"]
+ ),
+ tick,
+ 0,
+ )
+ lfo = _definition(definitions.get("lfos", {}), selections["lfo"])
+ if lfo:
+ cents[i] += _triangle_lfo(
+ tick, lfo["delay"], lfo["speed"], lfo["depth"]
+ )
+ vib = _definition(definitions.get("vibratos", {}), selections["vib"])
+ if vib:
+ cents[i] += _triangle_lfo(
+ tick,
+ vib.get("delay_ticks", 0),
+ vib.get("period_ticks", 0),
+ vib.get("depth_cents", 0),
+ )
+ gli = _definition(definitions.get("glides", {}), selections["glide"])
+ if gli:
+ duration = gli.get("duration_ticks", 0)
+ cents[i] += (
+ gli.get("initial_offset_cents", 0) * max(0.0, 1.0 - tick / duration)
+ if duration
+ else 0
+ )
+ env = _definition(definitions.get("envelopes", {}), selections["env"])
+ if env:
+ vol_values[i] = _pyxel_envelope(env.get("points", []), tick, volume)
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
+ self._render_noise(indices, vol_values, note)
+ elif self.channel_type == "triangle":
+ self._render_triangle(indices, note, cents)
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
+ self._render_pulse(indices, note, cents, vol_values, duty_values, sweep)
+ def _render_pulse(self, indices, note, cents, volumes, duties, sweep):
+ phase = 0.0
+ sweep_elapsed = 0.0
+ timer = frequency_to_timer(
+ midi_to_freq(note.note_number + cents[0] / 100), "pulse"
+ )
+ for j, output_index in enumerate(indices):
+ if sweep and sweep.speed > 0 and sweep.depth != 0:
+ sweep_elapsed += 1 / self.sample_rate
+ interval = (sweep.speed + 1) / 120.0
+ if sweep_elapsed >= interval:
+ shift = abs(sweep.depth)
+ delta = timer >> shift
+ if sweep.depth < 0:
+ timer -= delta + (1 if self.channel_name == "Pulse1" else 0)
+ else:
+ timer += delta
+ sweep_elapsed -= interval
+ else:
+ timer = frequency_to_timer(
+ midi_to_freq(note.note_number + cents[j] / 100), "pulse"
+ )
+ target = (
+ timer + (timer >> abs(sweep.depth))
+ if sweep and sweep.depth > 0
+ else timer
+ )
+ if timer < 8 or target > 0x7FF or timer > 0x7FF:
+ self.wave[output_index] = 0
+ continue
+ phase = (
+ phase + timer_to_frequency(timer, "pulse") * 8 / self.sample_rate
+ ) % 8
+ self.wave[output_index] = PULSE_SEQUENCES[int(duties[j]) & 3][
+ int(phase)
+ ] * max(0, min(15, volumes[j]))
+ gate_index = int(len(indices) * max(0, min(1, note.gate_time)))
+ self.wave[indices[gate_index:]] = 0
-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 _render_triangle(self, indices, note, cents):
+ phase = self._triangle_phase
+ for j, output_index in enumerate(indices):
+ timer = frequency_to_timer(
+ midi_to_freq(note.note_number + cents[j] / 100), "triangle"
+ )
+ phase = (
+ phase + timer_to_frequency(timer, "triangle") * 32 / self.sample_rate
+ ) % 32
+ self.wave[output_index] = TRIANGLE_SEQUENCE[int(phase)]
+ self._triangle_phase = phase
+ # The 2A03 holds the current DAC value when gated, rather than forcing zero.
+ gate_index = int(len(indices) * max(0, min(1, note.gate_time)))
+ if 0 < gate_index < len(indices):
+ self.wave[indices[gate_index:]] = self.wave[indices[gate_index - 1]]
+
+ def _render_noise(self, indices, volumes, note):
+ period_index = (
+ 8 if note.note_number == 0 else max(0, min(15, 15 - note.note_number // 8))
+ )
+ cycles_per_sample = CPU_CLOCK / (NOISE_PERIODS[period_index] * self.sample_rate)
+ accumulator = self._noise_accumulator
+ lfsr = self._noise_lfsr
+ for j, output_index in enumerate(indices):
+ accumulator += cycles_per_sample
+ for _ in range(int(accumulator)):
+ tap = 6 if self.noise_mode else 1
+ feedback = (lfsr & 1) ^ ((lfsr >> tap) & 1)
+ lfsr = (lfsr >> 1) | (feedback << 14)
+ accumulator %= 1.0
+ self.wave[output_index] = 0 if lfsr & 1 else max(0, min(15, volumes[j]))
+ self._noise_accumulator = accumulator
+ self._noise_lfsr = lfsr
+ gate_index = int(len(indices) * max(0, min(1, note.gate_time)))
+ self.wave[indices[gate_index:]] = 0
def synthesize(
@@ -231,23 +496,35 @@ def synthesize(
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())
+ total_ticks = max(
+ (ch.total_ticks for ch in note_sequence.channels.values()), default=0
+ )
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
+ tempo_map = build_tempo_map(note_sequence)
+ rendered: dict[str, np.ndarray] = {}
+ for name, channel in note_sequence.channels.items():
+ rendered[name] = ChannelSynthesizer(
+ channel.channel_type, sample_rate, name
+ ).render(
+ channel.events,
+ total_ticks,
+ note_sequence.bpm,
+ mode,
+ note_sequence.definitions,
+ tempo_map,
+ )
+ zero = np.zeros(max(map(len, rendered.values())), dtype=np.float64)
+ mixed = apu_mix(
+ rendered.get("Pulse1", zero),
+ rendered.get("Pulse2", zero),
+ rendered.get("Triangle", zero),
+ rendered.get("Noise", zero),
+ )
+ if normalize and mixed.size and np.max(np.abs(mixed)) > 0:
+ mixed /= np.max(np.abs(mixed))
+ mixed = np.clip(mixed, -1, 1)
+ return mixed, len(mixed) / sample_rate, errors
def write_wav(path: Path, wave_data: np.ndarray, sample_rate: int) -> list[ErrorDetail]:
@@ -269,29 +546,37 @@ def write_wav(path: Path, wave_data: np.ndarray, sample_rate: int) -> list[Error
column=0,
message=f"WAVファイルの出力に失敗しました: {exc}",
severity="error",
- hint="しばらく待ってから再度お試しください。",
+ 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
+ channels = (
+ note_sequence["channels"]
+ if isinstance(note_sequence, dict)
+ else 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]
+ for name, channel in channels.items():
+ events = channel["events"] if isinstance(channel, dict) else channel.events
+ total_ticks = (
+ channel["total_ticks"] if isinstance(channel, dict) else channel.total_ticks
+ )
+ notes = [
+ e
+ for e in events
+ if (
+ isinstance(e, NoteEvent)
+ or isinstance(e, dict)
+ and e.get("type") == "note"
+ )
+ ]
+ numbers = [
+ e["note_number"] if isinstance(e, dict) else e.note_number for e in notes
+ ]
+ octaves = [n // 12 - 1 for n in numbers]
summary.append(
{
"channel": name,