Wauplin HF staff commited on
Commit
57d05a0
1 Parent(s): 1615e08

Delete gallery_history.py

Browse files
Files changed (1) hide show
  1. gallery_history.py +0 -134
gallery_history.py DELETED
@@ -1,134 +0,0 @@
1
- """
2
- How to use:
3
- 1. Create a Space with a Persistent Storage attached. Filesystem will be available under `/data`.
4
- 2. Add `hf_oauth: true` to the Space metadata (README.md). Make sure to have Gradio>=3.41.0 configured.
5
- 3. Add `HISTORY_FOLDER` as a Space variable (example. `"/data/history"`).
6
- 4. Add `filelock` as dependency in `requirements.txt`.
7
- 5. Add history gallery to your Gradio app:
8
- a. Add imports: `from gallery_history import fetch_gallery_history, show_gallery_history`
9
- a. Add `history = show_gallery_history()` within `gr.Blocks` context.
10
- b. Add `.then(fn=fetch_gallery_history, inputs=[prompt, result], outputs=history)` on the generate event.
11
- """
12
- import json
13
- import os
14
- import numpy as np
15
- import shutil
16
- from pathlib import Path
17
- from PIL import Image
18
- from typing import Dict, List, Optional, Tuple
19
- from uuid import uuid4
20
-
21
- import gradio as gr
22
- from filelock import FileLock
23
-
24
- _folder = os.environ.get("HISTORY_FOLDER")
25
- if _folder is None:
26
- print(
27
- "'HISTORY_FOLDER' environment variable not set. User history will be saved "
28
- "locally and will be lost when the Space instance is restarted."
29
- )
30
- _folder = Path(__file__).parent / "history"
31
- if _folder.startswith("/data") and not os.path.exists("/data"):
32
- print(
33
- f"'HISTORY_FOLDER' environment variable is set to '{_folder}' which doesn't exist. User history will be saved "
34
- "locally and will be lost when the Space instance is restarted."
35
- )
36
- _folder = Path(__file__).parent / "history"
37
- HISTORY_FOLDER_PATH = Path(_folder)
38
-
39
- IMAGES_FOLDER_PATH = HISTORY_FOLDER_PATH / "images"
40
- IMAGES_FOLDER_PATH.mkdir(parents=True, exist_ok=True)
41
-
42
-
43
- def show_gallery_history():
44
- gr.Markdown(
45
- "## Your past generations\n\n(Log in to keep a gallery of your previous generations."
46
- " Your history will be saved and available on your next visit.)"
47
- )
48
- with gr.Column():
49
- with gr.Row():
50
- gr.LoginButton(min_width=250)
51
- gr.LogoutButton(min_width=250)
52
- gallery = gr.Gallery(
53
- label="Past images",
54
- show_label=True,
55
- elem_id="gallery",
56
- object_fit="contain",
57
- columns=4,
58
- height=512,
59
- preview=False,
60
- show_share_button=False,
61
- show_download_button=False,
62
- )
63
- gr.Markdown(
64
- "Make sure to save your images from time to time, this gallery may be deleted in the future."
65
- )
66
- gallery.attach_load_event(fetch_gallery_history, every=None)
67
- return gallery
68
-
69
-
70
- def fetch_gallery_history(
71
- prompt: Optional[str] = None,
72
- result: Optional[np.ndarray] = None,
73
- user: Optional[gr.OAuthProfile] = None,
74
- ):
75
- if user is None:
76
- return []
77
- try:
78
- if prompt is not None and result is not None: # None values means no new images
79
- new_image = Image.fromarray(result, 'RGB')
80
- return _update_user_history(user["preferred_username"], new_image, prompt)
81
- else:
82
- return _read_user_history(user["preferred_username"])
83
- except Exception as e:
84
- raise gr.Error(f"Error while fetching history: {e}") from e
85
-
86
-
87
- ####################
88
- # Internal helpers #
89
- ####################
90
-
91
-
92
- def _read_user_history(username: str) -> List[Tuple[str, str]]:
93
- """Return saved history for that user."""
94
- with _user_lock(username):
95
- path = _user_history_path(username)
96
- if path.exists():
97
- return json.loads(path.read_text())
98
- return [] # No history yet
99
-
100
-
101
- def _update_user_history(
102
- username: str, new_image: Image.Image, prompt: str
103
- ) -> List[Tuple[str, str]]:
104
- """Update history for that user and return it."""
105
- with _user_lock(username):
106
- # Read existing
107
- path = _user_history_path(username)
108
- if path.exists():
109
- images = json.loads(path.read_text())
110
- else:
111
- images = [] # No history yet
112
-
113
- # Copy image to persistent folder
114
- images = [(_copy_image(new_image), prompt)] + images
115
-
116
- # Save and return
117
- path.write_text(json.dumps(images))
118
- return images
119
-
120
-
121
- def _user_history_path(username: str) -> Path:
122
- return HISTORY_FOLDER_PATH / f"{username}.json"
123
-
124
-
125
- def _user_lock(username: str) -> FileLock:
126
- """Ensure history is not corrupted if concurrent calls."""
127
- return FileLock(f"{_user_history_path(username)}.lock")
128
-
129
-
130
- def _copy_image(new_image: Image.Image) -> str:
131
- """Copy image to the persistent storage."""
132
- dst = str(IMAGES_FOLDER_PATH / f"{uuid4().hex}.png")
133
- new_image.save(dst)
134
- return dst