SaliencyMapを使って視線誘導したい

はじめに

Saliencyとは

• サリエンシー（Saliency） ＝「顕著性」「目立ちやすさ」
• 画像や映像において、人間の注意を引きつけやすい特徴や領域を指す

サリエンシーマップとは

• 画像や映像の中で、各領域がどの程度目立つかを視覚的に表現したもの
  • 人間の視覚的注意を予測するモデルとして使用される

サリエンシーマップの主な用途

• マーケティング：広告の効果を予測・分析
• UI設計：重要な要素が目立つようにデザインを調整
• コンピュータビジョン：物体検出や画像分割のための前処理として使用

本題

やりたいこと

• 意図した情報やメッセージを効果的に伝えるための分析を簡単に行いたい
• サリエンシーマップを使うことで、自然に目が引きつけられる領域を特定し、それに基づいて効果的な視線誘導ができているかを確認したい

つくってみる

• 画像、動画、PDFをインプットに処理できるようにした
  • 処理結果として、サリエンシーマップを重ねた画像、動画がzipファイルでDLできるようにする
• tkinterを使ってGUIで動かせるようにしてみた

動作イメージ

実行コード

# 必要なライブラリのインストール
pip install opencv-python-headless   # OpenCV
pip install numpy                    # NumPy
pip install pillow                   # Pillow
pip install pdf2image                # PDF to image変換
pip install imageio                  # ImageIO

brew install poppler

Saliency_Map_GUI.py

import tkinter as tk
from tkinter import ttk, filedialog, messagebox
import cv2
import numpy as np
from PIL import Image, ImageTk
import os
import zipfile
from pdf2image import convert_from_path
import imageio
import tempfile
from datetime import datetime

