LLaVA-NeXTによる商業施設向けAI案内（ソースコードと実行結果）

REM Python をシステム領域にインストール
winget install --scope machine --id Python.Python.3.12 -e --silent
REM Python のパス設定
set "PYTHON_PATH=C:\Program Files\Python312"
set "PYTHON_SCRIPTS_PATH=C:\Program Files\Python312\Scripts"
echo "%PATH%" | find /i "%PYTHON_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_PATH%" /M >nul
echo "%PATH%" | find /i "%PYTHON_SCRIPTS_PATH%" >nul
if errorlevel 1 setx PATH "%PATH%;%PYTHON_SCRIPTS_PATH%" /M >nul

winget install --scope machine Codeium.Windsurf -e --silent

REM CUDA をシステム領域にインストール
winget install --scope machine --id Nvidia.CUDA --version 12.6 -e
REM CUDA のパス設定
set "CUDA_PATH=C:\Program Files\NVIDIA GPU Computing Toolkit\CUDA\v12.6"
if exist "%CUDA_PATH%" setx CUDA_PATH "%CUDA_PATH%" /M >nul
if exist "%CUDA_PATH%" setx CUDNN_PATH "%CUDA_PATH%" /M >nul

"""
- プログラム名: LLaVA-NeXTによる商業施設向けAI案内プログラム
- 特徴技術名: LLaVA-NeXT（v1.6）
- 出典: Liu, H., Li, C., Li, Y., Li, B., Zhang, Y., Shen, S., & Lee, Y. J. (2024). LLaVA-NeXT: Improved reasoning, OCR, and world knowledge. Retrieved from https://llava-vl.github.io/blog/2024-01-30-llava-next/
- 特徴機能: 動的高解像度画像処理による視覚認識とOCR能力（従来の4倍の解像度で画像解析し、光学文字認識と視覚的推論を統合）
- 学習済みモデル: llava-hf/llava-v1.6-mistral-7b-hf（Mistral-7Bをベースとしたマルチモーダル言語モデル、商用利用可能、多言語対応）、URL: https://huggingface.co/llava-hf/llava-v1.6-mistral-7b-hf
- 方式設計
  - 関連利用技術:
    - OpenCV（リアルタイム画像・動画処理）
    - Transformers（深層学習モデル処理）
    - PIL（画像操作・テキスト描画）
    - tkinter（ファイル選択GUI）
  - 入力と出力:
    - 入力: 動画（ユーザは「0:動画ファイル，1:カメラ，2:サンプル動画」のメニューで選択．0:動画ファイルの場合はtkinterでファイル選択．1の場合はOpenCVでカメラが開く．2の場合はhttps://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.aviを使用）
    - 出力: 処理結果をOpenCV画面でリアルタイム表示．OpenCV画面内に処理結果をテキストで表示．各フレームごとにprint()で処理結果を表示．プログラム終了時にprint()で表示した処理結果をresult.txtファイルに保存し，「result.txtに保存」したことをprint()で表示．プログラム開始時にプログラムの概要とユーザが行う必要がある操作をprint()で表示．
  - 処理手順: フレーム取得→PIL変換→LLaVA-NeXTによる多言語案内情報解析→結果表示→result.txt保存
  - 前処理、後処理: 前処理：BGR→RGB変換とPIL形式変換、後処理：日本語フォント適用とテキスト重畳表示
  - 追加処理: 動的高解像度画像処理（AnyRes）による認識精度向上、商業施設特化の案内情報抽出プロンプト適用
  - 調整を必要とする設定値: ANALYSIS_QUESTION（解析質問内容、商業施設の特性に応じた案内情報取得のためのプロンプト文字列）
- 将来方策: 商業施設の種類（駅、空港、ショッピングモール等）に応じた専用プロンプトテンプレートの動的切り替え機能
- その他の重要事項: CUDA対応GPU推奨、日本語フォント（meiryo.ttc）必須、transformers>=4.39.0必須
- 前準備:
  pip install torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
  pip install transformers pillow opencv-python
"""

