# 各時代の音声認識アプローチ比較（概念コード）
 
# 1. 規則ベース（1960s-1980s）
def rule_based_stt(audio):
    """手動で定義した音素規則に基づく認識"""
    phonemes = extract_phonemes_by_rules(audio)
    words = match_phoneme_dictionary(phonemes)
    return words
 
# 2. 統計モデル（1990s-2010s）
def hmm_based_stt(audio):
    """HMM + GMM による確率的認識"""
    features = extract_mfcc(audio)           # メル周波数ケプストラム係数
    phoneme_probs = gmm_score(features)      # 各音素の確率計算
    best_path = viterbi_decode(phoneme_probs) # 最尤パス探索
    words = language_model_rescore(best_path) # 言語モデルで再スコア
    return words
 
# 3. End-to-End DL（2020s）
def transformer_stt(audio):
    """Transformer ベースの End-to-End 認識"""
    mel_spec = compute_mel_spectrogram(audio)
    encoder_out = transformer_encoder(mel_spec)
    text = transformer_decoder(encoder_out)  # 直接テキスト出力
    return text

# 各時代の技術的特徴を詳細に比較
 
paradigm_details = {
    "規則ベース (1950s-1970s)": {
        "特徴": [
            "音素ルールを手動で定義",
            "フォルマント周波数の分析に基づく",
            "限られた語彙（数十〜数百語）",
            "話者依存（特定の話者にのみ対応）",
        ],
        "代表的システム": {
            "Audrey (1952)": "10個の数字を認識。単一話者。認識率97%",
            "Shoebox (1961)": "16の英語単語を認識。音声で計算機を操作",
            "HARPY (1976)": "1011語の語彙。連続音声認識の先駆け",
        },
        "限界": "語彙追加に膨大な労力。環境変化に脆弱。多話者対応困難",
    },
    "統計モデル (1980s-2000s)": {
        "特徴": [
            "HMM（隠れマルコフモデル）による時系列モデリング",
            "GMM（混合ガウスモデル）による音響モデル",
            "N-gram言語モデルによる文脈理解",
            "Viterbiアルゴリズムによる最尤パス探索",
        ],
        "代表的システム": {
            "Sphinx (CMU)": "最初のオープンソース音声認識システム",
            "HTK (Cambridge)": "HMMの代表的ツールキット。研究の標準に",
            "Dragon NaturallySpeaking": "商用音声認識ソフトの先駆け",
        },
        "限界": "特徴量設計に専門知識が必要。長距離依存の処理が困難",
    },
    "ディープラーニング (2010s)": {
        "特徴": [
            "DNN-HMM ハイブリッド: DNNで音響モデルを置換",
            "RNN/LSTM: 時系列データの長期依存性を学習",
            "Attention機構: 入出力のアライメントを自動学習",
            "End-to-End: 特徴抽出からテキスト出力まで単一モデル",
        ],
        "ブレークスルー": {
            "2012 DNN-HMM": "Hintonらが音響モデルにDNNを適用。WER 30%改善",
            "2014 Seq2Seq": "音声認識をシーケンス変換問題として定式化",
            "2016 WaveNet": "生波形からの直接音声合成。自然さが飛躍的向上",
            "2017 Transformer": "Self-Attention機構の提案。NLPの革命",
        },
        "限界": "大量の学習データが必要。計算コストが高い",
    },
    "基盤モデル (2020s-現在)": {
        "特徴": [
            "大規模事前学習による汎用的な音声理解",
            "マルチタスク学習（認識+翻訳+言語検出）",
            "ゼロショット/フューショット適応",
            "マルチモーダル統合（音声+テキスト+画像）",
        ],
        "ブレークスルー": {
            "Whisper (2022)": "680K時間のデータで学習。多言語対応",
            "VALL-E (2023)": "3秒の参照音声でゼロショットTTS",
            "GPT-4o (2024)": "音声を直接理解・生成するマルチモーダルモデル",
            "Gemini Live (2024)": "リアルタイム音声対話",
        },
        "今後の方向": "エッジデバイスでの高速推論、パーソナライゼーション",
    },
}
 
