Update app.py

import gradio as gr
import torch
import random
import numpy as np
import datetime

# 履歴保存
from huggingface_hub import HfApi
from huggingface_hub import login

import os
# HF_TOKEN 環境変数からトークンを明示的に読み込む
hf_token_value = os.getenv("HF_TOKEN")

if hf_token_value:
api = HfApi(token=hf_token_value)
print("token ok.")
else:
# トークンが設定されていない場合の警告と代替処理
print("HF_TOKEN error")
api = HfApi() # トークンなしで初期化

# 画像をアップロードするリポジトリID
HF_REPO_ID = "cocoat/images"

# 公開用画像をアップロードするリポジトリID
PUBLIC_REPO_ID = "cocoat/opendata"

# Space内で画像を保存するディレクトリ
SPACE_IMAGE_DIR = "generated_images"
os.makedirs(SPACE_IMAGE_DIR, exist_ok=True)

# 公開リポジトリの画像ディレクトリ
PUBLIC_IMAGE_DIR = "generated_images"
os.makedirs(PUBLIC_IMAGE_DIR, exist_ok=True)

# 履歴ファイルを定義
HISTORY_FILE = "history/generation_history_coamixXL3.txt"

# 履歴をロードする関数
import os
import requests
def load_history():
history_data = []
hf_raw_file_url = f"https://huggingface.co/datasets/{HF_REPO_ID}/raw/main/{HISTORY_FILE}"
headers = {}
if hf_token_value:
headers["Authorization"] = f"Bearer {hf_token_value}"

try:
response = requests.get(hf_raw_file_url, headers=headers)
response.raise_for_status()

loaded_hub_paths = set() # 重複ロードを防ぐため

for line in response.text.splitlines():
parts = line.strip().split("|||")
if len(parts) == 2:
image_path_in_repo = parts[0]
caption = parts[1]
# 公開リポジトリの画像URLを生成
hub_image_url = f"https://huggingface.co/datasets/{PUBLIC_REPO_ID}/resolve/main/{image_path_in_repo}"

history_data.append((image_path_in_repo, caption, hub_image_url))
loaded_hub_paths.add(image_path_in_repo)
print(f"History loaded from Hub and matched with Space images: {len(history_data)} entries.")
except requests.exceptions.RequestException as e:
print(f"Error loading history from Hub via raw URL: {e}. Starting with empty history.")
except Exception as e:
print(f"An unexpected error occurred while parsing history: {e}. Starting with empty history.")

return history_data[:10]


# 履歴を初期化時にロード (修正された load_history を使用)
history = load_history()


# 履歴を初期化時にロード
history = load_history()

from PIL import Image
from diffusers import (
StableDiffusionXLPipeline,
EulerAncestralDiscreteScheduler,
DPMSolverMultistepScheduler
)
from huggingface_hub import hf_hub_download, HfApi

# デバイスと型の設定
device = "cuda" if torch.cuda.is_available() else "cpu"
torch_dtype = torch.float16 if torch.cuda.is_available() else torch.float32
MAX_SEED = np.iinfo(np.int32).max
MAX_SIZE = 2048

# モデルファイルのダウンロード
model_path = hf_hub_download(
repo_id="cocoat/cocoamix",
filename="recocoamixXL3_coamixXL3.safetensors"
)

# パイプライン構築
pipe = StableDiffusionXLPipeline.from_single_file(
model_path,
torch_dtype=torch_dtype,
use_safetensors=True
).to(device)

# スケジューラ設定
euler_scheduler = EulerAncestralDiscreteScheduler.from_config(
pipe.scheduler.config,
use_karras_sigmas=True
)
dpm_scheduler = DPMSolverMultistepScheduler.from_config(pipe.scheduler.config)
pipe.scheduler = euler_scheduler

def upload_image_to_hub(image_pil, prompt_text):
# ファイル名を生成（タイムスタンプとプロンプトの一部）
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
# プロンプトから安全なファイル名の一部を生成
# safe_prompt = "".join(c for c in prompt_text if c.isalnum() or c in (' ', '.', '_')).replace(' ', '_')[:30]
filename = f"image_{timestamp}.png"
filepath = f"temp_{filename}"
image_pil.save(filepath)

# Hubにアップロード
try:
# リポジトリ内にディレクトリを作成する場合は path_in_repo を使う
path_in_repo = f"generated_images/{filename}"
api.upload_file(
path_or_fileobj=filepath,
path_in_repo=path_in_repo,
repo_id=PUBLIC_REPO_ID, # 公開用リポジトリに変更
repo_type="dataset",
)