import cv2
import tkinter as tk
from tkinter import filedialog
import urllib.request
import torch
from transformers import LlavaNextProcessor, LlavaNextForConditionalGeneration
from PIL import Image, ImageDraw, ImageFont
import numpy as np
import time
from datetime import datetime

# 設定定数
MAX_NEW_TOKENS_MAIN = 300
TEMPERATURE = 0.2
TOP_P = 0.9

# 施設タイプごとのプロンプトテンプレート
FACILITY_PROMPTS = {
    'station': '駅構内の画像です。以下の項目を日本語で箇条書きで分析してください：\n・プラットフォーム番号、路線名、行き先\n・出口案内、階段、エスカレーター、エレベーターの位置\n・時刻表、運行情報、遅延情報\n・売店、トイレ、コインロッカーなどの施設\n・その他の案内表示や注意事項',
    'airport': '空港内の画像です。以下の項目を日本語で箇条書きで分析してください：\n・ゲート番号、搭乗口、チェックインカウンター\n・出発・到着情報、フライト情報\n・保安検査場、出入国審査、税関の位置\n・レストラン、免税店、ラウンジなどの施設\n・その他の案内表示や注意事項',
    'mall': 'ショッピングモール内の画像です。以下の項目を日本語で箇条書きで分析してください：\n・店舗名、ブランド名、営業時間\n・フロアガイド、エレベーター、エスカレーターの位置\n・レストラン、フードコート、トイレの位置\n・セール情報、イベント情報\n・その他の案内表示や注意事項',
    'general': 'この画像について以下の項目を日本語で箇条書きで詳しく分析してください：\n・場所の種類（屋内/屋外、建物の種類、具体的な場所名）\n・写っている物体や構造物の詳細\n・看板、案内図、標識、標示などに書かれている文字情報\n・時間帯、天候、季節などの状況\n・その他の特徴的で有用な情報\n※必ず日本語で回答してください。'
}

# 多言語プロンプト
MULTILINGUAL_PROMPT = '\n\n検出された英語の看板や表示がある場合は、その内容を日本語に翻訳して説明してください。'

print('\n=== LLaVA-NeXT商業施設向けAI案内システム ===')
print('このシステムは動画から看板、標識、商品、アイコン、ロゴを認識し、案内情報を提供します')
print('操作方法:')
print('  q キー: プログラム終了')
print('注意事項:')
print('  - GPU推奨（CPUでも動作可能ですが処理が遅くなります）')
print('  - 初回実行時はモデルのダウンロードに時間がかかります')

def load_model():
    """LLaVA-NeXTモデルとプロセッサをロード（GPUメモリ不足時はCPUにフォールバック）"""
    try:
        print('\nモデルをロード中...')
        model_id = 'llava-hf/llava-v1.6-mistral-7b-hf'

        # まずGPUでの読み込みを試みる
        if torch.cuda.is_available():
            try:
                print('GPUでのモデルロードを試みています...')
                model_kwargs = {
                    'torch_dtype': torch.float16,
                    'device_map': 'auto',
                    'low_cpu_mem_usage': True
                }
                model = LlavaNextForConditionalGeneration.from_pretrained(model_id, **model_kwargs)
                device = torch.device('cuda')
                model.eval()
                print(f'GPUでのモデルロード成功（デバイス: {device}）')

            except (RuntimeError, torch.cuda.OutOfMemoryError) as e:
                print(f'\nGPUメモリ不足のため、CPUモードに切り替えます')
                print(f'エラー詳細: {str(e)[:100]}...')
                torch.cuda.empty_cache()  # GPUメモリをクリア

                # CPUで再ロード
                model_kwargs = {
                    'torch_dtype': torch.float32,
                    'device_map': None,
                    'low_cpu_mem_usage': True
                }
                model = LlavaNextForConditionalGeneration.from_pretrained(model_id, **model_kwargs)
                device = torch.device('cpu')
                model.to(device)
                model.eval()
                print(f'CPUでのモデルロード成功（デバイス: {device}）')
        else:
            # GPUが利用不可の場合は最初からCPU
            print('GPUが利用できません。CPUモードで起動します')
            model_kwargs = {
                'torch_dtype': torch.float32,
                'device_map': None,
                'low_cpu_mem_usage': True
            }
            model = LlavaNextForConditionalGeneration.from_pretrained(model_id, **model_kwargs)
            device = torch.device('cpu')
            model.to(device)
            model.eval()
            print(f'CPUでのモデルロード完了（デバイス: {device}）')

        processor = LlavaNextProcessor.from_pretrained(model_id)
        print('モデルのロード完了')
        return model, processor, device

    except Exception as e:
        print(f'モデルのロードに失敗しました: {e}')
        print('transformers ライブラリをインストールしてください:')
        print('pip install transformers')
        exit()