# 各パラダイムのWER（単語誤り率）の推移
wer_history = {
    "1990年": {"技術": "HMM-GMM", "WER": "約40%", "対象": "読み上げ音声"},
    "2000年": {"技術": "HMM-GMM改良", "WER": "約20%", "対象": "読み上げ音声"},
    "2012年": {"技術": "DNN-HMM", "WER": "約15%", "対象": "会話音声"},
    "2016年": {"技術": "Seq2Seq + Attention", "WER": "約8%", "対象": "会話音声"},
    "2020年": {"技術": "Conformer", "WER": "約5%", "対象": "会話音声"},
    "2023年": {"技術": "Whisper large-v3", "WER": "約3%", "対象": "多言語会話"},
    "2025年": {"技術": "マルチモーダル基盤", "WER": "約2%", "対象": "多言語+ノイズ環境"},
}

# 音声AI エコシステムの主要プレイヤー（2025-2026年）
 
audio_ai_ecosystem = {
    "STT (音声認識)": {
        "オープンソース": ["Whisper (OpenAI)", "Vosk", "wav2vec 2.0"],
        "クラウドAPI": ["Google Speech-to-Text", "Azure Speech", "AWS Transcribe"],
        "特化型": ["Deepgram", "AssemblyAI", "Rev.ai"],
    },
    "TTS (音声合成)": {
        "オープンソース": ["VITS", "Bark", "Coqui TTS", "Piper"],
        "クラウドAPI": ["ElevenLabs", "Google TTS", "Azure TTS", "OpenAI TTS"],
        "特化型": ["PlayHT", "LMNT", "WellSaid Labs"],
    },
    "音声生成 (Music/Sound)": {
        "音楽生成": ["Suno", "Udio", "MusicGen (Meta)", "Stable Audio"],
        "効果音": ["AudioGen", "Make-An-Audio", "Stable Audio Open"],
        "ボイスクローン": ["RVC", "So-VITS-SVC", "OpenVoice"],
    },
    "マルチモーダル": {
        "対話型": ["GPT-4o Audio", "Gemini Live", "Claude Voice"],
        "リアルタイム": ["OpenAI Realtime API", "LiveKit", "Daily"],
    },
}

# STT（音声認識）の詳細分類
 
stt_taxonomy = {
    "アーキテクチャ別": {
        "CTC (Connectionist Temporal Classification)": {
            "説明": "入力と出力の長さが異なる問題を確率的に解決",
            "モデル例": ["wav2vec 2.0", "DeepSpeech 2"],
            "特徴": "デコーダ不要で高速。条件付き独立仮定が制約",
            "適用場面": "リアルタイム処理、エッジデバイス",
        },
        "Attention-based Encoder-Decoder": {
            "説明": "エンコーダの出力にAttentionを適用して逐次デコード",
            "モデル例": ["Whisper", "Conformer"],
            "特徴": "高精度だが自己回帰のため速度に制約",
            "適用場面": "オフライン高精度文字起こし",
        },
        "Transducer (RNN-T)": {
            "説明": "CTC + Attentionの良いとこ取り",
            "モデル例": ["Google USM", "Conformer Transducer"],
            "特徴": "ストリーミング対応かつ高精度",
            "適用場面": "リアルタイム音声認識、音声アシスタント",
        },
    },
    "処理モード別": {
        "バッチ処理": "録音済み音声を一括処理。最高精度",
        "ストリーミング": "リアルタイムで逐次認識。低遅延",
        "セミリアルタイム": "短いバッファで分割処理。バランス型",
    },
    "特殊機能": {
        "話者分離 (Diarization)": "誰が何を言ったかを識別",
        "感情認識": "声のトーンから感情を推定",
        "言語検出": "話されている言語を自動判定",
        "コードスイッチング": "複数言語が混在する発話に対応",
    },
}
 
# TTS（音声合成）の詳細分類
 
tts_taxonomy = {
    "生成方式別": {
        "自己回帰型": {
            "説明": "トークンを1つずつ逐次生成",
            "モデル例": ["Tacotron 2", "VALL-E", "Bark"],
            "特徴": "高品質だが生成速度が遅い",
        },
        "非自己回帰型": {
            "説明": "並列にトークンを生成",
            "モデル例": ["FastSpeech 2", "VITS"],
            "特徴": "高速だが品質はやや劣る場合がある",
        },
        "拡散モデル型": {
            "説明": "ノイズから徐々に音声を復元",
            "モデル例": ["Grad-TTS", "DiffGAN-TTS"],
            "特徴": "高品質だがステップ数に応じて速度が変化",
        },
        "フロー型": {
            "説明": "可逆変換で潜在空間から音声を生成",
            "モデル例": ["VITS (Flow + VAE)", "Glow-TTS"],
            "特徴": "高速かつ高品質のバランスが良い",
        },
    },
    "制御機能": {
        "プロソディ制御": "話速、ピッチ、強勢の調整",
        "感情制御": "喜怒哀楽の表現",
        "スタイル制御": "ニュース読み、会話、ナレーション等",
        "話者制御": "ゼロショット / フューショット話者適応",
    },
}

