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://github.com/opencv/opencv/blob/master/samples/data/vtest.aviを使用）
    - 出力: 処理結果をOpenCV画面でリアルタイム表示．OpenCV画面内に処理結果をテキストで表示．1秒間隔でprint()で処理結果を表示．プログラム終了時にprint()で表示した処理結果をresult.txtファイルに保存し，「result.txtに保存」したことをprint()で表示．プログラム開始時にプログラムの概要とユーザが行う必要がある操作をprint()で表示．
  - 処理手順: 1秒間隔でフレーム取得→PIL変換→LLaVA-NeXTによる多言語案内情報解析→結果表示→result.txt保存
  - 前処理、後処理: 前処理：BGR→RGB変換とPIL形式変換、後処理：日本語フォント適用とテキスト重畳表示
  - 追加処理: 動的高解像度画像処理（AnyRes）による認識精度向上、商業施設特化の案内情報抽出プロンプト適用
  - 調整を必要とする設定値: ANALYSIS_QUESTION（解析質問内容、商業施設の特性に応じた案内情報取得のためのプロンプト文字列）
- 将来方策: 商業施設の種類（駅、空港、ショッピングモール等）に応じた専用プロンプトテンプレートの動的切り替え機能
- その他の重要事項: CUDA対応GPU推奨、日本語フォント（msgothic.ttc）必須、transformers>=4.39.0必須
- 前準備: pip install -U torch torchvision torchaudio --index-url https://download.pytorch.org/whl/cu126
pip install transformers pillow requests opencv-python
"""

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

# 設定定数
MAX_NEW_TOKENS_MAIN = 300
MAX_NEW_TOKENS_SUB = 150
TEMPERATURE = 0.2
TOP_P = 0.9
USE_4BIT = False
ANALYSIS_INTERVAL = 1.0

# 軽量化モード用設定
LIGHTWEIGHT_MAX_NEW_TOKENS = 50
LIGHTWEIGHT_TEMPERATURE = 0.1
LIGHTWEIGHT_ANALYSIS_INTERVAL = 2.0

# 施設タイプごとのプロンプトテンプレート
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※必ず日本語で回答してください。'
}

# 言語検出パターン
LANGUAGE_PATTERNS = {
    'english': re.compile(r'[a-zA-Z]{3,}'),
    'japanese': re.compile(r'[\u3040-\u309f\u30a0-\u30ff\u4e00-\u9fff]+'),
}

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

FONT_PATH = 'C:/Windows/Fonts/msgothic.ttc'
FONT_SIZE_TIME = 30
FONT_SIZE_RESULT = 20

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8', line_buffering=True)

# 軽量化モードの選択
print('商業施設向けAI案内システム')
sys.stdout.flush()
print('軽量化モードを選択してください:')
sys.stdout.flush()
print('0: 通常モード（高精度）')
sys.stdout.flush()
print('1: 軽量化モード（エッジデバイス向け）')
sys.stdout.flush()

lightweight_mode = input('選択 (0/1): ').strip() == '1'

if lightweight_mode:
    print('軽量化モードで起動します')
    MAX_NEW_TOKENS_SUB = LIGHTWEIGHT_MAX_NEW_TOKENS
    TEMPERATURE = LIGHTWEIGHT_TEMPERATURE
    ANALYSIS_INTERVAL = LIGHTWEIGHT_ANALYSIS_INTERVAL
else:
    print('通常モードで起動します')

sys.stdout.flush()

print('このシステムは動画から看板、標識、商品、アイコン、ロゴを認識し、案内情報を提供します')
sys.stdout.flush()
print('操作方法:')
sys.stdout.flush()
print('- qキー: プログラム終了')
sys.stdout.flush()
print(f'- {ANALYSIS_INTERVAL}秒ごとに自動的に画像解析を実行します')
sys.stdout.flush()
print('操作: 動画選択後、自動で解析が開始されます')
sys.stdout.flush()

def load_model():
    try:
        model_id = 'llava-hf/llava-v1.6-mistral-7b-hf'

        model_kwargs = {
            'torch_dtype': torch.float16 if torch.cuda.is_available() else torch.float32,
            'device_map': 'auto' if torch.cuda.is_available() else None,
            'low_cpu_mem_usage': True
        }

        if USE_4BIT and torch.cuda.is_available():
            model_kwargs['load_in_4bit'] = True

        model = LlavaNextForConditionalGeneration.from_pretrained(model_id, **model_kwargs)

        if not torch.cuda.is_available():
            model.to('cpu')

        processor = LlavaNextProcessor.from_pretrained(model_id)
        return model, processor
    except Exception as e:
        print(f'モデルのロードに失敗しました: {e}')
        sys.stdout.flush()
        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 detect_languages(text):
    """テキストから検出された言語を判定"""
    detected_languages = []
    for lang, pattern in LANGUAGE_PATTERNS.items():
        if pattern.search(text):
            detected_languages.append(lang)
    return detected_languages

def clean_repetitive_text(text):
    """重複する行を削除する関数（完全一致のみ）"""
    lines = text.split('\n')
    cleaned_lines = []
    seen_lines = set()

    for line in lines:
        # 空白行や短い行は保持
        if len(line.strip()) < 5:
            cleaned_lines.append(line)
            continue

        # 完全一致する行のみスキップ
        line_stripped = line.strip()
        if line_stripped not in seen_lines:
            cleaned_lines.append(line)
            seen_lines.add(line_stripped)

    return '\n'.join(cleaned_lines)

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_SUB,
                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()

        # 重複テキストをクリーンアップ
        response = clean_repetitive_text(response)

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

        return response, facility_type

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

# 施設タイプの初期化
current_facility_type = 'general'
facility_type_detected = False

def analyze_frame(frame, model, processor):
    global current_facility_type, facility_type_detected

    image_rgb = cv2.cvtColor(frame, cv2.COLOR_BGR2RGB)
    pil_image = Image.fromarray(image_rgb)

    # 初回のみ施設タイプを自動検出
    if not facility_type_detected:
        # まず一般的なプロンプトで解析
        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'\n施設タイプを検出: {detected_type}')
            sys.stdout.flush()

    # 検出された施設タイプで解析
    response, _ = analyze_image(model, processor, pil_image, current_facility_type)

    # 検出された言語を表示
    detected_languages = detect_languages(response)
    if detected_languages:
        print(f'検出された言語: {", ".join(detected_languages)}')
        sys.stdout.flush()

    timestamp = time.strftime('%Y-%m-%d %H:%M:%S')
    print(f'\n[{timestamp}] 解析結果 (施設タイプ: {current_facility_type}):\n{response}')
    sys.stdout.flush()
    return timestamp, response

def video_processing(frame, results):
    display_frame = frame.copy()
    current_time = time.strftime('%Y-%m-%d %H:%M:%S')
    img_pil = Image.fromarray(cv2.cvtColor(display_frame, cv2.COLOR_BGR2RGB))
    draw = ImageDraw.Draw(img_pil)

    try:
        font_time = ImageFont.truetype(FONT_PATH, FONT_SIZE_TIME)
        font_result = ImageFont.truetype(FONT_PATH, FONT_SIZE_RESULT)
    except:
        font_time = ImageFont.load_default()
        font_result = ImageFont.load_default()

    draw.text((10, 10), current_time, font=font_time, fill=(0, 255, 0))

    if results:
        _, latest_result = results[-1]
        lines = latest_result.split('\n')
        y_offset = 50

        # 空行を除外し、実際の内容がある行のみ表示
        display_lines = [line for line in lines if line.strip()][:10]

        for i, line in enumerate(display_lines):
            if len(line) > 40:
                line = line[:40] + '...'
            draw.text((10, y_offset + i * 25), line, font=font_result, fill=(255, 255, 0))

    display_frame = cv2.cvtColor(np.array(img_pil), cv2.COLOR_RGB2BGR)
    return display_frame

model, processor = load_model()
results = []

print('0: 動画ファイル')
sys.stdout.flush()
print('1: カメラ')
sys.stdout.flush()
print('2: サンプル動画')
sys.stdout.flush()

choice = input('選択: ')
temp_file = None

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)
    cap.set(cv2.CAP_PROP_BUFFERSIZE, 1)
elif choice == '2':
    url = 'https://github.com/opencv/opencv/raw/master/samples/data/vtest.avi'
    filename = 'vtest.avi'
    try:
        urllib.request.urlretrieve(url, filename)
        temp_file = filename
        cap = cv2.VideoCapture(filename)
    except Exception as e:
        print(f'動画のダウンロードに失敗しました: {url}')
        sys.stdout.flush()
        print(f'エラー: {e}')
        sys.stdout.flush()
        exit()
else:
    print('無効な選択です')
    sys.stdout.flush()
    exit()

last_analysis_end_time = time.time()

try:
    while True:
        cap.grab()
        ret, frame = cap.retrieve()
        if not ret:
            break

        current_time = time.time()
        if current_time - last_analysis_end_time >= ANALYSIS_INTERVAL:
            print('認識中')
            sys.stdout.flush()
            timestamp, response = analyze_frame(frame, model, processor)
            results.append((timestamp, response))
            last_analysis_end_time = time.time()

        processed_frame = video_processing(frame, results)
        cv2.imshow('Video', processed_frame)

        key = cv2.waitKey(1) & 0xFF
        if key == ord('q'):
            break

finally:
    cap.release()
    cv2.destroyAllWindows()

    if results:
        with open('result.txt', 'w', encoding='utf-8') as f:
            for i, (timestamp, result) in enumerate(results):
                f.write(f'[{timestamp}] 解析結果 {i+1}:\n{result}\n\n')
        print('result.txtに保存')
        sys.stdout.flush()

    if temp_file and os.path.exists(temp_file):
        try:
            os.remove(temp_file)
        except:
            pass