| import os |
| import tempfile |
| from typing import Any |
|
|
| import gradio as gr |
| import httpx |
|
|
|
|
| |
| |
| BASE_URL = os.getenv("API_BASE_URL", "http://127.0.0.1:7860") |
|
|
|
|
| def _public_base_url(request: gr.Request | None) -> str: |
| if request is None: |
| return os.getenv("PUBLIC_BASE_URL", "").rstrip("/") |
| headers = {k.lower(): v for k, v in dict(request.headers).items()} |
| origin = headers.get("origin") |
| if origin: |
| return origin.rstrip("/") |
| proto = headers.get("x-forwarded-proto") |
| host = headers.get("x-forwarded-host") or headers.get("host") |
| if proto and host: |
| return f"{proto}://{host}".rstrip("/") |
| if host: |
| return f"https://{host}".rstrip("/") |
| return os.getenv("PUBLIC_BASE_URL", "").rstrip("/") |
|
|
|
|
| def create_gradio_app(): |
| """Create and return the Gradio Blocks app for Music Memories.""" |
|
|
| def _maybe_int(value: Any) -> int | None: |
| if value is None: |
| return None |
| try: |
| return int(value) |
| except Exception: |
| return None |
|
|
| def _json_request( |
| method: str, |
| path: str, |
| *, |
| params: dict[str, Any] | None = None, |
| data: dict[str, Any] | None = None, |
| files: dict[str, Any] | None = None, |
| follow_redirects: bool = True, |
| timeout: float = 15.0, |
| ) -> dict[str, Any]: |
| url = f"{BASE_URL}{path}" |
| with httpx.Client(follow_redirects=follow_redirects, timeout=timeout) as client: |
| resp = client.request(method, url, params=params, data=data, files=files) |
| resp.raise_for_status() |
| return resp.json() |
|
|
| def _redirect_location(path: str, *, params: dict[str, Any] | None = None) -> str: |
| url = f"{BASE_URL}{path}" |
| with httpx.Client(follow_redirects=False, timeout=15.0) as client: |
| resp = client.get(url, params=params) |
| if resp.status_code in (301, 302, 303, 307, 308): |
| return resp.headers.get("location") or "" |
| resp.raise_for_status() |
| return "" |
|
|
| def add_song_fn(title, artist, album, duration, bpm, energy_level, lyrics, audio_path): |
| try: |
| |
| data: dict[str, Any] = {"title": title, "artist": artist} |
| if album: |
| data["album"] = album |
| duration_i = _maybe_int(duration) |
| if duration_i is not None: |
| data["duration"] = str(duration_i) |
| bpm_i = _maybe_int(bpm) |
| if bpm_i is not None: |
| data["bpm"] = str(bpm_i) |
| energy_i = _maybe_int(energy_level) |
| if energy_i is not None: |
| data["energy_level"] = str(energy_i) |
| if lyrics: |
| data["lyrics"] = lyrics |
|
|
| if audio_path: |
| filename = os.path.basename(str(audio_path)) |
| with open(str(audio_path), "rb") as f: |
| files = {"audio_file": (filename, f, "audio/mpeg")} |
| out = _json_request("POST", "/songs", data=data, files=files) |
| else: |
| out = _json_request("POST", "/songs", data=data) |
|
|
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def add_user_fn(name): |
| try: |
| out = _json_request("POST", "/users", params={"name": name}, timeout=10.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def add_memory_fn(user_id, description, date, song_id): |
| try: |
| params: dict[str, Any] = {"user_id": _maybe_int(user_id), "description": description} |
| if params["user_id"] is None: |
| raise ValueError("user_id is required") |
| if date: |
| params["date"] = date |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is not None: |
| params["song_id"] = song_id_i |
| out = _json_request("POST", "/memories", params=params) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def add_playlist_fn(name, vibe_code, mood_description): |
| try: |
| params: dict[str, Any] = {"name": name} |
| if vibe_code: |
| params["vibe_code"] = vibe_code |
| if mood_description: |
| params["mood_description"] = mood_description |
| out = _json_request("POST", "/playlists", params=params) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def add_context_fn(weather, time_of_day, location_type): |
| try: |
| params: dict[str, Any] = {} |
| if weather: |
| params["weather"] = weather |
| if time_of_day: |
| params["time_of_day"] = time_of_day |
| if location_type: |
| params["location_type"] = location_type |
| out = _json_request("POST", "/contexts", params=params, timeout=10.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def get_song_fn(song_id): |
| try: |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is None: |
| raise ValueError("song_id is required") |
| out = _json_request("GET", f"/songs/{song_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def delete_song_fn(song_id): |
| try: |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is None: |
| raise ValueError("song_id is required") |
| out = _json_request("DELETE", f"/songs/{song_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def upload_song_audio_fn(song_id, audio_path): |
| try: |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is None: |
| raise ValueError("song_id is required") |
| if not audio_path: |
| raise ValueError("audio file is required") |
| filename = os.path.basename(str(audio_path)) |
| with open(str(audio_path), "rb") as f: |
| files = {"audio_file": (filename, f, "audio/mpeg")} |
| out = _json_request("POST", f"/songs/{song_id_i}/upload-audio", files=files) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def song_stream_url_fn(song_id, request: gr.Request | None = None): |
| try: |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is None: |
| raise ValueError("song_id is required") |
| public_base = _public_base_url(request) |
| if public_base: |
| return "Success!", f"{public_base}/songs/{song_id_i}/stream" |
| return "Success!", f"/songs/{song_id_i}/stream" |
| except Exception as e: |
| return "Error", str(e) |
|
|
| def song_download_fn(song_id): |
| try: |
| song_id_i = _maybe_int(song_id) |
| if song_id_i is None: |
| raise ValueError("song_id is required") |
| url = f"{BASE_URL}/songs/{song_id_i}/download" |
| with httpx.Client(timeout=30.0) as client: |
| resp = client.get(url) |
| resp.raise_for_status() |
| tmp = tempfile.NamedTemporaryFile(delete=False, suffix=".mp3") |
| tmp.write(resp.content) |
| tmp.flush() |
| tmp.close() |
| return "Success!", tmp.name |
| except Exception as e: |
| return "Error", None |
|
|
| def get_user_fn(user_id): |
| try: |
| user_id_i = _maybe_int(user_id) |
| if user_id_i is None: |
| raise ValueError("user_id is required") |
| out = _json_request("GET", f"/users/{user_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def list_user_memories_fn(user_id): |
| try: |
| user_id_i = _maybe_int(user_id) |
| if user_id_i is None: |
| raise ValueError("user_id is required") |
| data = _json_request("GET", f"/users/{user_id_i}/memories", timeout=10.0) |
| memories = data.get("memories", []) |
| table = [[m.get("id"), m.get("user_id"), (m.get("description") or "")[:50], m.get("date", "")] for m in memories] |
| return f"Found {len(memories)} memories", table |
| except Exception as e: |
| return "Error", [] |
|
|
| def user_activity_fn(user_id, limit, action_type): |
| try: |
| user_id_i = _maybe_int(user_id) |
| if user_id_i is None: |
| raise ValueError("user_id is required") |
| params: dict[str, Any] = {} |
| limit_i = _maybe_int(limit) |
| if limit_i is not None: |
| params["limit"] = limit_i |
| if action_type: |
| params["action_type"] = action_type |
| out = _json_request("GET", f"/users/{user_id_i}/activity", params=params) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def delete_playlist_fn(playlist_id): |
| try: |
| playlist_id_i = _maybe_int(playlist_id) |
| if playlist_id_i is None: |
| raise ValueError("playlist_id is required") |
| out = _json_request("DELETE", f"/playlists/{playlist_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def list_playlists_fn(): |
| try: |
| data = _json_request("GET", "/playlists", timeout=10.0) |
| playlists = data.get("playlists", []) |
| return [[p.get("id"), p.get("name"), p.get("vibe_code", "")] for p in playlists] |
| except Exception as e: |
| return [["Error", str(e), ""]] |
|
|
| def delete_memory_fn(memory_id): |
| try: |
| memory_id_i = _maybe_int(memory_id) |
| if memory_id_i is None: |
| raise ValueError("memory_id is required") |
| out = _json_request("DELETE", f"/memories/{memory_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def list_contexts_fn(): |
| try: |
| data = _json_request("GET", "/contexts", timeout=10.0) |
| contexts = data.get("contexts", []) |
| return [[c.get("id"), c.get("weather", ""), c.get("time_of_day", ""), c.get("location_type", "")] for c in contexts] |
| except Exception as e: |
| return [["Error", str(e), "", ""]] |
|
|
| def delete_context_fn(context_id): |
| try: |
| context_id_i = _maybe_int(context_id) |
| if context_id_i is None: |
| raise ValueError("context_id is required") |
| out = _json_request("DELETE", f"/contexts/{context_id_i}") |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def search_songs_fn(query, n_results): |
| try: |
| data = _json_request("GET", "/search/songs", params={"q": query, "n": n_results}, timeout=15.0) |
| results = data.get("results", []) |
| table = [[r.get("id"), r.get("title"), r.get("artist"), f"{float(r.get('distance', 0.0)):.4f}"] for r in results] if results else [] |
| return f"Found {len(results)} songs", table |
| except Exception as e: |
| return "Error", [] |
|
|
| def search_memories_fn(query, n_results): |
| try: |
| data = _json_request("GET", "/search/memories", params={"q": query, "n": n_results}, timeout=15.0) |
| results = data.get("results", []) |
| table = [[r.get("id"), r.get("user_id"), (r.get("document") or "")[:50], f"{float(r.get('distance', 0.0)):.4f}"] for r in results] if results else [] |
| return f"Found {len(results)} memories", table |
| except Exception as e: |
| return "Error", [] |
|
|
| def search_contexts_fn(query, n_results): |
| try: |
| data = _json_request("GET", "/search/contexts", params={"q": query, "n": n_results}, timeout=15.0) |
| results = data.get("results", []) |
| table = [[r.get("id"), r.get("weather"), r.get("time_of_day"), r.get("location_type"), f"{float(r.get('distance', 0.0)):.4f}"] for r in results] if results else [] |
| return f"Found {len(results)} contexts", table |
| except Exception: |
| return "Error", [] |
|
|
| def search_playlists_fn(query, n_results): |
| try: |
| data = _json_request("GET", "/search/playlists", params={"q": query, "n": n_results}, timeout=15.0) |
| results = data.get("results", []) |
| table = [[r.get("id"), r.get("name"), f"{float(r.get('distance', 0.0)):.4f}"] for r in results] if results else [] |
| return f"Found {len(results)} playlists", table |
| except Exception as e: |
| return "Error", [] |
|
|
| def list_songs_fn(): |
| try: |
| data = _json_request("GET", "/songs", timeout=10.0) |
| songs = data.get("songs", []) |
| return [[s.get("id"), s.get("title"), s.get("artist"), s.get("album", ""), s.get("bpm", ""), s.get("has_audio"), s.get("play_count")] for s in songs] |
| except Exception as e: |
| return [["Error", str(e), "", "", "", "", ""]] |
|
|
| def list_users_fn(): |
| try: |
| data = _json_request("GET", "/users", timeout=10.0) |
| users = data.get("users", []) |
| return [[u.get("id"), u.get("name"), u.get("created_at", "")] for u in users] |
| except Exception as e: |
| return [["Error", str(e), ""]] |
|
|
| def list_memories_fn(): |
| try: |
| data = _json_request("GET", "/memories", timeout=10.0) |
| memories = data.get("memories", []) |
| return [[m.get("id"), m.get("user_id"), (m.get("description") or "")[:50], m.get("date", ""), m.get("song_id", "")] for m in memories] |
| except Exception as e: |
| return [["Error", str(e), "", "", ""]] |
|
|
| def list_history_fn(user_id, limit): |
| try: |
| params: dict[str, Any] = {} |
| user_id_i = _maybe_int(user_id) |
| if user_id_i is not None: |
| params["user_id"] = user_id_i |
| limit_i = _maybe_int(limit) |
| if limit_i is not None: |
| params["limit"] = limit_i |
| out = _json_request("GET", "/history", params=params, timeout=20.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def add_history_fn(user_id, song_id, context_id, duration_seconds): |
| try: |
| params: dict[str, Any] = {} |
| user_id_i = _maybe_int(user_id) |
| song_id_i = _maybe_int(song_id) |
| if user_id_i is None or song_id_i is None: |
| raise ValueError("user_id and song_id are required") |
| params["user_id"] = user_id_i |
| params["song_id"] = song_id_i |
| context_id_i = _maybe_int(context_id) |
| if context_id_i is not None: |
| params["context_id"] = context_id_i |
| dur_i = _maybe_int(duration_seconds) |
| if dur_i is not None: |
| params["duration_seconds"] = dur_i |
| out = _json_request("POST", "/history", params=params, timeout=20.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def analytics_summary_fn(hours): |
| try: |
| hours_i = _maybe_int(hours) |
| params = {"hours": hours_i} if hours_i is not None else {} |
| out = _json_request("GET", "/analytics/summary", params=params, timeout=20.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def analytics_top_songs_fn(limit): |
| try: |
| limit_i = _maybe_int(limit) |
| params = {"limit": limit_i} if limit_i is not None else {} |
| out = _json_request("GET", "/analytics/top-songs", params=params, timeout=20.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def analytics_popular_searches_fn(hours, limit): |
| try: |
| params: dict[str, Any] = {} |
| hours_i = _maybe_int(hours) |
| limit_i = _maybe_int(limit) |
| if hours_i is not None: |
| params["hours"] = hours_i |
| if limit_i is not None: |
| params["limit"] = limit_i |
| out = _json_request("GET", "/analytics/popular-searches", params=params, timeout=20.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def storage_files_fn(): |
| try: |
| out = _json_request("GET", "/storage/files", timeout=20.0) |
| files = out.get("files", []) |
| table = [[f.get("object_name"), f.get("size"), f.get("last_modified")] for f in files] |
| return f"Found {len(files)} files", table |
| except Exception as e: |
| return "Error", [] |
|
|
| def health_fn(): |
| try: |
| out = _json_request("GET", "/health", timeout=10.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| def root_fn(): |
| try: |
| out = _json_request("GET", "/", timeout=10.0) |
| return "Success!", out |
| except Exception as e: |
| return "Error", {"error": str(e)} |
|
|
| with gr.Blocks(title="Music Memories UI") as demo: |
| gr.Markdown("# 🎵 Music Memories") |
| gr.Markdown("Manage your music library, memories, and playlists with semantic search.") |
|
|
| with gr.Tab("🎶 Songs"): |
| with gr.Group(): |
| gr.Markdown("### Add Song") |
| song_title = gr.Textbox(label="Title", placeholder="Song title") |
| song_artist = gr.Textbox(label="Artist", placeholder="Artist name") |
| song_album = gr.Textbox(label="Album", placeholder="Album (optional)") |
| song_duration = gr.Number(label="Duration (seconds)", precision=0) |
| song_bpm = gr.Number(label="BPM", precision=0) |
| song_energy = gr.Slider(1, 10, value=5, label="Energy Level") |
| song_lyrics = gr.Textbox(label="Lyrics", placeholder="Lyrics for semantic search", lines=3) |
| song_audio = gr.File(label="Audio File (MP3, optional)", file_types=[".mp3"], type="filepath") |
| song_add_btn = gr.Button("Add Song") |
| song_add_status = gr.Textbox(label="Status") |
| song_add_output = gr.JSON(label="Response") |
| song_add_btn.click( |
| fn=add_song_fn, |
| inputs=[song_title, song_artist, song_album, song_duration, song_bpm, song_energy, song_lyrics, song_audio], |
| outputs=[song_add_status, song_add_output], |
| ) |
|
|
| with gr.Group(): |
| gr.Markdown("### Get / Delete Song") |
| song_id = gr.Number(label="Song ID", precision=0) |
| song_get_btn = gr.Button("Get Song") |
| song_del_btn = gr.Button("Delete Song") |
| song_action_status = gr.Textbox(label="Status") |
| song_action_output = gr.JSON(label="Response") |
| song_get_btn.click(fn=get_song_fn, inputs=[song_id], outputs=[song_action_status, song_action_output]) |
| song_del_btn.click(fn=delete_song_fn, inputs=[song_id], outputs=[song_action_status, song_action_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Audio: Upload / Stream / Download") |
| audio_song_id = gr.Number(label="Song ID", precision=0) |
| audio_upload = gr.File(label="MP3 File", file_types=[".mp3"], type="filepath") |
| audio_upload_btn = gr.Button("Upload/Update Audio") |
| audio_upload_status = gr.Textbox(label="Status") |
| audio_upload_output = gr.JSON(label="Response") |
| audio_upload_btn.click( |
| fn=upload_song_audio_fn, |
| inputs=[audio_song_id, audio_upload], |
| outputs=[audio_upload_status, audio_upload_output], |
| ) |
|
|
| stream_btn = gr.Button("Get Stream URL") |
| stream_status = gr.Textbox(label="Status") |
| stream_url = gr.Textbox(label="Presigned Stream URL") |
| stream_btn.click(fn=song_stream_url_fn, inputs=[audio_song_id], outputs=[stream_status, stream_url]) |
|
|
| download_btn = gr.Button("Download MP3 via API") |
| download_status = gr.Textbox(label="Status") |
| download_file = gr.File(label="Downloaded File") |
| download_btn.click(fn=song_download_fn, inputs=[audio_song_id], outputs=[download_status, download_file]) |
| |
| with gr.Group(): |
| gr.Markdown("### All Songs") |
| songs_table = gr.Dataframe(headers=["ID", "Title", "Artist", "Album", "BPM", "Has Audio", "Play Count"], label="Songs") |
| songs_load_btn = gr.Button("Load Songs") |
| songs_load_btn.click(fn=list_songs_fn, outputs=songs_table) |
|
|
| with gr.Tab("👤 Users"): |
| with gr.Group(): |
| gr.Markdown("### Add User") |
| user_name = gr.Textbox(label="Name", placeholder="User name") |
| user_add_btn = gr.Button("Add User") |
| user_add_status = gr.Textbox(label="Status") |
| user_add_output = gr.JSON(label="Response") |
| user_add_btn.click(fn=add_user_fn, inputs=user_name, outputs=[user_add_status, user_add_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Get User") |
| get_user_id = gr.Number(label="User ID", precision=0) |
| get_user_btn = gr.Button("Get User") |
| get_user_status = gr.Textbox(label="Status") |
| get_user_output = gr.JSON(label="Response") |
| get_user_btn.click(fn=get_user_fn, inputs=[get_user_id], outputs=[get_user_status, get_user_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### User Memories") |
| um_user_id = gr.Number(label="User ID", precision=0) |
| um_btn = gr.Button("Load User Memories") |
| um_summary = gr.Textbox(label="Results") |
| um_table = gr.Dataframe(headers=["ID", "User ID", "Description", "Date"], label="Memories") |
| um_btn.click(fn=list_user_memories_fn, inputs=[um_user_id], outputs=[um_summary, um_table]) |
|
|
| with gr.Group(): |
| gr.Markdown("### User Activity") |
| ua_user_id = gr.Number(label="User ID", precision=0) |
| ua_limit = gr.Number(label="Limit", value=50, precision=0) |
| ua_action_type = gr.Textbox(label="Action Type (optional)") |
| ua_btn = gr.Button("Load Activity") |
| ua_status = gr.Textbox(label="Status") |
| ua_output = gr.JSON(label="Response") |
| ua_btn.click(fn=user_activity_fn, inputs=[ua_user_id, ua_limit, ua_action_type], outputs=[ua_status, ua_output]) |
| |
| with gr.Group(): |
| gr.Markdown("### All Users") |
| users_table = gr.Dataframe(headers=["ID", "Name", "Created At"], label="Users") |
| users_load_btn = gr.Button("Load Users") |
| users_load_btn.click(fn=list_users_fn, outputs=users_table) |
|
|
| with gr.Tab("💭 Memories"): |
| with gr.Group(): |
| gr.Markdown("### Add Memory") |
| mem_user_id = gr.Number(label="User ID", precision=0) |
| mem_desc = gr.Textbox(label="Description", placeholder="Describe the memory...", lines=2) |
| mem_date = gr.Textbox(label="Date (optional)", placeholder="YYYY-MM-DD") |
| mem_song_id = gr.Number(label="Song ID (optional)", precision=0) |
| mem_add_btn = gr.Button("Add Memory") |
| mem_add_status = gr.Textbox(label="Status") |
| mem_add_output = gr.JSON(label="Response") |
| mem_add_btn.click(fn=add_memory_fn, inputs=[mem_user_id, mem_desc, mem_date, mem_song_id], outputs=[mem_add_status, mem_add_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Delete Memory") |
| del_mem_id = gr.Number(label="Memory ID", precision=0) |
| del_mem_btn = gr.Button("Delete") |
| del_mem_status = gr.Textbox(label="Status") |
| del_mem_output = gr.JSON(label="Response") |
| del_mem_btn.click(fn=delete_memory_fn, inputs=[del_mem_id], outputs=[del_mem_status, del_mem_output]) |
| |
| with gr.Group(): |
| gr.Markdown("### All Memories") |
| memories_table = gr.Dataframe(headers=["ID", "User ID", "Description", "Date", "Song ID"], label="Memories") |
| memories_load_btn = gr.Button("Load Memories") |
| memories_load_btn.click(fn=list_memories_fn, outputs=memories_table) |
|
|
| with gr.Tab("🎵 Playlists"): |
| with gr.Group(): |
| gr.Markdown("### Add Playlist") |
| pl_name = gr.Textbox(label="Name", placeholder="Playlist name") |
| pl_vibe = gr.Textbox(label="Vibe Code", placeholder="e.g., chill, energetic") |
| pl_mood = gr.Textbox(label="Mood Description", placeholder="Describe the mood journey...", lines=2) |
| pl_add_btn = gr.Button("Add Playlist") |
| pl_add_status = gr.Textbox(label="Status") |
| pl_add_output = gr.JSON(label="Response") |
| pl_add_btn.click(fn=add_playlist_fn, inputs=[pl_name, pl_vibe, pl_mood], outputs=[pl_add_status, pl_add_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### All Playlists") |
| playlists_table = gr.Dataframe(headers=["ID", "Name", "Vibe Code"], label="Playlists") |
| playlists_load_btn = gr.Button("Load Playlists") |
| playlists_load_btn.click(fn=list_playlists_fn, outputs=playlists_table) |
|
|
| with gr.Group(): |
| gr.Markdown("### Delete Playlist") |
| del_pl_id = gr.Number(label="Playlist ID", precision=0) |
| del_pl_btn = gr.Button("Delete") |
| del_pl_status = gr.Textbox(label="Status") |
| del_pl_output = gr.JSON(label="Response") |
| del_pl_btn.click(fn=delete_playlist_fn, inputs=[del_pl_id], outputs=[del_pl_status, del_pl_output]) |
|
|
| with gr.Tab("🌤️ Contexts"): |
| with gr.Group(): |
| gr.Markdown("### Add Context") |
| ctx_weather = gr.Textbox(label="Weather", placeholder="e.g., rainy, sunny") |
| ctx_time = gr.Textbox(label="Time of Day", placeholder="e.g., morning, night") |
| ctx_location = gr.Textbox(label="Location Type", placeholder="e.g., home, gym, car") |
| ctx_add_btn = gr.Button("Add Context") |
| ctx_add_status = gr.Textbox(label="Status") |
| ctx_add_output = gr.JSON(label="Response") |
| ctx_add_btn.click(fn=add_context_fn, inputs=[ctx_weather, ctx_time, ctx_location], outputs=[ctx_add_status, ctx_add_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### All Contexts") |
| ctx_table = gr.Dataframe(headers=["ID", "Weather", "Time of Day", "Location"], label="Contexts") |
| ctx_load_btn = gr.Button("Load Contexts") |
| ctx_load_btn.click(fn=list_contexts_fn, outputs=ctx_table) |
|
|
| with gr.Group(): |
| gr.Markdown("### Delete Context") |
| del_ctx_id = gr.Number(label="Context ID", precision=0) |
| del_ctx_btn = gr.Button("Delete") |
| del_ctx_status = gr.Textbox(label="Status") |
| del_ctx_output = gr.JSON(label="Response") |
| del_ctx_btn.click(fn=delete_context_fn, inputs=[del_ctx_id], outputs=[del_ctx_status, del_ctx_output]) |
|
|
| with gr.Tab("🔍 Semantic Search"): |
| with gr.Group(): |
| gr.Markdown("### Search Songs by Vibe") |
| search_songs_query = gr.Textbox(label="Query", placeholder="e.g., 'sad breakup song', 'workout energy'") |
| search_songs_n = gr.Slider(1, 20, value=5, step=1, label="Results") |
| search_songs_btn = gr.Button("Search") |
| search_songs_summary = gr.Textbox(label="Results") |
| search_songs_table = gr.Dataframe(headers=["ID", "Title", "Artist", "Distance"], label="Songs") |
| search_songs_btn.click(fn=search_songs_fn, inputs=[search_songs_query, search_songs_n], outputs=[search_songs_summary, search_songs_table]) |
| |
| with gr.Group(): |
| gr.Markdown("### Search Memories") |
| search_mem_query = gr.Textbox(label="Query", placeholder="e.g., 'summer road trip', 'first dance'") |
| search_mem_n = gr.Slider(1, 20, value=5, step=1, label="Results") |
| search_mem_btn = gr.Button("Search") |
| search_mem_summary = gr.Textbox(label="Results") |
| search_mem_table = gr.Dataframe(headers=["ID", "User ID", "Description", "Distance"], label="Memories") |
| search_mem_btn.click(fn=search_memories_fn, inputs=[search_mem_query, search_mem_n], outputs=[search_mem_summary, search_mem_table]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Search Contexts") |
| search_ctx_query = gr.Textbox(label="Query", placeholder="e.g., 'rainy night drive', 'sunny beach'") |
| search_ctx_n = gr.Slider(1, 20, value=5, step=1, label="Results") |
| search_ctx_btn = gr.Button("Search") |
| search_ctx_summary = gr.Textbox(label="Results") |
| search_ctx_table = gr.Dataframe(headers=["ID", "Weather", "Time", "Location", "Distance"], label="Contexts") |
| search_ctx_btn.click(fn=search_contexts_fn, inputs=[search_ctx_query, search_ctx_n], outputs=[search_ctx_summary, search_ctx_table]) |
| |
| with gr.Group(): |
| gr.Markdown("### Search Playlists by Mood") |
| search_pl_query = gr.Textbox(label="Query", placeholder="e.g., 'chill evening vibes', 'party energy'") |
| search_pl_n = gr.Slider(1, 20, value=5, step=1, label="Results") |
| search_pl_btn = gr.Button("Search") |
| search_pl_summary = gr.Textbox(label="Results") |
| search_pl_table = gr.Dataframe(headers=["ID", "Name", "Distance"], label="Playlists") |
| search_pl_btn.click(fn=search_playlists_fn, inputs=[search_pl_query, search_pl_n], outputs=[search_pl_summary, search_pl_table]) |
|
|
| with gr.Tab("ℹ️ Health"): |
| with gr.Group(): |
| gr.Markdown("### Root") |
| root_btn = gr.Button("GET /") |
| root_status = gr.Textbox(label="Status") |
| root_output = gr.JSON(label="Response") |
| root_btn.click(fn=root_fn, outputs=[root_status, root_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Health") |
| health_btn = gr.Button("GET /health") |
| health_status = gr.Textbox(label="Status") |
| health_output = gr.JSON(label="Response") |
| health_btn.click(fn=health_fn, outputs=[health_status, health_output]) |
|
|
| with gr.Tab("▶️ History"): |
| with gr.Group(): |
| gr.Markdown("### List History") |
| hist_user_id = gr.Number(label="User ID (optional)", precision=0) |
| hist_limit = gr.Number(label="Limit", value=50, precision=0) |
| hist_list_btn = gr.Button("Load History") |
| hist_status = gr.Textbox(label="Status") |
| hist_output = gr.JSON(label="Response") |
| hist_list_btn.click(fn=list_history_fn, inputs=[hist_user_id, hist_limit], outputs=[hist_status, hist_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Add Play Event") |
| hist_add_user = gr.Number(label="User ID", precision=0) |
| hist_add_song = gr.Number(label="Song ID", precision=0) |
| hist_add_ctx = gr.Number(label="Context ID (optional)", precision=0) |
| hist_add_dur = gr.Number(label="Duration Seconds (optional)", precision=0) |
| hist_add_btn = gr.Button("Add History") |
| hist_add_status = gr.Textbox(label="Status") |
| hist_add_output = gr.JSON(label="Response") |
| hist_add_btn.click( |
| fn=add_history_fn, |
| inputs=[hist_add_user, hist_add_song, hist_add_ctx, hist_add_dur], |
| outputs=[hist_add_status, hist_add_output], |
| ) |
|
|
| with gr.Tab("📊 Analytics"): |
| with gr.Group(): |
| gr.Markdown("### Summary") |
| an_hours = gr.Number(label="Hours", value=24, precision=0) |
| an_sum_btn = gr.Button("Get Summary") |
| an_sum_status = gr.Textbox(label="Status") |
| an_sum_output = gr.JSON(label="Response") |
| an_sum_btn.click(fn=analytics_summary_fn, inputs=[an_hours], outputs=[an_sum_status, an_sum_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Top Songs") |
| an_limit = gr.Number(label="Limit", value=10, precision=0) |
| an_top_btn = gr.Button("Get Top Songs") |
| an_top_status = gr.Textbox(label="Status") |
| an_top_output = gr.JSON(label="Response") |
| an_top_btn.click(fn=analytics_top_songs_fn, inputs=[an_limit], outputs=[an_top_status, an_top_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Popular Searches") |
| ps_hours = gr.Number(label="Hours", value=24, precision=0) |
| ps_limit = gr.Number(label="Limit", value=10, precision=0) |
| ps_btn = gr.Button("Get Popular Searches") |
| ps_status = gr.Textbox(label="Status") |
| ps_output = gr.JSON(label="Response") |
| ps_btn.click(fn=analytics_popular_searches_fn, inputs=[ps_hours, ps_limit], outputs=[ps_status, ps_output]) |
|
|
| with gr.Group(): |
| gr.Markdown("### Storage Files") |
| sf_btn = gr.Button("List MP3 Files") |
| sf_summary = gr.Textbox(label="Results") |
| sf_table = gr.Dataframe(headers=["Object", "Size", "Last Modified"], label="Files") |
| sf_btn.click(fn=storage_files_fn, outputs=[sf_summary, sf_table]) |
|
|
| return demo |
|
|
|
|
| |
| gradio_app = create_gradio_app() |
|
|
| if __name__ == "__main__": |
| gradio_app.launch(share=False, server_name="0.0.0.0", server_port=7860) |
|
|