# 音声AIアプリケーションの典型的なパイプライン
 
import numpy as np
 
class AudioAIPipeline:
    """音声入力 → 処理 → 音声出力の基本パイプライン"""
 
    def __init__(self):
        self.stt_model = None   # 音声認識モデル
        self.llm = None         # 言語モデル
        self.tts_model = None   # 音声合成モデル
 
    def process(self, audio_input: np.ndarray) -> np.ndarray:
        # Step 1: 音声認識（STT）
        text = self.stt_model.transcribe(audio_input)
        print(f"認識結果: {text}")
 
        # Step 2: テキスト処理（LLM）
        response = self.llm.generate(text)
        print(f"応答テキスト: {response}")
 
        # Step 3: 音声合成（TTS）
        audio_output = self.tts_model.synthesize(response)
        print(f"合成音声: {len(audio_output)} samples")
 
        return audio_output
 
    def streaming_process(self, audio_stream):
        """ストリーミング版パイプライン"""
        for chunk in audio_stream:
            partial_text = self.stt_model.transcribe_streaming(chunk)
            if partial_text.is_final:
                response = self.llm.generate(partial_text.text)
                yield self.tts_model.synthesize_streaming(response)

# 音声AI市場の成長予測（概算）
 
market_data = {
    "2023": {"市場規模_億ドル": 120, "主要トレンド": "LLM統合の始まり"},
    "2024": {"市場規模_億ドル": 180, "主要トレンド": "マルチモーダルAIの台頭"},
    "2025": {"市場規模_億ドル": 260, "主要トレンド": "リアルタイム音声対話の普及"},
    "2026": {"市場規模_億ドル": 350, "主要トレンド": "パーソナライズド音声AIの成熟"},
}
 
# 成長率計算
for year in ["2024", "2025", "2026"]:
    prev = market_data[str(int(year) - 1)]["市場規模_億ドル"]
    curr = market_data[year]["市場規模_億ドル"]
    growth = (curr - prev) / prev * 100
    print(f"{year}年: ${curr}B（前年比 +{growth:.0f}%）- {market_data[year]['主要トレンド']}")

# 音声AI市場のセグメント別分析
 
market_segments = {
    "コンシューマー向け音声アシスタント": {
        "市場規模_2025": "約80億ドル",
        "主要プレイヤー": ["Apple Siri", "Google Assistant", "Amazon Alexa"],
        "成長ドライバー": "スマートスピーカー普及、車載AI統合",
        "課題": "プライバシー懸念、多言語対応品質",
    },
    "エンタープライズ音声ソリューション": {
        "市場規模_2025": "約50億ドル",
        "主要プレイヤー": ["Nuance (Microsoft)", "Google CCAI", "Amazon Connect"],
        "成長ドライバー": "コールセンター自動化、会議文字起こし",
        "課題": "セキュリティ要件、既存システム統合",
    },
    "音楽・クリエイティブ": {
        "市場規模_2025": "約20億ドル",
        "主要プレイヤー": ["Suno", "Udio", "Stable Audio"],
        "成長ドライバー": "コンテンツ制作需要、BGM自動生成",
        "課題": "著作権問題、品質の一貫性",
    },
    "ヘルスケア": {
        "市場規模_2025": "約15億ドル",
        "主要プレイヤー": ["Nuance DAX", "Amazon Transcribe Medical"],
        "成長ドライバー": "電子カルテ音声入力、遠隔医療",
        "課題": "医療用語精度、規制対応（HIPAA等）",
    },
    "教育・アクセシビリティ": {
        "市場規模_2025": "約10億ドル",
        "主要プレイヤー": ["Google Live Transcribe", "Otter.ai", "Microsoft Teams"],
        "成長ドライバー": "オンライン学習、聴覚障害者支援",
        "課題": "多言語対応、リアルタイム精度",
    },
}
 