# アップロードされたファイルのURLを構築する（PUBLIC_REPO_IDを使用）
uploaded_file_url = f"https://huggingface.co/datasets/{PUBLIC_REPO_ID}/resolve/main/{path_in_repo}"
print(f"Uploaded {filepath} to {uploaded_file_url}")

# 公開リポジトリの古いファイルを削除するロジック
current_files = api.list_repo_files(repo_id=PUBLIC_REPO_ID, repo_type="dataset")
# PUBLIC_IMAGE_DIR (generated_images) 以下のpngファイルを抽出し、新しいものからソート
generated_images_in_public = sorted([f for f in current_files if f.startswith(PUBLIC_IMAGE_DIR) and f.endswith('.png')], reverse=True)

# 10枚を超える場合、古いファイルを削除
if len(generated_images_in_public) > 10:
files_to_delete = generated_images_in_public[10:]
for file_to_delete in files_to_delete:
try:
api.delete_file(
path_in_repo=file_to_delete,
repo_id=PUBLIC_REPO_ID,
repo_type="dataset",
commit_message=f"Delete old image: {file_to_delete}"
)
print(f"Deleted old public image: {file_to_delete}")
except Exception as del_e:
print(f"Error deleting old public image {file_to_delete}: {del_e}")
return uploaded_file_url, path_in_repo
except Exception as e:
print(f"Error uploading image to Hub: {e}")
return None, None
finally:
# 一時ファイルを削除
if os.path.exists(filepath):
os.remove(filepath)
print(f"一時画像ファイル {filepath} を削除しました。")

def upload_image_to_private_hub(image_pil, prompt_text):
timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
filename = f"image_{timestamp}.png"
filepath = f"temp_private_{filename}" # 一時ファイル名が重複しないように private を追加
image_pil.save(filepath)

try:
path_in_repo = f"generated_images/{filename}"
api.upload_file(
path_or_fileobj=filepath,
path_in_repo=path_in_repo,
repo_id=HF_REPO_ID, # 非公開リポジトリ
repo_type="dataset",
)
print(f"Uploaded {filepath} to private Hub: {path_in_repo}")
return path_in_repo # 履歴ファイルに記録するリポジトリ内パスを返す
except Exception as e:
print(f"Error uploading image to private Hub: {e}")
return None
finally:
if os.path.exists(filepath):
os.remove(filepath)
print(f"一時プライベート画像ファイル {filepath} を削除しました。")

def make_html_table(caption):
formatted_caption = caption.replace("|-|", "\n")
rows = formatted_caption.split("\n")
html = '<table style="width:100%;border-collapse:collapse;background:#fffaf1;color:#000">'
for row in rows:
if ": " in row:
key, val = row.split(": ", 1)
html += (
# f'{key}: {val}\n'
f'<tr><th style="text-align:left;border:1px solid #ddd;padding:4px;">{key}</th>'
f'<td style="border:1px solid #ddd;padding:4px;">{val}</td></tr>'
)
html += '</table>'
return html

def create_dummy_image(width=512, height=512, alpha=0):
return Image.new("RGBA", (width, height), (0, 0, 0, alpha))

def update_history_tables_on_select(evt: gr.SelectData):
if evt.index is not None and 0 <= evt.index < len(history):
selected_caption = history[evt.index][1]
return make_html_table(selected_caption)
return ""

def update_history():
tables_html = "".join(
f'<div style="margin-bottom:12px">{make_html_table(item[1])}</div>'
for item in history
)
return tables_html

def infer(prompt, neg, seed, rand, w, h, cfg, steps, scheduler_type,
progress=gr.Progress(track_tqdm=True)):
if rand:
seed = random.randint(0, MAX_SEED)
generator = torch.Generator(device=device).manual_seed(seed)

pipe.scheduler = euler_scheduler if scheduler_type == "Euler Ancestral" else dpm_scheduler
pipe.scheduler.set_timesteps(steps)

def _callback(pipeline, step_idx, timestep, callback_kwargs):
progress(step_idx / steps, desc=f"Step {step_idx}/{steps}")
return callback_kwargs

output = pipe(
prompt=prompt,
negative_prompt=neg or None,
guidance_scale=cfg,
num_inference_steps=steps,
width=w,
height=h,
generator=generator,
callback_on_step_end=_callback
)
img = output.images[0]

caption_text = (
f"Prompt: {prompt}\n"
f"Negative: {neg or 'None'}\n"
f"Seed: {seed}\n"
f"Size: {w}×{h}\n"
f"CFG: {cfg}\n"
f"Steps: {steps}\n"
f"Scheduler: {scheduler_type}"
)

caption_text_for_history = caption_text.replace("\n", "|-|").strip()
# 画像をHubにアップロードし、そのURLとリポジトリ内パスを取得
# 公開用リポジトリにアップロード
uploaded_image_url, path_in_public_repo_for_history = upload_image_to_hub(img, prompt)
# 非公開リポジトリにアップロード
path_in_private_repo_for_history = upload_image_to_private_hub(img, prompt)

