mousefox.app.adminframe

  1"""Home of `AdminFrame`."""
  2
  3import arrow
  4import functools
  5import json
  6import kvex as kx
  7import pgnet
  8import pprint
  9
 10
 11_STATUSES = {s.value: s for s in pgnet.Status}
 12_STATUS_COLORS = {
 13    pgnet.Status.OK.value: kx.XColor.from_hex("00ff00"),
 14    pgnet.Status.UNEXPECTED.value: kx.XColor.from_hex("bbbb00"),
 15    pgnet.Status.BAD.value: kx.XColor.from_hex("ff0000"),
 16}
 17
 18
 19class AdminFrame(kx.XFrame):
 20    """Widget for admin controls."""
 21
 22    _conpath = "client.user.admin"
 23
 24    def __init__(self, client: pgnet.Client):
 25        """Initialize the class with a client."""
 26        super().__init__()
 27        self._client = client
 28        self._make_widgets()
 29        self.app.controller.set_active_callback(self._conpath, self.set_focus)
 30        self.app.controller.bind(f"{self._conpath}.focus", self.set_focus)
 31        self.app.controller.bind(f"{self._conpath}.debug", self._request_debug)
 32
 33    def _make_widgets(self):
 34        with self.app.subtheme_context("accent"):
 35            title = kx.fwrap(kx.XLabel(text="Admin Panel", bold=True, font_size="36sp"))
 36            title.set_size(y="40sp")
 37        # Requests frame
 38        requests_placeholder = kx.XPlaceholder(
 39            button_text="Get requests",
 40            callback=self._refresh_requests,
 41        )
 42        self.requests_frame = kx.XContainer(requests_placeholder)
 43        custom_input_label = kx.XLabel(
 44            text="Custom packet builder",
 45            bold=True,
 46            underline=True,
 47            font_size="18dp",
 48        )
 49        custom_input_title = kx.wrap(custom_input_label)
 50        custom_input_title.set_size(y="40dp")
 51        packet_input_widgets = dict(
 52            message=kx.XInputPanelWidget(
 53                label="Message:",
 54                label_hint=0.2,
 55                orientation="vertical",
 56            ),
 57            payload=kx.XInputPanelWidget(
 58                label="Payload JSON:",
 59                label_hint=0.2,
 60                orientation="vertical",
 61            ),
 62        )
 63        self.packet_input = kx.XInputPanel(
 64            packet_input_widgets,
 65            reset_text="",
 66            invoke_text="Send packet",
 67        )
 68        self.packet_input.bind(on_invoke=self._on_packet_input)
 69        self.packet_input.set_size(y="100dp")
 70        self.custom_packet_frame = kx.XDBox()
 71        self.custom_packet_frame.add_widgets(custom_input_title, self.packet_input)
 72        # Response labels
 73        self.response_label = kx.XLabel(
 74            font_name="RobotoMono-Regular",
 75            padding=(10, 10),
 76            halign="left",
 77            valign="top",
 78            fixed_width=True,
 79        )
 80        response_label_frame = kx.XScroll(view=self.response_label)
 81        self.debug_label = kx.XLabel(
 82            font_name="RobotoMono-Regular",
 83            padding=(10, 10),
 84            halign="left",
 85            valign="top",
 86            fixed_width=True,
 87        )
 88        debug_label_frame = kx.fwrap(kx.XScroll(view=self.debug_label))
 89        debug_label_frame.set_size(x="300dp")
 90        # Assemble
 91        self.requests_frame.set_size(x="300dp")
 92        bottom_frame = kx.XBox()
 93        bottom_frame.add_widgets(
 94            debug_label_frame,
 95            response_label_frame,
 96            self.requests_frame,
 97        )
 98        main_frame = kx.XBox(orientation="vertical")
 99        main_frame.add_widgets(title, kx.pwrap(bottom_frame))
100        self.add_widget(kx.pwrap(main_frame))
101
102    def _refresh_requests(self):
103        self._client.send(pgnet.Packet(pgnet.util.Request.HELP), self._on_help_response)
104
105    def _on_help_response(self, response: pgnet.Response):
106        main_stack = kx.XDBox()
107        for request, params in response.payload.items():
108            panel_widgets = {
109                name: kx.XInputPanelWidget(label=f"{name}:", widget=ptype)
110                for name, ptype in params.items()
111            }
112            with self.app.subtheme_context("secondary"):
113                panel = kx.XInputPanel(
114                    panel_widgets,
115                    reset_text="",
116                    invoke_text=request,
117                    fill_button=True,
118                )
119                panel.on_invoke = functools.partial(self._on_request_invoke, request)
120                panel = panel
121            with self.app.subtheme_context("accent"):
122                text = request.removeprefix("__pgnet__.")
123                text = text.replace("_", " ").capitalize()
124                lbl = kx.XLabel(text=text, bold=True, font_size="18dp")
125                lbl = kx.pwrap(kx.fwrap(lbl))
126                lbl.set_size(y="40dp")
127            main_stack.add_widgets(lbl, panel)
128        if self.custom_packet_frame.parent:
129            self.custom_packet_frame.parent.remove_widget(self.custom_packet_frame)
130        main_stack.add_widget(self.custom_packet_frame)
131        scroll = kx.XScroll(view=main_stack)
132        self.requests_frame.content = kx.fpwrap(scroll, subtheme_name="secondary")
133
134    def _request_debug(self, *args):
135        self._client.send(
136            pgnet.Packet(pgnet.util.Request.DEBUG),
137            self._response_callback,
138        )
139
140    def _on_request_invoke(self, request: str, values: dict):
141        self._client.send(pgnet.Packet(request, values), self._response_callback)
142
143    def _on_packet_input(self, w, values):
144        message = values["message"]
145        payload_text = values["payload"]
146        try:
147            payload = json.loads(payload_text)
148        except Exception:
149            payload = dict()
150        packet = pgnet.Packet(message, payload)
151        self._client.send(packet, self._response_callback)
152        self.packet_input.set_focus("message")
153
154    def _response_callback(self, response: pgnet.Response):
155        sb = self.subtheme
156        status = _STATUSES[response.status]
157        status_color = _STATUS_COLORS[status]
158        statusstr = status_color.markup(status.name)
159        timestr = arrow.get(response.created_on).to("local").format("HH:mm:ss MMM DD")
160        debug_strs = [
161            f"{sb.fg2.markup('Status:')} {status.value} ({statusstr})",
162            f"{sb.fg2.markup('Created:')} {timestr}",
163            response.debug_repr,
164        ]
165        self.debug_label.text = "\n\n".join(debug_strs)
166        response_strs = [
167            f"{sb.fg2.markup('Response:')} {response.message}",
168        ]
169        for k, v in response.payload.items():
170            vstr = v if isinstance(v, str) else pprint.pformat(v, width=10_000)
171            response_strs.append(f"\n[u]{sb.fg2.markup(k)}[/u]\n{vstr}")
172        self.response_label.text = "\n".join(response_strs)
173
174    def set_focus(self, *args):
175        """Refresh requests frame on focus if empty."""
176        if not self.requests_frame.content:
177            self._refresh_requests()