# 技術トレンド（2025-2026年）
tech_trends = {
    "マルチモーダル統合": {
        "説明": "音声・テキスト・画像を統一的に処理するモデル",
        "代表例": "GPT-4o, Gemini, Claude",
        "影響": "音声アシスタントの対話品質が飛躍的に向上",
    },
    "リアルタイム音声対話": {
        "説明": "300ms以下の応答遅延を実現する技術",
        "代表例": "OpenAI Realtime API, LiveKit",
        "影響": "電話相当の自然な対話がAIで可能に",
    },
    "パーソナライゼーション": {
        "説明": "個人の声、話し方、好みに適応するAI",
        "代表例": "ボイスクローニング、適応型TTS",
        "影響": "ユーザー体験の個人最適化",
    },
    "エッジAI音声処理": {
        "説明": "スマートフォン・IoTデバイスでの音声AI実行",
        "代表例": "Apple Neural Engine, Qualcomm AI Engine",
        "影響": "プライバシー保護、オフライン動作、低遅延",
    },
    "音声透かしとAI検出": {
        "説明": "AI生成音声の識別と来歴追跡技術",
        "代表例": "AudioSeal (Meta), Watermarking standards",
        "影響": "ディープフェイク対策、信頼性確保",
    },
}

# 音声AI開発環境の構築ガイド
 
development_environment = {
    "ハードウェア推奨": {
        "GPU": "NVIDIA RTX 3060以上 (VRAM 8GB+)",
        "RAM": "16GB以上（大規模モデルは32GB推奨）",
        "ストレージ": "SSD 256GB以上（モデルキャッシュ用）",
        "マイク": "USBコンデンサーマイク（開発テスト用）",
    },
    "ソフトウェア基盤": {
        "OS": "Ubuntu 22.04+ / macOS 13+ / Windows 11",
        "Python": "3.10-3.12",
        "CUDA": "12.1+（NVIDIA GPU使用時）",
        "FFmpeg": "6.0+",
    },
    "主要パッケージ": {
        "音声処理": ["librosa", "soundfile", "pydub", "torchaudio"],
        "AI/ML": ["torch", "transformers", "openai", "faster-whisper"],
        "Web/API": ["fastapi", "websockets", "aiohttp"],
        "音声I/O": ["pyaudio", "sounddevice"],
    },
}
 
# 環境構築スクリプト（概念）
setup_commands = """
# Python仮想環境の作成
python -m venv audio_ai_env
source audio_ai_env/bin/activate
 
# 基本パッケージのインストール
pip install torch torchaudio --index-url https://download.pytorch.org/whl/cu121
pip install transformers accelerate
pip install librosa soundfile pydub
pip install openai faster-whisper
pip install fastapi uvicorn websockets
 
# FFmpeg のインストール（Ubuntu）
sudo apt install ffmpeg
 
# 動作確認
python -c "import torch; print(f'PyTorch: {torch.__version__}, CUDA: {torch.cuda.is_available()}')"
python -c "import librosa; print(f'librosa: {librosa.__version__}')"
"""

# 音声AI Hello World - 最小のSTT + TTS パイプライン
 
from openai import OpenAI
 
def audio_ai_hello_world():
    """音声AIの最小構成デモ"""
    client = OpenAI()
 
    # Step 1: 音声ファイルの文字起こし（STT）
    with open("sample_audio.mp3", "rb") as audio_file:
        transcription = client.audio.transcriptions.create(
            model="whisper-1",
            file=audio_file,
            language="ja",
        )
    print(f"認識結果: {transcription.text}")
 
    # Step 2: LLMで応答生成
    chat_response = client.chat.completions.create(
        model="gpt-4o",
        messages=[
            {"role": "system", "content": "簡潔に日本語で回答してください。"},
            {"role": "user", "content": transcription.text},
        ],
    )
    response_text = chat_response.choices[0].message.content
    print(f"応答: {response_text}")
 
    # Step 3: 音声合成（TTS）
    speech = client.audio.speech.create(
        model="tts-1",
        voice="nova",
        input=response_text,
    )
    speech.stream_to_file("response.mp3")
    print("音声ファイル生成完了: response.mp3")
 
# ローカルWhisperを使った代替実装
def local_stt_example():
    """ローカルWhisperによる文字起こし"""
    from faster_whisper import WhisperModel
 
    model = WhisperModel("base", device="cpu", compute_type="int8")
    segments, info = model.transcribe("sample_audio.mp3", language="ja")
 
    print(f"検出言語: {info.language} (確率: {info.language_probability:.2f})")
    for segment in segments:
        print(f"[{segment.start:.1f}s - {segment.end:.1f}s] {segment.text}")

