🐅
手元の音源を分割してマルコフ連鎖で無限ループ再生
事前準備
- 音源
ffmpeg- ライブラリインストール
pip install pydub==0.25.1 pygame==2.5.2
スクリプト
スクリプト
markov.py
from __future__ import annotations
import random
import time
import traceback
from argparse import ArgumentParser, Namespace
from collections import defaultdict
from dataclasses import dataclass
from enum import Enum, auto
from pathlib import Path
from typing import Callable, DefaultDict, Dict, List, Optional
import pygame
from pydub import AudioSegment
from pygame.mixer import Sound
DEFAULT_AUDIO_INPUT_EXT: str = "mp3"
STR_MARKOV_PRESETS: Dict[str, Callable[[Optional[int]], StrMarkovPreset]] = {
key: lambda audio_offset: value(
audio_offset
if isinstance(audio_offset, int) and 0 <= audio_offset
else 346
)
for key, value in dict(
deer=lambda _audio_offset: StrMarkovPreset(
name="deer",
chains=[
StrMarkovChainInfo(
kind=StrMarkovChainKind.init,
ids=[
"ぬ_1",
"ん_1",
],
states=[
StrMarkovState(
id="ぬ_1",
text="ぬ",
audio_start=_audio_offset,
audio_end=_audio_offset + 179,
),
StrMarkovState(
id="ん_1",
text="ん",
audio_start=_audio_offset + 179,
audio_end=_audio_offset + 314,
),
StrMarkovState(
id="silent_1",
text="",
is_start=True,
is_end=True,
),
],
links=[
StrMarkovLink(
current_state="silent_1",
next_state="ぬ_1",
),
StrMarkovLink(
current_state="ぬ_1",
next_state="ん_1",
),
StrMarkovLink(
current_state="ん_1",
next_state="silent_1",
),
],
),
StrMarkovChainInfo(
kind=StrMarkovChainKind.main,
ids=[
"し_1",
"か_1",
"の_1",
"こ_1",
"の_1",
"こ_1",
"の_1",
"こ_1",
"こ_1",
"し_1",
"た_1",
"ん_1",
"た_1",
"ん_1",
"rest_1",
],
states=[
StrMarkovState(
id="し_1",
text="し",
audio_start=_audio_offset + 314,
audio_end=_audio_offset + 514,
),
StrMarkovState(
id="か_1",
text="か",
audio_start=_audio_offset + 514,
audio_end=_audio_offset + 691,
),
StrMarkovState(
id="の_1",
text="の",
audio_start=_audio_offset + 691,
audio_end=_audio_offset + 855,
),
StrMarkovState(
id="こ_1",
text="こ",
audio_start=_audio_offset + 855,
audio_end=_audio_offset + 1015,
),
StrMarkovState(
id="た_1",
text="た",
audio_start=_audio_offset + 2323,
audio_end=_audio_offset + 2492,
),
StrMarkovState(
id="ん_1",
text="ん",
audio_start=_audio_offset + 2492,
audio_end=_audio_offset + 2655,
),
StrMarkovState(
id="rest_1",
text="",
audio_start=_audio_offset + 2655,
audio_end=_audio_offset + 2827,
is_start=True,
is_end=True,
),
],
links=[
StrMarkovLink(
current_state="rest_1",
next_state="rest_1",
),
StrMarkovLink(
current_state="rest_1",
next_state="し_1",
),
StrMarkovLink(
current_state="し_1",
next_state="か_1",
),
StrMarkovLink(
current_state="か_1",
next_state="の_1",
),
StrMarkovLink(
current_state="の_1",
next_state="こ_1",
),
StrMarkovLink(
current_state="こ_1",
next_state="の_1",
weight=1 / 2,
),
StrMarkovLink(
current_state="こ_1",
next_state="こ_1",
weight=1 / 4,
),
StrMarkovLink(
current_state="こ_1",
next_state="し_1",
weight=1 / 4,
),
StrMarkovLink(
current_state="し_1",
next_state="た_1",
),
StrMarkovLink(
current_state="た_1",
next_state="ん_1",
),
StrMarkovLink(
current_state="ん_1",
next_state="た_1",
),
StrMarkovLink(
current_state="ん_1",
next_state="rest_1",
),
],
),
],
)
).items()
}
def main() -> None:
parser: ArgumentParser = ArgumentParser()
parser.add_argument(
"-p",
"--preset",
choices=STR_MARKOV_PRESETS.keys(),
default="deer",
help="preset",
)
parser.add_argument(
"-c",
"--correct_only",
action="store_true",
help="play correct answer only",
)
parser.add_argument(
"-e",
"--exit",
action="store_true",
help="exit when correct answer",
)
parser.add_argument(
"-os",
"--out_start_only",
action="store_true",
help="out even if everything is start state",
)
parser.add_argument(
"-s",
"--save_audio",
action="store_true",
help="save splitted audio",
)
parser.add_argument("-i", "--input_file", help="input source file")
parser.add_argument(
"-ext",
"--default_ext",
help="default ext of input source file",
)
parser.add_argument(
"-o",
"--offset_audio",
type=int,
help="offset audio of input source file",
)
parser.add_argument(
"-is",
"--include_start",
action="store_true",
help="include start state",
)
parser.add_argument(
"-n",
"--no_play",
action="store_true",
help="no play audio",
)
args: Namespace = parser.parse_args()
try:
runner: StrMarkovChainRunner = StrMarkovChainRunner(
STR_MARKOV_PRESETS[args.preset](args.offset_audio),
correct_only=args.correct_only,
exit=args.exit,
skip_start_only=not args.out_start_only,
include_start=args.include_start,
input_file=args.input_file,
play_audio=not args.no_play,
save_audio=args.save_audio,
)
runner.run()
except Exception:
traceback.print_exc()
AUDIO_SEGMENT_SUBDIR: str = "splitted"
class StrMarkovChainKind(Enum):
init = auto()
main = auto()
class StrMarkovChainRunner:
def __init__(
self,
preset: StrMarkovPreset,
correct_only: bool = False,
exit: bool = False,
skip_start_only: bool = False,
input_file: str = None,
play_audio: bool = False,
save_audio: bool = False,
include_start: bool = False,
) -> None:
self.preset: StrMarkovPreset = preset
self.chain: StrMarkovChain = StrMarkovChain(
self.preset,
include_start=include_start,
)
self.correct_only: bool = correct_only
self.exit: bool = exit
self.skip_start_only: bool = skip_start_only
self.play_audio: bool = play_audio
self.sounds: Dict[StrMarkovState, tuple[Sound, float]] = (
self._check_audio_file(
save_audio=save_audio, input_file=input_file
)
if self.play_audio
else None
)
def run(self) -> None:
print("markov chain transition start:", self.preset.name)
get_count: int = 0
correct_count: int = 0
try:
if StrMarkovChainKind.init in self.chain.chains:
init_transition_end: bool = False
while not init_transition_end:
result: List[StrMarkovState] = self.chain.transition(
kind=StrMarkovChainKind.init
)
if self._is_start_only(
result, kind=StrMarkovChainKind.init
):
continue
print(
f"{StrMarkovChainKind.init.name} transition:",
[s.text for s in result],
*(("play",) if self.play_audio else ()),
)
if self.play_audio:
[self._play(*self.sounds[s]) for s in result]
init_transition_end = True
main_transition_end: bool = False
while not main_transition_end:
try:
result = self.chain.transition()
if self._is_start_only(result):
continue
get_count += 1
result_strs: List[str] = [s.text for s in result]
is_correct: bool = self.chain.is_correct_answer(result)
if is_correct:
correct_count += 1
elif self.correct_only:
continue
if self.play_audio:
zfill_len: int = max((2, len(f"{len(result)}")))
for i, r in enumerate(result):
sound, seconds = self.sounds[r]
print(
"count:",
f"{get_count:,}",
f"({is_correct}):",
result_strs,
"play :",
f'"{r.text}"',
"(" + f"{i + 1}".zfill(zfill_len),
"/",
f"{len(result)}".zfill(zfill_len) + ")",
)
self._play(sound, seconds)
else:
print(
"count:",
f"{get_count:,}",
f"({is_correct}):",
result_strs,
)
print(
"count:",
f"{get_count:,}",
*(
(
"CORRECT ANSWER!!!!!!!!!!:",
f'"{self.preset.answer_str(StrMarkovChainKind.main)}"',
)
if is_correct
else (f"({is_correct}):",)
),
self._summary(get_count, correct_count),
)
if is_correct and self.exit:
main_transition_end = True
except KeyboardInterrupt:
main_transition_end = True
finally:
print(
"markov chain transition end:",
f"{self.preset.name}\nsummary:",
self._summary(get_count, correct_count),
)
def _check_audio_file(
self,
input_file: str = None,
save_audio: bool = False,
) -> Dict[StrMarkovState, tuple[Sound, float]]:
pygame.init()
input_path, segments_path = self.preset.make_audio_path(
input_file=input_file
)
if save_audio or 0 < len(
[p for p in segments_path.values() if not p.is_file()]
):
print("split audio start", input_path)
input_audio: AudioSegment = AudioSegment.from_file(input_path)
for state, save_path in segments_path.items():
if save_audio or not save_path.is_file():
audio_segment: AudioSegment = input_audio[
state.audio_start : state.audio_end
]
if not save_path.parent.is_dir():
save_path.parent.mkdir(parents=True, exist_ok=True)
audio_segment.export(save_path, format="wav")
print("split audio end", input_path)
return {
s: (Sound(p), AudioSegment.from_file(p).duration_seconds)
for s, p in segments_path.items()
}
def _is_start_only(
self,
result: List[StrMarkovState],
kind: StrMarkovChainKind = StrMarkovChainKind.main,
) -> bool:
return self.skip_start_only and 0 == len(
[r for r in result if r != self.chain.chains[kind].start]
)
@classmethod
def _play(cls, sound: Sound, seconds: float) -> None:
start: float = time.time()
sound.play()
while time.time() - start <= seconds:
time.sleep(0.0003)
@classmethod
def _summary(cls, get_count: int, correct_count: int) -> str:
return (
" ".join(
[
f"{correct_count:,}",
"/",
f"{get_count:,}",
f"({int((correct_count / get_count) * 100 * 1000**2) / (1000**2)} %)",
]
)
if 0 != get_count
else "no data"
)
class StrMarkovChain:
def __init__(
self,
preset: StrMarkovPreset,
include_start: bool = False,
) -> None:
self.preset: StrMarkovPreset = preset
self.include_start: bool = include_start
self.chains: Dict[StrMarkovChainKind, StrMarkovChainStructure] = (
self.preset.create_chains()
)
self.answer_states: Dict[StrMarkovChainKind, List[StrMarkovState]] = {
kind: ([chain.start] if self.include_start else [])
+ self.preset.answer_states(kind)
+ [chain.end]
for kind, chain in self.chains.items()
}
self.end_counts: Dict[StrMarkovChainKind, int] = {}
for kind, states in self.answer_states.items():
end_count: int = 0
for st in reversed(states):
if st.is_end:
end_count += 1
else:
break
self.end_counts.update({kind: end_count})
def transition(
self, kind: StrMarkovChainKind = StrMarkovChainKind.main
) -> List[StrMarkovState]:
chain: StrMarkovChainStructure = self.chains[kind]
end_count: int = self.end_counts[kind]
current_state: StrMarkovState = chain.start
next_state: StrMarkovState = None
state_changed: bool = False
pass_end_count: int = 0
result: List[StrMarkovState] = (
[current_state] if self.include_start else []
)
while pass_end_count < end_count:
next_state = random.choices(
chain.chain[current_state],
weights=chain.weights[current_state],
)[0]
state_changed = state_changed or current_state != next_state
if state_changed:
result.append(next_state)
if 0 < pass_end_count and next_state != chain.end:
pass_end_count = 0
elif next_state == chain.end:
pass_end_count += 1
current_state = next_state
return result
def is_correct_answer(
self,
result: List[StrMarkovState],
kind: StrMarkovChainKind = StrMarkovChainKind.main,
) -> bool:
if len(result) != len(self.answer_states[kind]):
return False
for i, r in enumerate(result):
if r != self.answer_states[kind][i]:
return False
return True
@dataclass
class StrMarkovState:
id: str
text: str
audio_start: Optional[int] = 0
audio_end: Optional[int] = 0
is_start: Optional[bool] = False
is_end: Optional[bool] = False
def __hash__(self) -> int:
return "_".join(
[
f"{v}"
for v in [
self.id,
self.audio_start,
self.audio_end,
self.is_start,
self.is_end,
]
]
).__hash__()
def __eq__(self, value: object) -> bool:
return (
isinstance(value, StrMarkovState)
and value.__hash__() == self.__hash__()
)
@dataclass
class StrMarkovLink:
current_state: str
next_state: str
weight: Optional[int | float] = 1
@dataclass
class StrMarkovChainInfo:
kind: StrMarkovChainKind
ids: List[str]
states: List[StrMarkovState]
links: List[StrMarkovLink]
@dataclass
class StrMarkovChainStructure:
kind: StrMarkovChainKind
answer_states: List[StrMarkovState]
start: StrMarkovState
end: StrMarkovState
chain: DefaultDict[StrMarkovState, List[StrMarkovState]]
weights: DefaultDict[StrMarkovState, List[int | float]]
@dataclass
class StrMarkovPreset:
name: str
chains: List[StrMarkovChainInfo]
def make_audio_path(
self, input_file: str = None, default_input_ext: str = None
) -> tuple[Path, Dict[StrMarkovState, Path]]:
segments_path: Dict[StrMarkovState, Path] = {}
self._check_duplicate()
for info in self.chains:
segments_path.update(
{
state: Path(AUDIO_SEGMENT_SUBDIR)
/ f"{self.name}_{state.audio_start}_{state.audio_end}_{state.text}.wav"
for state in info.states
}
)
return (
Path(
input_file
or f"{self.answer_str(StrMarkovChainKind.main)}.{default_input_ext or DEFAULT_AUDIO_INPUT_EXT}"
),
segments_path,
)
def create_chains(
self,
) -> Dict[StrMarkovChainKind, StrMarkovChainStructure]:
chains: Dict[StrMarkovChainKind, StrMarkovChainStructure] = {}
self._check_duplicate()
if StrMarkovChainKind.main not in [c.kind for c in self.chains]:
raise ValueError(f"{StrMarkovChainKind.main.name} chain not found")
for info in self.chains:
chain: DefaultDict[StrMarkovState, List[StrMarkovState]] = (
defaultdict(list)
)
weights: DefaultDict[StrMarkovState, List[int | float]] = (
defaultdict(list)
)
chain_dict: Dict[str, StrMarkovState] = {
s.id: s for s in info.states
}
for link in info.links:
if link.next_state in [
c.id for c in chain[chain_dict[link.current_state]]
]:
raise ValueError(f"already appended: {link.next_state}")
chain[chain_dict[link.current_state]].append(
chain_dict[link.next_state]
)
weights[chain_dict[link.current_state]].append(link.weight)
start: StrMarkovState = None
end: StrMarkovState = None
for state in info.states:
if not isinstance(start, StrMarkovState) and state.is_start:
start = state
if not isinstance(end, StrMarkovState) and state.is_end:
end = state
for position, state in dict(start=start, end=end).items():
if not isinstance(state, StrMarkovState):
raise ValueError(
f"{position} state is not defined: {state}"
)
if state not in chain:
raise ValueError(
f"{position} state not found in chains: {state}"
)
chains.update(
{
info.kind: StrMarkovChainStructure(
kind=info.kind,
answer_states=info.states,
start=start,
end=end,
chain=chain,
weights=weights,
)
}
)
return chains
def answer_states(self, kind: StrMarkovChainKind) -> List[StrMarkovState]:
states: List[StrMarkovState] = []
for info in self.chains:
if info.kind == kind:
for a in info.ids:
for s in info.states:
if a == s.id:
states.append(s)
break
return states
raise ValueError(f"kind not found : {kind}")
def answer_str(self, kind: StrMarkovChainKind) -> str:
return "".join([s.text for s in self.answer_states(kind)])
def _check_duplicate(self) -> None:
all_kinds: set[StrMarkovChainKind] = set()
for info in self.chains:
if info.kind not in all_kinds:
all_kinds.add(info.kind)
else:
raise ValueError(f"duplicate kind: {info.kind.name}")
if len(set([s.id for s in info.states])) != len(
[s.id for s in info.states]
):
raise ValueError(
"duplicate id existed", self.name, info.kind.name
)
if __name__ == "__main__":
main()
使用方法
スクリプト実行
command
python markov.py -i 音源ファイルパス
終了はCtrl + C。
引数
引数
-
-p
プリセット名
サンプルはdeerのみ -
-c
正答のみ出力 -
-e
正答で終了 -
-os
遷移結果の全状態が初期位置だけの場合でも出力 -
-s
分割音声ファイル保存
既存ファイルは強制上書き
splitted/以下に保存 -
-i
音源ファイルパス
未指定の場合の既定ファイルパス：正答文字列.{-extの文字列} -
-ext
音源ファイルパス未指定の場合の拡張子
デフォルトはmp3 -
-o
分割音声ファイル作成時の、先頭からのオフセット (ミリ秒)
未指定の場合プリセット毎の既定値使用 -
-is
初期位置の状態を正答に含める -
-n
音声を再生しない (コンソール出力のみ)
-sの指定は無効化
実行例
python markov.py -i 音源ファイルパス
markov chain transition start: deer
init transition: ['ぬ', 'ん', ''] play
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "し" (01 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "た" (02 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "ん" (03 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "た" (04 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "ん" (05 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "" (06 / 07)
count: 1 (False): ['し', 'た', 'ん', 'た', 'ん', '', ''] play : "" (07 / 07)
count: 1 (False): 0 / 1 (0.0 %)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "し" (01 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "か" (02 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "の" (03 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "こ" (04 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "し" (05 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "た" (06 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "ん" (07 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "" (08 / 09)
count: 2 (False): ['し', 'か', 'の', 'こ', 'し', 'た', 'ん', '', ''] play : "" (09 / 09)
count: 2 (False): 0 / 2 (0.0 %)
count: 3 (False): ['し', 'た', 'ん', '', ''] play : "し" (01 / 05)
count: 3 (False): ['し', 'た', 'ん', '', ''] play : "た" (02 / 05)
count: 3 (False): ['し', 'た', 'ん', '', ''] play : "ん" (03 / 05)
count: 3 (False): ['し', 'た', 'ん', '', ''] play : "" (04 / 05)
count: 3 (False): ['し', 'た', 'ん', '', ''] play : "" (05 / 05)
count: 3 (False): 0 / 3 (0.0 %)
・・・
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "し" (01 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "か" (02 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "の" (03 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "こ" (04 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "の" (05 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "こ" (06 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "の" (07 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "こ" (08 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "こ" (09 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "し" (10 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "た" (11 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "ん" (12 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "た" (13 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "ん" (14 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "" (15 / 16)
count: 4,434 (True): ['し', 'か', 'の', 'こ', 'の', 'こ', 'の', 'こ', 'こ', 'し', 'た', 'ん', 'た', 'ん', '', ''] play : "" (16 / 16)
count: 4,434 CORRECT ANSWER!!!!!!!!!!: "しかのこのこのここしたんたん" 1 / 4,434 (0.022552 %)
・・・
markov chain transition end: deer
summary: 4 / 5,174 (0.077309 %)
プリセット変更方法
変数STR_MARKOV_PRESETS参照。
Discussion