class SaliencyApp:
    def __init__(self, root):
        self.root = root
        self.root.title("Saliency Map Generator")
        
        # メインフレーム
        self.main_frame = ttk.Frame(root, padding="10")
        self.main_frame.grid(row=0, column=0, sticky=(tk.W, tk.E, tk.N, tk.S))
        
        # プレビューキャンバス
        self.canvas = tk.Canvas(self.main_frame, width=600, height=400)
        self.canvas.grid(row=0, column=0, columnspan=3, pady=5)
        
        # ボタン
        ttk.Button(self.main_frame, text="画像を処理", command=self.process_image).grid(row=1, column=0, pady=5, padx=5)
        ttk.Button(self.main_frame, text="動画を処理", command=self.process_video).grid(row=1, column=1, pady=5, padx=5)
        ttk.Button(self.main_frame, text="PDFを処理", command=self.process_pdf).grid(row=1, column=2, pady=5, padx=5)
        
        # ステータスバー
        self.status_var = tk.StringVar()
        self.status_var.set("準備完了")
        ttk.Label(self.main_frame, textvariable=self.status_var).grid(row=2, column=0, columnspan=3, pady=5)
        
        # 一時ファイル用のディレクトリを作成
        self.temp_dir = tempfile.mkdtemp()

    def __del__(self):
        # 終了時に一時ディレクトリを削除
        try:
            import shutil
            shutil.rmtree(self.temp_dir)
        except:
            pass

    def __saliency(self, src):
        saliency = cv2.saliency.StaticSaliencySpectralResidual_create()
        (success, saliency_map) = saliency.computeSaliency(src)
        if not success:
            return (False, None)
        saliency_map = (saliency_map * 255).astype("uint8")
        heatmap = cv2.applyColorMap(saliency_map, cv2.COLORMAP_JET)
        weight = cv2.addWeighted(src, 0.7, heatmap, 0.5, 1.0)
        return (True, weight)

    def update_preview(self, cv_image):
        # OpenCV画像をPIL形式に変換
        image = cv2.cvtColor(cv_image, cv2.COLOR_BGR2RGB)
        image = Image.fromarray(image)
        
        # キャンバスのサイズに合わせてリサイズ
        canvas_width = self.canvas.winfo_width()
        canvas_height = self.canvas.winfo_height()
        image.thumbnail((canvas_width, canvas_height), Image.Resampling.LANCZOS)
        
        # PhotoImageに変換してキャンバスに表示
        photo = ImageTk.PhotoImage(image)
        self.canvas.create_image(canvas_width//2, canvas_height//2, image=photo, anchor="center")
        self.canvas.image = photo

    def create_zip(self, files, zip_name):
        timestamp = datetime.now().strftime("%Y%m%d_%H%M%S")
        zip_path = filedialog.asksaveasfilename(
            defaultextension=".zip",
            initialfile=f"{zip_name}_{timestamp}.zip",
            filetypes=[("ZIP files", "*.zip")]
        )
        if not zip_path:
            return None
            
        with zipfile.ZipFile(zip_path, 'w') as zipf:
            for file in files:
                if os.path.exists(file):
                    zipf.write(file, os.path.basename(file))
                    os.remove(file)  # 一時ファイルを削除
        
        return zip_path

    def process_image(self):
        file_paths = filedialog.askopenfilenames(
            filetypes=[("Image files", "*.jpg *.jpeg *.png *.bmp *.gif")]
        )
        if not file_paths:
            return
            
        processed_files = []
        for file_path in file_paths:
            self.status_var.set(f"画像を処理中... {os.path.basename(file_path)}")
            self.root.update()
            
            try:
                # 画像を読み込み
                image = cv2.imread(file_path)
                if image is None:
                    raise ValueError(f"画像の読み込みに失敗しました: {file_path}")
                
                # サリエンシーマップを適用
                success, result = self.__saliency(image)
                if not success:
                    raise ValueError(f"サリエンシーマップの生成に失敗しました: {file_path}")
                
                # プレビューを更新
                self.update_preview(result)
                
                # 一時ファイルとして保存
                temp_file = os.path.join(self.temp_dir, f"saliency_{os.path.basename(file_path)}")
                cv2.imwrite(temp_file, result)
                processed_files.append(temp_file)
                
            except Exception as e:
                messagebox.showwarning("警告", str(e))
        
        if processed_files:
            zip_path = self.create_zip(processed_files, "image_results")
            if zip_path:
                self.status_var.set(f"保存完了: {os.path.basename(zip_path)}")
            else:
                self.status_var.set("保存がキャンセルされました")
        else:
            self.status_var.set("処理可能な画像がありませんでした")
            
        self.root.update()

    def process_video(self):
        file_path = filedialog.askopenfilename(
            filetypes=[("Video files", "*.mp4 *.avi *.mov")]
        )
        if not file_path:
            return
            
        self.status_var.set("動画を処理中...")
        self.root.update()
        
        try:
            # 動画を読み込み
            video = cv2.VideoCapture(file_path)
            if not video.isOpened():
                raise ValueError("動画の読み込みに失敗しました")
            
            # 一時ファイルとして保存
            temp_file = os.path.join(self.temp_dir, f"saliency_{os.path.basename(file_path)}")
            
            # 出力用のビデオライターを準備
            fourcc = cv2.VideoWriter_fourcc(*'mp4v')
            fps = video.get(cv2.CAP_PROP_FPS)
            width = int(video.get(cv2.CAP_PROP_FRAME_WIDTH))
            height = int(video.get(cv2.CAP_PROP_FRAME_HEIGHT))
            writer = cv2.VideoWriter(temp_file, fourcc, fps, (width, height))
            
            frame_count = 0
            while True:
                ret, frame = video.read()
                if not ret:
                    break
                    
                # サリエンシーマップを適用
                success, result = self.__saliency(frame)
                if success:
                    writer.write(result)
                    if frame_count % 30 == 0:  # 30フレームごとにプレビューを更新
                        self.update_preview(result)
                        self.status_var.set(f"フレーム {frame_count} を処理中...")
                        self.root.update()
                frame_count += 1
            
            video.release()
            writer.release()
            
            # ZIPファイルとして保存
            zip_path = self.create_zip([temp_file], "video_results")
            if zip_path:
                self.status_var.set(f"保存完了: {os.path.basename(zip_path)}")
            else:
                self.status_var.set("保存がキャンセルされました")
            
        except Exception as e:
            messagebox.showerror("エラー", str(e))
            self.status_var.set("エラーが発生しました")
            
        self.root.update()

    def process_pdf(self):
        file_path = filedialog.askopenfilename(
            filetypes=[("PDF files", "*.pdf")]
        )
        if not file_path:
            return
            
        self.status_var.set("PDFを処理中...")
        self.root.update()
        
        try:
            # PDFを画像に変換
            pages = convert_from_path(file_path)
            processed_files = []
            
            for i, page in enumerate(pages):
                # PIL ImageをOpenCV形式に変換
                page_cv = cv2.cvtColor(np.array(page), cv2.COLOR_RGB2BGR)
                
                # サリエンシーマップを適用
                success, result = self.__saliency(page_cv)
                if success:
                    # プレビューを更新
                    self.update_preview(result)
                    self.status_var.set(f"ページ {i+1}/{len(pages)} を処理中...")
                    self.root.update()
                    
                    # 一時ファイルとして保存
                    temp_file = os.path.join(self.temp_dir, f"saliency_page_{i+1}.jpg")
                    cv2.imwrite(temp_file, result)
                    processed_files.append(temp_file)
            
            if processed_files:
                zip_path = self.create_zip(processed_files, "pdf_results")
                if zip_path:
                    self.status_var.set(f"保存完了: {os.path.basename(zip_path)}")
                else:
                    self.status_var.set("保存がキャンセルされました")
            else:
                self.status_var.set("処理可能なページがありませんでした")
            
        except Exception as e:
            messagebox.showerror("エラー", str(e))
            self.status_var.set("エラーが発生しました")
            
        self.root.update()

def main():
    root = tk.Tk()
    app = SaliencyApp(root)
    root.mainloop()

if __name__ == "__main__":
    main()