# BAD: 単一のクラウドAPIに全依存
class BadAudioService:
    def transcribe(self, audio):
        # 1つのAPIだけに依存 → 障害時に全停止
        return openai_client.audio.transcriptions.create(
            model="whisper-1", file=audio
        )
 
# GOOD: フォールバック付きマルチプロバイダー
class GoodAudioService:
    def __init__(self):
        self.providers = [
            OpenAITranscriber(),
            GoogleTranscriber(),
            LocalWhisperTranscriber(),  # ローカルフォールバック
        ]
 
    def transcribe(self, audio):
        for provider in self.providers:
            try:
                return provider.transcribe(audio)
            except Exception as e:
                logger.warning(f"{provider.name} failed: {e}")
                continue
        raise AllProvidersFailedError("全プロバイダーが失敗")

# BAD: 生の音声をそのままモデルに入力
def bad_transcribe(raw_audio):
    return model.transcribe(raw_audio)  # ノイズ、無音区間がそのまま
 
# GOOD: 前処理パイプラインを通す
def good_transcribe(raw_audio):
    # Step 1: ノイズ除去
    cleaned = noise_reduction(raw_audio)
    # Step 2: 無音区間除去（VAD: Voice Activity Detection）
    segments = vad_segment(cleaned)
    # Step 3: 正規化（音量レベル調整）
    normalized = normalize_audio(segments, target_db=-20)
    # Step 4: リサンプリング（モデル要求のサンプルレートに合わせる）
    resampled = resample(normalized, target_sr=16000)
    return model.transcribe(resampled)

# BAD: 音声データのセキュリティを考慮しない
class BadAudioHandler:
    def process(self, audio_data):
        # 音声データをログに書き出し（個人情報漏洩リスク）
        logger.info(f"Processing audio: {audio_data[:100]}")
        # 平文でクラウドに送信
        requests.post("http://api.example.com/transcribe", data=audio_data)
        # 処理後のデータを削除しない
        save_to_disk(audio_data, "tmp/audio_cache/")
 
# GOOD: セキュリティを考慮した実装
class GoodAudioHandler:
    def process(self, audio_data):
        # ログには識別子のみ記録
        audio_id = generate_uuid()
        logger.info(f"Processing audio: id={audio_id}, size={len(audio_data)}B")
 
        try:
            # TLS暗号化通信で送信
            result = requests.post(
                "https://api.example.com/transcribe",
                data=audio_data,
                headers={"Authorization": f"Bearer {get_api_key()}"},
            )
 
            return result.json()
        finally:
            # 処理完了後、音声データを確実に削除
            del audio_data
            # ディスクキャッシュも削除
            cleanup_temp_files(audio_id)

# BAD: 各処理を直列実行してレイテンシが積み上がる
def bad_voice_assistant(audio):
    text = stt(audio)          # 1.5秒
    enhanced = llm(text)       # 2.0秒
    response = tts(enhanced)   # 1.5秒
    return response            # 合計: 5.0秒（会話として不自然）
 
# GOOD: ストリーミング + パイプライン並列化
async def good_voice_assistant(audio_stream):
    # STTストリーミング: 音声チャンクごとに認識
    async for text_chunk in stt_streaming(audio_stream):
        if text_chunk.is_final:
            # LLMストリーミング: トークンごとに出力
            async for response_token in llm_streaming(text_chunk.text):
                # TTSストリーミング: 文単位で音声合成
                if is_sentence_end(response_token):
                    sentence = buffer.flush()
                    async for audio_chunk in tts_streaming(sentence):
                        yield audio_chunk
    # 合計体感遅延: 0.5-1.0秒（ストリーミングにより大幅短縮）

# 音声AI開発でよく遭遇する問題と解決策
 
