File size: 5,214 Bytes
d8ebb7a
 
 
 
c566758
 
d8ebb7a
 
 
c566758
d8ebb7a
 
 
 
 
c566758
d8ebb7a
 
 
 
 
 
a3c4cfe
 
 
d8ebb7a
 
 
 
 
 
 
 
 
 
 
 
c566758
a3c4cfe
 
 
 
 
 
 
 
c566758
d8ebb7a
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
a3c4cfe
 
 
 
 
 
 
c566758
d8ebb7a
 
a3c4cfe
d8ebb7a
 
 
 
 
a3c4cfe
d8ebb7a
 
 
 
a3c4cfe
 
c566758
 
a3c4cfe
c566758
 
d8ebb7a
a3c4cfe
 
 
 
 
d8ebb7a
 
 
 
 
 
 
 
 
 
 
 
a3c4cfe
c566758
 
d8ebb7a
c566758
 
a3c4cfe
 
d8ebb7a
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
"""Admin status tab (spec section 12). Callbacks re-check the username; hiding is not enough."""

from __future__ import annotations

import logging

import gradio as gr

from src.services.access import is_admin
from src.services.flags import RuntimeFlags
from src.services.limits import LimitService
from src.services.session import VisitorSession
from src.ui.components import ApiVisibility, Localizer

REFRESH_SECONDS = 5
logger = logging.getLogger(__name__)


def build_admin_tab(
    *,
    demo: gr.Blocks,
    limits: LimitService,
    flags: RuntimeFlags,
    photo_studio: bool,
    high_quality: bool,
    loc: Localizer,
    session: gr.State,
    api_visibility: ApiVisibility,
) -> gr.Tab:
    t = loc.t
    visibility = {"api_visibility": api_visibility, "show_progress": "hidden"}

    with loc.make(
        gr.Tab, lambda lang: {"label": t(lang, "tab.admin")}, id="admin", visible=False
    ) as tab:
        summary = gr.Markdown()
        paused = loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.pause")})
        photo_open = (
            loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.photo")})
            if photo_studio
            else None
        )
        quality_on = (
            loc.make(gr.Checkbox, lambda lang: {"label": t(lang, "admin.quality")})
            if high_quality
            else None
        )
        ask_reset = loc.make(gr.Button, lambda lang: {"value": t(lang, "admin.reset")})
        with gr.Column(visible=False) as confirm:
            loc.make(gr.Markdown, lambda lang: {"value": t(lang, "admin.reset_confirm")})
            with gr.Row():
                yes = loc.make(
                    gr.Button, lambda lang: {"value": t(lang, "admin.reset_yes")}, variant="stop"
                )
                no = loc.make(gr.Button, lambda lang: {"value": t(lang, "admin.reset_cancel")})
    timer = gr.Timer(REFRESH_SECONDS, active=False)

    def render(lang: str) -> str:
        snap = limits.snapshot()
        latency = ", ".join(f"{m} {s:.1f} s" for m, s in sorted(snap.p50_seconds.items()))
        return "\n\n".join(
            [
                t(lang, "admin.paused" if snap.paused else "admin.running"),
                t(
                    lang,
                    "admin.usage",
                    used=snap.images_used,
                    max_images=snap.max_images,
                    spend=f"{snap.spend_usd:.2f}",
                    max_cost=f"{snap.max_cost_usd:.2f}",
                ),
                t(lang, "admin.activity", in_flight=snap.in_flight, **snap.recent),
                t(lang, "admin.latency", details=latency or t(lang, "admin.latency_none")),
            ]
        )

    def flag_state() -> list:
        state = []
        if photo_open is not None:
            state.append(flags.photo_studio)
        if quality_on is not None:
            state.append(flags.high_quality)
        return state

    def on_load(current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return [gr.skip()] * (4 + len(flag_state()))
        return (
            gr.update(visible=True),
            gr.Timer(active=True),
            render(current.lang),
            limits.paused,
            *flag_state(),
        )

    def on_refresh(current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return [gr.skip()] * (2 + len(flag_state()))
        return render(current.lang), limits.paused, *flag_state()

    def on_photo(value: bool, request: gr.Request):
        if is_admin(request.username):
            flags.set_photo_studio(bool(value))
            logger.info("admin action: photo studio %s", "opened" if value else "closed")

    def on_quality(value: bool, request: gr.Request):
        if is_admin(request.username):
            flags.set_high_quality(bool(value))
            logger.info("admin action: high-quality model %s", "on" if value else "off")

    def on_pause(value: bool, current: VisitorSession, request: gr.Request):
        if not is_admin(request.username):
            return gr.skip()
        limits.set_paused(bool(value))
        return render(current.lang)

    def on_confirm_reset(current: VisitorSession, request: gr.Request):
        if is_admin(request.username):
            limits.reset_counters()
            return gr.update(visible=False), render(current.lang)
        return gr.skip(), gr.skip()

    extra = [box for box in (photo_open, quality_on) if box is not None]
    demo.load(on_load, [session], [tab, timer, summary, paused, *extra], **visibility)
    timer.tick(on_refresh, [session], [summary, paused, *extra], **visibility)
    paused.input(on_pause, [paused, session], [summary], **visibility)
    if photo_open is not None:
        photo_open.input(on_photo, [photo_open], None, **visibility)
    if quality_on is not None:
        quality_on.input(on_quality, [quality_on], None, **visibility)
    ask_reset.click(lambda: gr.update(visible=True), None, [confirm], **visibility)
    no.click(lambda: gr.update(visible=False), None, [confirm], **visibility)
    yes.click(on_confirm_reset, [session], [confirm, summary], **visibility)
    return tab