# 履歴を更新
global history
# Hubへのアップロードが成功した場合のみ履歴に追加
# historyリストには (非公開リポジトリのパス, キャプション, 公開リポジトリのURL)

app.py

@@ -71,11 +71,6 @@ def load_history():
 
     return history_data[:10]
 
-
-# 履歴を初期化時にロード (修正された load_history を使用)
-history = load_history()
-
-
 # 履歴を初期化時にロード
 history = load_history()
 
@@ -162,10 +157,6 @@ def upload_image_to_hub(image_pil, prompt_text):
         print(f"Error uploading image to Hub: {e}")
         return None, None
     finally:
-        # 一時ファイルを削除
-        if os.path.exists(filepath):
-            os.remove(filepath)
-            print(f"一時画像ファイル {filepath} を削除しました。")
 
 def upload_image_to_private_hub(image_pil, prompt_text):
     timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
@@ -187,9 +178,6 @@ def upload_image_to_private_hub(image_pil, prompt_text):
         print(f"Error uploading image to private Hub: {e}")
         return None
     finally:
-        if os.path.exists(filepath):
-            os.remove(filepath)
-            print(f"一時プライベート画像ファイル {filepath} を削除しました。")
 
 def make_html_table(caption):
     formatted_caption = caption.replace("|-|", "\n")
@@ -224,88 +212,103 @@ def update_history():
 
 def infer(prompt, neg, seed, rand, w, h, cfg, steps, scheduler_type,
           progress=gr.Progress(track_tqdm=True)):
-    if rand:
-        seed = random.randint(0, MAX_SEED)
-    generator = torch.Generator(device=device).manual_seed(seed)
-
-    pipe.scheduler = euler_scheduler if scheduler_type == "Euler Ancestral" else dpm_scheduler
-    pipe.scheduler.set_timesteps(steps)
-
-    def _callback(pipeline, step_idx, timestep, callback_kwargs):
-        progress(step_idx / steps, desc=f"Step {step_idx}/{steps}")
-        return callback_kwargs
-
-    output = pipe(
-        prompt=prompt,
-        negative_prompt=neg or None,
-        guidance_scale=cfg,
-        num_inference_steps=steps,
-        width=w,
-        height=h,
-        generator=generator,
-        callback_on_step_end=_callback
-    )
-    img = output.images[0]
-
-    caption_text = (
-        f"Prompt: {prompt}\n"
-        f"Negative: {neg or 'None'}\n"
-        f"Seed: {seed}\n"
-        f"Size: {w}×{h}\n"
-        f"CFG: {cfg}\n"
-        f"Steps: {steps}\n"
-        f"Scheduler: {scheduler_type}"
-    )
-
-    caption_text_for_history = caption_text.replace("\n", "|-|").strip()
-    # 画像をHubにアップロードし、そのURLとリポジトリ内パスを取得
-    # 公開用リポジトリにアップロード
-    uploaded_image_url, path_in_public_repo_for_history = upload_image_to_hub(img, prompt)
-    # 非公開リポジトリにアップロード
-    path_in_private_repo_for_history = upload_image_to_private_hub(img, prompt)
-
-    # 履歴を更新
-    global history
-    # Hubへのアップロードが成功した場合のみ履歴に追加
-    # historyリストには (非公開リポジトリのパス, キャプション, 公開リポジトリのURL) の形式で保存
-    if path_in_private_repo_for_history and uploaded_image_url:
-        history.insert(0, (path_in_private_repo_for_history, caption_text_for_history, uploaded_image_url))
-    else:
-        print(f"Skipping history update due to failed Hub upload.")
-
-    history_max_items = 10
-    if len(history) > history_max_items:
-        history.pop()
-
-    # 履歴ファイルを更新し、Hubにアップロードする
-    temp_history_filepath = "temp_history.txt"
-    with open(temp_history_filepath, "w", encoding="utf-8") as f:
-        for img_path_in_repo, cap_text, _ in history:
-            f.write(f"{img_path_in_repo}|||{cap_text}\n")
+    timestamp = datetime.datetime.now().strftime("%Y%m%d%H%M%S")
+    filename = f"image_{timestamp}.png"
+    filepath = f"temp_{filename}" # ここで一時ファイルのパスを定義
 
     try:
-        api.upload_file(
-            path_or_fileobj=temp_history_filepath,
-            path_in_repo=HISTORY_FILE,
-            repo_id=HF_REPO_ID,
-            repo_type="dataset",
+        
+        if rand:
+            seed = random.randint(0, MAX_SEED)
+        generator = torch.Generator(device=device).manual_seed(seed)
+
+        pipe.scheduler = euler_scheduler if scheduler_type == "Euler Ancestral" else dpm_scheduler
+        pipe.scheduler.set_timesteps(steps)
+
+        def _callback(pipeline, step_idx, timestep, callback_kwargs):
+            progress(step_idx / steps, desc=f"Step {step_idx}/{steps}")
+            return callback_kwargs
+
+        output = pipe(
+            prompt=prompt,
+            negative_prompt=neg or None,
+            guidance_scale=cfg,
+            num_inference_steps=steps,
+            width=w,
+            height=h,
+            generator=generator,
+            callback_on_step_end=_callback
         )
-        print(f"History file '{HISTORY_FILE}' updated on Hugging Face Hub.")
-    except Exception as e:
-        print(f"Error updating history file on Hub: {e}")
-    finally:
-        if os.path.exists(temp_history_filepath):
-            os.remove(temp_history_filepath)
+        img = output.images[0]
+
+        img.save(filepath)
+
+        caption_text = (
+            f"Prompt: {prompt}\n"
+            f"Negative: {neg or 'None'}\n"
+            f"Seed: {seed}\n"
+            f"Size: {w}×{h}\n"
+            f"CFG: {cfg}\n"
+            f"Steps: {steps}\n"
+            f"Scheduler: {scheduler_type}"
+        )
+
+        caption_text_for_history = caption_text.replace("\n", "|-|").strip()
+        # 画像をHubにアップロードし、そのURLとリポジトリ内パスを取得
+            # 公開用リポジトリにアップロード
+            uploaded_image_url, path_in_public_repo_for_history = upload_image_to_hub(filepath, filename)
+            # 非公開リポジトリにアップロード
+            path_in_private_repo_for_history = upload_image_to_private_hub(filepath, filename)
+
+
+        # 履歴を更新
+        global history
+        # Hubへのアップロードが成功した場合のみ履歴に追加
+        # historyリストには (非公開リポジトリのパス, キャプション, 公開リポジトリのURL) の形式で保存
+        if path_in_private_repo_for_history and uploaded_image_url:
+            history.insert(0, (path_in_private_repo_for_history, caption_text_for_history, uploaded_image_url))
+        else:
+            print(f"Skipping history update due to failed Hub upload.")
+
+        history_max_items = 10
+        if len(history) > history_max_items:
+            history.pop()
+
+        # 履歴ファイルを更新し、Hubにアップロードする
+        temp_history_filepath = "temp_history.txt"
+        with open(temp_history_filepath, "w", encoding="utf-8") as f:
+            for img_path_in_repo, cap_text, _ in history:
+                f.write(f"{img_path_in_repo}|||{cap_text}\n")
+
+        try:
+            api.upload_file(
+                path_or_fileobj=temp_history_filepath,
+                path_in_repo=HISTORY_FILE,
+                repo_id=HF_REPO_ID,
+                repo_type="dataset",
+            )
+            print(f"History file '{HISTORY_FILE}' updated on Hugging Face Hub.")
+        except Exception as e:
+            print(f"Error updating history file on Hub: {e}")
+        finally:
+            if os.path.exists(temp_history_filepath):
+                os.remove(temp_history_filepath)
 
-    progress(1.0, desc="Done!")
+        progress(1.0, desc="Done!")
 
-    # ギャラリー表示用のアイテムリストを生成（Hub上のURLを使用）
-    gallery_items = [(item[2], item[1].replace("|-|", "\n")) for item in history]
+        # ギャラリー表示用のアイテムリストを生成（Hub上のURLを使用）
+        gallery_items = [(item[2], item[1].replace("|-|", "\n")) for item in history]
     
-    processed_img, processed_gallery_items = process_image(img, gallery_items)
+        processed_img, processed_gallery_items = process_image(img, gallery_items)
     
-    latest_caption_table = make_html_table(caption_text)
-    return processed_img, processed_gallery_items, latest_caption_table
+        latest_caption_table = make_html_table(caption_text)
+        return processed_img, processed_gallery_items, latest_caption_table
+
+    finally:
+        # infer関数の最後に一時ファイルを削除
+        if os.path.exists(filepath):
+            os.remove(filepath)
+            print(f"生成画像の一時ファイル {filepath} を削除しました。")
 
 import gc
 import torch
@@ -373,6 +376,9 @@ html, .gradio-container, .dark, .dark * {
   word-wrap:break-word !important;
   display: none !important;
 }
+.caption.svelte-842rpi.svelte-842rpi{
+  display: none !important;
+}
 .gradio-spinner { display: none !important; }
 #custom-loader {
   align-items: center;