troubleshooting_guide = {
    "問題1: STTの精度が低い": {
        "症状": "認識結果に誤りが多い、ハルシネーションが発生",
        "原因と対策": [
            ("サンプルレート不一致", "モデルの要求するSR（通常16kHz）にリサンプリング"),
            ("ノイズ環境", "前処理でノイズ除去を行う。VADで非音声区間を除去"),
            ("モデルサイズ不足", "large-v3モデルに変更。faster-whisperで高速化"),
            ("言語指定なし", "language='ja'を明示的に指定"),
            ("プロンプトなし", "initial_promptに固有名詞リストを含める"),
        ],
    },
    "問題2: TTSの音声が不自然": {
        "症状": "イントネーションが変、読み間違いがある",
        "原因と対策": [
            ("テキスト前処理不足", "数字、略語、記号を読み仮名に変換"),
            ("文が長すぎる", "句読点で分割して合成"),
            ("モデル選択の問題", "用途に合ったモデル/ボイスに変更"),
            ("サンプリングパラメータ", "temperature/top_kを調整"),
        ],
    },
    "問題3: GPU メモリ不足": {
        "症状": "CUDA OOM、処理が途中で停止",
        "原因と対策": [
            ("モデルが大きすぎる", "量子化（INT8/FP16）を適用"),
            ("バッチサイズが大きい", "バッチサイズを縮小、チャンク分割処理"),
            ("メモリリーク", "torch.cuda.empty_cache()を適宜呼び出し"),
            ("複数モデルの同時ロード", "使用後にモデルをアンロード"),
        ],
    },
    "問題4: レイテンシが高い": {
        "症状": "応答に数秒以上かかる",
        "原因と対策": [
            ("バッチ処理", "ストリーミング処理に切り替え"),
            ("モデルサイズ", "小さいモデル or ONNX最適化を使用"),
            ("ネットワーク", "エッジ処理（ローカル実行）に切り替え"),
            ("直列処理", "パイプライン並列化を導入"),
        ],
    },
    "問題5: 日本語特有の問題": {
        "症状": "漢字の読み間違い、助詞の認識ミス",
        "原因と対策": [
            ("ファインチューニング不足", "ReazonSpeech等の日本語データで追加学習"),
            ("形態素解析の不足", "MeCab/Sudachiによる後処理"),
            ("固有名詞の未登録", "カスタム語彙/辞書の追加"),
            ("方言・口語体", "ドメイン特化データでの追加学習"),
        ],
    },
}

# 音声AIシステムのパフォーマンス最適化チェックリスト
 
performance_checklist = {
    "モデル最適化": [
        "[ ] 適切なモデルサイズの選択（用途に対して過大でないか）",
        "[ ] FP16/INT8 量子化の適用",
        "[ ] ONNX Runtime への変換",
        "[ ] TensorRT（NVIDIA GPU向け）の検討",
        "[ ] バッチ推論の活用（オフライン処理時）",
    ],
    "パイプライン最適化": [
        "[ ] ストリーミング処理の導入",
        "[ ] 非同期処理（async/await）の活用",
        "[ ] パイプラインの並列化（STT/LLM/TTS同時進行）",
        "[ ] 結果キャッシュの実装",
        "[ ] 不要な前処理ステップの省略",
    ],
    "インフラ最適化": [
        "[ ] GPUリソースの適切な割り当て",
        "[ ] モデルの事前ロード（コールドスタート回避）",
        "[ ] 接続プーリングの実装",
        "[ ] CDN/エッジキャッシュの活用（TTS結果）",
        "[ ] オートスケーリングの設定",
    ],
    "音声データ最適化": [
        "[ ] 適切なサンプルレートへの統一（16kHz for STT）",
        "[ ] VADによる非音声区間の除去",
        "[ ] 音声圧縮（Opus for WebRTC, FLAC for API）",
        "[ ] チャンク分割による段階的処理",
    ],
}

# 演習1: 基本実装のテンプレート
class Exercise1:
    """基本的な実装パターンの演習"""
 
    def __init__(self):
        self.data = []
 
    def validate_input(self, value):
        """入力値の検証"""
        if value is None:
            raise ValueError("入力値がNoneです")
        return True
 
    def process(self, value):
        """データ処理のメインロジック"""
        self.validate_input(value)
        self.data.append(value)
        return self.data
 
    def get_results(self):
        """処理結果の取得"""
        return {
            'count': len(self.data),
            'data': self.data
        }
 
# テスト
def test_exercise1():
    ex = Exercise1()
    assert ex.process(1) == [1]
    assert ex.process(2) == [1, 2]
    assert ex.get_results()['count'] == 2
 
    try:
        ex.process(None)
        assert False, "例外が発生するべき"
    except ValueError:
        pass
 
    print("全テスト合格!")
 
test_exercise1()

# 演習2: 応用パターン
from typing import List, Dict, Optional
from datetime import datetime
 