def detect_facility_type(text):
    """画像解析結果から施設タイプを推定"""
    text_lower = text.lower()

    # 駅の特徴的なキーワード
    if any(keyword in text for keyword in ['プラットフォーム', '路線', '時刻表', '改札', '乗り場', 'platform', 'railway', 'train']):
        return 'station'
    # 空港の特徴的なキーワード
    elif any(keyword in text for keyword in ['ゲート', '搭乗', 'フライト', '出発', '到着', 'gate', 'flight', 'departure', 'arrival']):
        return 'airport'
    # ショッピングモールの特徴的なキーワード
    elif any(keyword in text for keyword in ['店舗', 'ショップ', 'フロア', 'セール', 'レストラン', 'shop', 'store', 'floor', 'sale']):
        return 'mall'

    return 'general'

def analyze_image(model, processor, image, facility_type='general'):
    """画像を解析して案内情報を抽出"""
    try:
        # 施設タイプに応じたプロンプトを選択
        base_prompt = FACILITY_PROMPTS.get(facility_type, FACILITY_PROMPTS['general'])
        prompt = base_prompt + MULTILINGUAL_PROMPT

        conversation = [
            {
                'role': 'user',
                'content': [
                    {'type': 'image'},
                    {'type': 'text', 'text': prompt},
                ],
            },
        ]

        prompt = processor.apply_chat_template(conversation, add_generation_prompt=True)
        inputs = processor(images=image, text=prompt, return_tensors='pt')

        if torch.cuda.is_available():
            inputs = {k: v.to('cuda') for k, v in inputs.items()}

        with torch.inference_mode():
            output = model.generate(
                **inputs,
                max_new_tokens=MAX_NEW_TOKENS_MAIN,
                do_sample=True,
                temperature=TEMPERATURE,
                top_p=TOP_P,
                pad_token_id=processor.tokenizer.eos_token_id,
                eos_token_id=processor.tokenizer.eos_token_id,
                repetition_penalty=1.2
            )

        response = processor.decode(output[0], skip_special_tokens=True)

        # LLaVA-NeXTの応答形式に対応
        if '[/INST]' in response:
            response = response.split('[/INST]')[-1].strip()
        elif 'ASSISTANT:' in response:
            response = response.split('ASSISTANT:')[-1].strip()

        # 結果が空または短い場合のフォールバック
        if len(response.strip()) < 10:
            return "画像の解析に失敗しました。"

        # 重複行の削除
        lines = response.split('\n')
        seen = set()
        cleaned = []
        for line in lines:
            if line.strip() and line.strip() not in seen:
                cleaned.append(line)
                seen.add(line.strip())

        return '\n'.join(cleaned)

    except Exception as e:
        return f'画像解析エラー: {e}'

# モデルをロード（deviceも返り値に含める）
model, processor, device = load_model()
print(f'デバイス: {str(device)}')

# GPU使用時の最適化
if device.type == 'cuda':
    torch.backends.cudnn.benchmark = True