class AdvancedExercise:
    """応用パターンの演習"""
 
    def __init__(self, max_size: int = 100):
        self._items: List[Dict] = []
        self._max_size = max_size
        self._created_at = datetime.now()
 
    def add(self, key: str, value: any) -> bool:
        """アイテムの追加（サイズ制限付き）"""
        if len(self._items) >= self._max_size:
            return False
        self._items.append({
            'key': key,
            'value': value,
            'timestamp': datetime.now().isoformat()
        })
        return True
 
    def find(self, key: str) -> Optional[Dict]:
        """キーによる検索"""
        for item in reversed(self._items):
            if item['key'] == key:
                return item
        return None
 
    def remove(self, key: str) -> bool:
        """キーによる削除"""
        for i, item in enumerate(self._items):
            if item['key'] == key:
                self._items.pop(i)
                return True
        return False
 
    def stats(self) -> Dict:
        """統計情報"""
        return {
            'total_items': len(self._items),
            'max_size': self._max_size,
            'usage_percent': len(self._items) / self._max_size * 100,
            'uptime': str(datetime.now() - self._created_at)
        }
 
# テスト
def test_advanced():
    ex = AdvancedExercise(max_size=3)
    assert ex.add("a", 1) == True
    assert ex.add("b", 2) == True
    assert ex.add("c", 3) == True
    assert ex.add("d", 4) == False  # サイズ制限
    assert ex.find("b")['value'] == 2
    assert ex.remove("b") == True
    assert ex.find("b") is None
    stats = ex.stats()
    assert stats['total_items'] == 2
    print("応用テスト全合格!")
 
test_advanced()

# 演習3: パフォーマンス最適化
import time
from functools import lru_cache
 
# 最適化前（O(n^2)）
def slow_search(data: list, target: int) -> int:
    """非効率な検索"""
    for i in range(len(data)):
        for j in range(i + 1, len(data)):
            if data[i] + data[j] == target:
                return (i, j)
    return (-1, -1)
 
# 最適化後（O(n)）
def fast_search(data: list, target: int) -> tuple:
    """ハッシュマップを使った効率的な検索"""
    seen = {}
    for i, num in enumerate(data):
        complement = target - num
        if complement in seen:
            return (seen[complement], i)
        seen[num] = i
    return (-1, -1)
 
# ベンチマーク
def benchmark():
    import random
    data = list(range(5000))
    random.shuffle(data)
    target = data[100] + data[4000]
 
    start = time.time()
    result1 = slow_search(data, target)
    slow_time = time.time() - start
 
    start = time.time()
    result2 = fast_search(data, target)
    fast_time = time.time() - start
 
    print(f"非効率版: {slow_time:.4f}秒")
    print(f"効率版:   {fast_time:.6f}秒")
    print(f"高速化率: {slow_time/fast_time:.0f}倍")
 
benchmark()

# 設計判断の記録テンプレート
class ArchitectureDecisionRecord:
    """ADR (Architecture Decision Record) の作成"""
 
    def __init__(self, title: str):
        self.title = title
        self.context = ""
        self.decision = ""
        self.consequences = []
        self.alternatives = []
 
    def set_context(self, context: str):
        """背景と課題の記述"""
        self.context = context
        return self
 
    def set_decision(self, decision: str):
        """決定内容の記述"""
        self.decision = decision
        return self
 
    def add_consequence(self, consequence: str, positive: bool = True):
        """結果の追加"""
        self.consequences.append({
            'description': consequence,
            'type': 'positive' if positive else 'negative'
        })
        return self
 
    def add_alternative(self, name: str, reason_rejected: str):
        """却下した代替案の追加"""
        self.alternatives.append({
            'name': name,
            'reason_rejected': reason_rejected
        })
        return self
 
    def to_markdown(self) -> str:
        """Markdown形式で出力"""
        md = f"# ADR: {self.title}\n\n"
        md += f"## 背景\n{self.context}\n\n"
        md += f"## 決定\n{self.decision}\n\n"
        md += "## 結果\n"
        for c in self.consequences:
            icon = "✅" if c['type'] == 'positive' else "⚠️"
            md += f"- {icon} {c['description']}\n"
        md += "\n## 却下した代替案\n"
        for a in self.alternatives:
            md += f"- **{a['name']}**: {a['reason_rejected']}\n"
        return md

# チーム間のAPI契約定義
from dataclasses import dataclass
from typing import List, Optional
from enum import Enum
 
class Priority(Enum):
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"
 
@dataclass
class APIContract:
    """チーム間のAPI契約"""
    endpoint: str
    method: str
    owner_team: str
    consumers: List[str]
    sla_ms: int  # レスポンスタイムSLA
    priority: Priority
 
    def validate_sla(self, actual_ms: int) -> bool:
        """SLA準拠の確認"""
        return actual_ms <= self.sla_ms
 
    def to_openapi(self) -> dict:
        """OpenAPI形式で出力"""
        return {
            'path': self.endpoint,
            'method': self.method,
            'x-owner': self.owner_team,
            'x-consumers': self.consumers,
            'x-sla-ms': self.sla_ms
        }
 
# 使用例
contracts = [
    APIContract(
        endpoint="/api/v1/users",
        method="GET",
        owner_team="user-team",
        consumers=["order-team", "notification-team"],
        sla_ms=200,
        priority=Priority.HIGH
    ),
    APIContract(
        endpoint="/api/v1/orders",
        method="POST",
        owner_team="order-team",
        consumers=["payment-team", "inventory-team"],
        sla_ms=500,
        priority=Priority.CRITICAL
    )
]

# セキュアコーディング例
import hashlib
import secrets
import hmac
from typing import Optional
 
class SecurityUtils:
    """セキュリティユーティリティ"""
 
    @staticmethod
    def generate_token(length: int = 32) -> str:
        """暗号学的に安全なトークン生成"""
        return secrets.token_urlsafe(length)
 
    @staticmethod
    def hash_password(password: str, salt: Optional[str] = None) -> tuple:
        """パスワードのハッシュ化"""
        if salt is None:
            salt = secrets.token_hex(16)
        hashed = hashlib.pbkdf2_hmac(
            'sha256',
            password.encode('utf-8'),
            salt.encode('utf-8'),
            iterations=100000
        )
        return hashed.hex(), salt
 
    @staticmethod
    def verify_password(password: str, hashed: str, salt: str) -> bool:
        """パスワードの検証"""
        new_hash, _ = SecurityUtils.hash_password(password, salt)
        return hmac.compare_digest(new_hash, hashed)
 
    @staticmethod
    def sanitize_input(value: str) -> str:
        """入力値のサニタイズ"""
        dangerous_chars = ['<', '>', '"', "'", '&', '\\']
        result = value
        for char in dangerous_chars:
            result = result.replace(char, '')
        return result.strip()
 
# 使用例
token = SecurityUtils.generate_token()
hashed, salt = SecurityUtils.hash_password("my_password")
is_valid = SecurityUtils.verify_password("my_password", hashed, salt)

# マイグレーションスクリプトのテンプレート
import json
import logging
from pathlib import Path
from datetime import datetime
from typing import List, Dict, Callable
 
logger = logging.getLogger(__name__)
 
class MigrationRunner:
    """段階的マイグレーション実行エンジン"""
 
    def __init__(self, migration_dir: str):
        self.migration_dir = Path(migration_dir)
        self.migrations: List[Dict] = []
        self.completed: List[str] = []
 
    def register(self, version: str, description: str,
                 up: Callable, down: Callable):
        """マイグレーションの登録"""
        self.migrations.append({
            'version': version,
            'description': description,
            'up': up,
            'down': down,
            'registered_at': datetime.now().isoformat()
        })
 
    def run_up(self, target_version: str = None):
        """マイグレーションの実行（アップグレード）"""
        for migration in self.migrations:
            if migration['version'] in self.completed:
                continue
            logger.info(f"実行中: {migration['version']} - "
                       f"{migration['description']}")
            try:
                migration['up']()
                self.completed.append(migration['version'])
                logger.info(f"完了: {migration['version']}")
            except Exception as e:
                logger.error(f"失敗: {migration['version']}: {e}")
                raise
            if target_version and migration['version'] == target_version:
                break
 
    def run_down(self, target_version: str):
        """マイグレーションのロールバック"""
        for migration in reversed(self.migrations):
            if migration['version'] not in self.completed:
                continue
            if migration['version'] == target_version:
                break
            logger.info(f"ロールバック: {migration['version']}")
            migration['down']()
            self.completed.remove(migration['version'])
 
    def status(self) -> Dict:
        """マイグレーション状態の確認"""
        return {
            'total': len(self.migrations),
            'completed': len(self.completed),
            'pending': len(self.migrations) - len(self.completed),
            'versions': {
                m['version']: 'completed'
                if m['version'] in self.completed else 'pending'
                for m in self.migrations
            }
        }