# グローバル変数
frame_count = 0
results_log = []
current_facility_type = 'general'
facility_type_detected = False

def video_frame_processing(frame):
    """動画フレームを処理"""
    global frame_count, current_facility_type, facility_type_detected
    current_time = time.time()
    frame_count += 1

    # BGR→RGB変換とPIL形式変換
    image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    pil_image = Image.fromarray(image_rgb)

    # 初回のみ施設タイプを自動検出
    if not facility_type_detected and frame_count == 1:
        initial_response = analyze_image(model, processor, pil_image, 'general')
        detected_type = detect_facility_type(initial_response)
        if detected_type != 'general':
            current_facility_type = detected_type
            facility_type_detected = True
            print(f'施設タイプを検出: {detected_type}')

    # 画像解析
    result = analyze_image(model, processor, pil_image, current_facility_type)

    # 結果を画面に表示
    processed_frame = frame.copy()
    FONT_PATH = 'C:/Windows/Fonts/meiryo.ttc'
    FONT_SIZE = 20

    try:
        font = ImageFont.truetype(FONT_PATH, FONT_SIZE)
    except:
        font = ImageFont.load_default()

    img_pil = Image.fromarray(cv2.cvtColor(processed_frame, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)

    # 解析結果の最初の3行を表示
    lines = result.split('\n')[:3]
    for i, line in enumerate(lines):
        if len(line) > 40:
            line = line[:40] + '...'
        draw.text((10, 30 + i * 25), line, font=font, fill=(0, 255, 0))

    processed_frame = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)

    # 結果を簡潔に整形
    result_summary = result.replace('\n', ' ')[:100] + '...' if len(result) > 100 else result.replace('\n', ' ')

    return processed_frame, result_summary, current_time

print("\n0: 動画ファイル")
print("1: カメラ")
print("2: サンプル動画")

choice = input("選択: ")

if choice == '0':
    root = tk.Tk()
    root.withdraw()
    path = filedialog.askopenfilename()
    if not path:
        exit()
    cap = cv2.VideoCapture(path)
elif choice == '1':
    cap = cv2.VideoCapture(0, cv2.CAP_DSHOW)
    if not cap.isOpened():
        cap = cv2.VideoCapture(0)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
else:
    # サンプル動画ダウンロード・処理
    SAMPLE_URL = 'https://raw.githubusercontent.com/opencv/opencv/master/samples/data/vtest.avi'
    SAMPLE_FILE = 'vtest.avi'
    urllib.request.urlretrieve(SAMPLE_URL, SAMPLE_FILE)
    cap = cv2.VideoCapture(SAMPLE_FILE)

if not cap.isOpened():
    print('動画ファイル・カメラを開けませんでした')
    exit()

# メイン処理
print('\n=== 動画処理開始 ===')
print('操作方法:')
print('  q キー: プログラム終了')
try:
    while True:
        ret, frame = cap.read()
        if not ret:
            break

        MAIN_FUNC_DESC = "商業施設AI案内"
        processed_frame, result, current_time = video_frame_processing(frame)
        cv2.imshow(MAIN_FUNC_DESC, processed_frame)
        if choice == '1':  # カメラの場合
            print(datetime.fromtimestamp(current_time).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3], result)
        else:  # 動画ファイルの場合
            print(frame_count, result)
        results_log.append(result)
        if cv2.waitKey(1) & 0xFF == ord('q'):
            break
finally:
    print('\n=== プログラム終了 ===')
    cap.release()
    cv2.destroyAllWindows()
    if results_log:
        with open('result.txt', 'w', encoding='utf-8') as f:
            f.write('=== 結果 ===\n')
            f.write(f'処理フレーム数: {frame_count}\n')
            f.write(f'使用デバイス: {str(device).upper()}\n')
            if device.type == 'cuda':
                f.write(f'GPU: {torch.cuda.get_device_name(0)}\n')
            f.write('\n')
            f.write('\n'.join(results_log))
        print(f'\n処理結果をresult.txtに保存しました')