zodiac.toga.app

  1#  # # <!-- // /*  SPDX-License-Identifier: MPL-2.0*/ -->
  2#  # # <!-- // /*  d a r k s h a p e s */ -->
  3
  4import os
  5import asyncio
  6import requests
  7from requests.exceptions import ConnectionError, ConnectTimeout
  8from urllib3.exceptions import MaxRetryError, NewConnectionError
  9from typing import Callable
 10
 11import toga
 12import toga.app
 13from toga import Key
 14from toga.constants import Direction
 15from toga.style import Pack
 16
 17from zodiac.streams.model_stream import ModelStream
 18from zodiac.streams.task_stream import TaskStream
 19from zodiac.streams.token_stream import TokenStream
 20import platform
 21from dspy import Prediction, streamify, context as dspy_context, inspect_history
 22
 23OS_NAME = platform.system  # replace with config from sdbx later
 24
 25
 26class Interface(toga.App):
 27    formatted_units = [" ❖ chr", " ⟐ tok", ' " sec ']
 28    bg_graph = "#070708"
 29    bg_text = "#1B1B1B"  # "#1B1B1B"  # "#09090B"
 30    bg = bg_text
 31    bg_static = "#5D5E62"
 32    activity = "#8122C4"
 33
 34    static = Pack(color="#727378")
 35    fg_static = Pack(color="#8D8E94")
 36    scroll_buffer = 5000  # chunks required to scroll down
 37    graph_disabled = "http://localhost"
 38    graph_server = "http://127.0.0.1:8188"
 39    status_info = ("Connecting...", "Server?", "Ready.", "Done.", "No File.", "Read Failed.", "Attached.", "Copied.")
 40    _is_cancelled = False
 41
 42    async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> toga.Widget:
 43        """Process and synthesize input data based on selected model.\n
 44        :param widget: The UI widget that triggered this action, typically used for state management.\n
 45        :type widget: toga.widgets
 46        :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
 47        :type external: bool"""
 48        from zodiac.toga.signatures import ready_predictor
 49
 50        self.response_panel.value += f"{os.path.basename(self.registry_entry.model)} :\n"
 51        await self.token_stream.set_tokenizer(self.registry_entry)
 52        prompts = {}
 53        if self.message_panel.value:
 54            cache = False
 55            prompts.setdefault("text", self.message_panel.value)
 56        # prompts.setdefault("audio",[0]) if else []
 57        # prompts.setdefault("image",[]) if image": []
 58        if stream := self.output_types.value == "text":
 59            context_data, predictor_data = await ready_predictor(self.registry_entry, dspy_stream=stream, async_stream=stream, cache=cache)
 60            await self.stream_text(prompts, context_data, predictor_data)
 61        else:
 62            content = await self.generate_media(prompts, self.registry_entry)  # context_data, predictor_data)
 63            return widget
 64        return widget
 65
 66    async def stream_text(self, prompts, context_data, predictor_data):
 67        from zodiac.toga.signatures import Predictor
 68        from litellm.types.utils import ModelResponseStream  # StatusStreamingCallback
 69        from dspy.streaming import StatusMessage, StreamResponse
 70
 71        self.response_panel.scroll_to_bottom()
 72        with dspy_context(**context_data):
 73            self.program = streamify(Predictor(), **predictor_data)
 74            async for prediction in self.program(question=prompts["text"]):
 75                if isinstance(prediction, ModelResponseStream) and prediction["choices"][0]["delta"]["content"]:
 76                    self.response_panel.value += prediction["choices"][0]["delta"]["content"]
 77                elif isinstance(prediction, StreamResponse) or hasattr(prediction, "chunk"):
 78                    self.response_panel.value += str(prediction.chunk)
 79                elif isinstance(prediction, Prediction) or hasattr(prediction, "answer"):
 80                    self.response_panel.value += str(prediction.answer)
 81                elif isinstance(prediction, StatusMessage) or hasattr(prediction, "message"):
 82                    self.status_display.text = self.status_text_prefix + str(prediction.message)
 83        self.response_panel.value += "\n--\n\n"
 84        return prediction
 85
 86    async def generate_media(self, prompts, registry_entry) -> None:  # , predictor_data
 87        from nnll.tensor_pipe.construct_pipe import ConstructPipeline
 88        from nnll.tensor_pipe.inference import run_inference
 89        from zodiac.streams.class_stream import best_package
 90        from zodiac.providers.constants import MIR_DB
 91
 92        pkg_data = await best_package(pkg_data=registry_entry)
 93        print(pkg_data)
 94        constructor = ConstructPipeline()
 95        pipe_data = await constructor.create_pipeline(registry_entry, pkg_data, MIR_DB)
 96        content = await run_inference(pipe_data, prompts, out_type=self.output_types.value)
 97        return content
 98        # from zodiac.toga.signatures import Predictor
 99
100        # return prediction
101
102    async def halt(self, widget, **kwargs) -> None:
103        """Stop processing prompt\n
104        :param widget: The calling widget object"""
105        if not self.program.done():
106            import gc
107
108            del self.program
109            gc.collect()
110            self.status_display.text = self.status_text_prefix + "Cancelled."
111
112    async def empty_prompt(self, widget, **kwargs) -> None:
113        """Clears the prompt input area.
114        :param widget: Triggering widget"""
115        self.message_panel.value = ""
116
117    async def copy_reply(self, widget, **kwargs) -> None:
118        """Push the reply into the clipboard
119        :param widget: Triggering widget"""
120        import pyperclip
121
122        pyperclip.copy(self.response_panel.value)
123        self.status_display.text = self.status_text_prefix + self.status_info[7]
124
125    async def attach_file(self, widget, **kwargs) -> None:
126        """Attaches a file's contents to the prompt area.
127        :param widget: Triggering widget"""
128        import json
129
130        try:
131            file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
132            self.status_display.text = f"Read. {file_path_named}"
133            if file_path_named is not None:
134                from nnll.metadata.json_io import read_json_file
135
136                file_contents = read_json_file(file_path_named)
137                self.message_panel.scroll_to_bottom()
138                self.message_panel.value = json.dumps(file_contents)
139                self.status_display.text = self.status_text_prefix + self.status_info[6]
140            else:
141                self.status_display.text = self.status_text_prefix + self.status_info[4]
142        except (ValueError, json.JSONDecodeError):
143            self.status_display.text = self.status_text_prefix + self.status_info[5]
144
145    async def reset_position(self, widget, **kwargs) -> None:
146        """Scrolls text panel to bottom after content update.
147        :param widget: text panel widget
148        """
149        setattr(self, "position_counter", getattr(self, "position_counter", 0) + 1)
150        if max(self.scroll_buffer, self.position_counter) >= self.scroll_buffer:
151            self.position_counter = 0
152            widget.scroll_to_bottom()
153
154    async def on_select_handler(self, widget, **kwargs) -> None:
155        """React to input/output choice\n
156        :param widget: The widget that triggered the event."""
157        selection = widget.value
158        if self.model_stream._graph.registry_entries is not None:
159            self.registry_entry = next(iter(registry["entry"] for registry in self.model_stream._graph.registry_entries if selection in registry["entry"].model))
160            await self.populate_task_stack()
161            await self.token_stream.set_tokenizer(self.registry_entry)
162        else:
163            self.registry_entry = "No model..."
164
165    async def model_graph(self):
166        """Builds the model graph."""
167        await self.model_stream.model_graph()
168
169    async def token_estimate(self, widget, **kwargs) -> None:
170        """Updates character and token count based on user input.
171        :param widget: Input widget providing text"""
172        token_count, character_count = await self.token_stream.token_count(message=self.message_panel.value)
173        self.character_stats.text = "{:02}".format(character_count) + "".join(self.formatted_units[0])
174        self.token_stats.text = "{:02}".format(token_count) + "".join(self.formatted_units[1])
175        self.time_stats.text = "{:02}".format(0.0) + "".join(self.formatted_units[2])
176
177    async def populate_in_types(self) -> None:
178        """Builds the input types selection."""
179        in_edge_names = await self.model_stream.show_edges()
180        self.input_types.items = in_edge_names
181
182    async def populate_out_types(self) -> None:
183        """Builds the output types selection."""
184        out_edges = await self.model_stream.show_edges(target=True)
185        self.output_types.items = out_edges
186
187    async def populate_model_stack(self, widget: toga.Widget = None, **kwargs) -> None:
188        """Builds the model stack selection dropdown."""
189
190        await self.model_stream.clear()
191        if self.input_types.value and self.output_types.value:
192            models = await self.model_stream.trace_models(self.input_types.value, self.output_types.value)
193            self.model_stack.items = models  # [model[0][:20] for model in models if len(model[0]) > 20]
194            await self.token_estimate(widget=self.message_panel)
195
196    async def populate_task_stack(self, widget: toga.Widget = None, **kwargs) -> None:
197        """Builds the task stack selection dropdown."""
198        selection = self.model_stack.value
199        if self.model_stream._graph.registry_entries:
200            registry_entry = next(
201                iter(
202                    registry["entry"]  # formatting
203                    for registry in self.model_stream._graph.registry_entries  # formatting
204                    if selection in registry["entry"].model
205                )
206            )
207        else:
208            registry_entry = "No models..."
209        await self.task_stream.set_filter_type(self.input_types.value, self.output_types.value)
210        if registry_entry and not isinstance(registry_entry, str):
211            tasks = await self.task_stream.filter_tasks(registry_entry)
212        else:
213            tasks = ""
214
215        self.task_stack.items = tasks
216
217    async def switch_tabs(self, widget: toga.Widget = None, **kwargs) -> None:
218        """Switches between text and graph tabs.
219        :param widget: The triggering widget (optional), defaults to None"""
220        self.browser_panel.evaluate_javascript("location.reload();")
221        self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
222        self.final_layout.style.background_color = self.bg
223        self.final_layout.refresh()
224        self.status_display.text += self.status_info[0]
225        await self.ping_server(widget=self.status_display)
226
227    async def ping_server(self, widget: toga.Widget, **kwargs) -> toga.Widget:
228        self.browser_panel.url = self.graph_server
229        try:
230            request = requests.get(self.graph_server, timeout=(3, 3))
231            if request is not None:
232                if hasattr(request, "status_code"):
233                    status = request.status_code
234                if (hasattr(request, "ok") and request.ok) or (hasattr(request, "reason") and request.reason == "OK"):
235                    await self.active_server()
236                elif hasattr(request, "json"):
237                    status = request.json()
238                    if status.get("result") == "OK":
239                        await self.active_server()
240                else:
241                    self.browser_panel.url = self.graph_disabled
242                    await self.active_server(False)
243            else:
244                self.browser_panel.url = self.graph_disabled
245                await self.active_server(False)
246        except (ConnectTimeout, ConnectionError, ConnectionRefusedError, MaxRetryError, NewConnectionError, OSError):
247            await self.active_server(False)
248            pass
249        return widget
250
251    async def active_server(self, enabled: bool = True):
252        if not enabled:
253            status_info = self.status_info[1]
254            self.browser_panel.url = self.graph_disabled
255        else:
256            status_info = self.status_info[2]
257            self.browser_panel.url = self.graph_server
258        for info in self.status_info:
259            self.status_display.text = self.status_display.text.replace(info, "")
260        self.status_display.text += status_info
261
262    def initialize_inputs(self):
263        """Initializes UI elements for input handling."""
264        self.character_stats = toga.Label("{:02}".format(0) + "".join(self.formatted_units[0]), **self.fg_static)
265        self.token_stats = toga.Label("{:02}".format(0) + "".join(self.formatted_units[1]), **self.fg_static)
266        self.time_stats = toga.Label("{:02}".format(0.0) + "".join(self.formatted_units[2]), **self.fg_static)
267        self.input_types = toga.Selection(items=[], on_change=self.populate_model_stack)
268        self.output_types = toga.Selection(items=[], on_change=self.populate_model_stack)
269        self.model_stack = toga.Selection(items=[], on_change=self.on_select_handler)
270        self.task_stack = toga.Selection(items=[], style=Pack(align_items="end"))
271        self.message_panel = toga.MultilineTextInput(placeholder="Prompt", on_change=self.token_estimate, style=Pack(flex=0.66, margin=10))
272        self.browser_panel = toga.WebView(url=self.graph_server, id="Graph ")
273        self.audio_panel = toga.Canvas()
274        self.response_panel = toga.MultilineTextInput(readonly=True, placeholder="Response", style=Pack(flex=5), on_change=self.reset_position)
275
276    def initialize_static(self) -> None:
277        """Create the main input fields"""
278
279        status_bar = toga.Row(
280            children=[
281                toga.Column(
282                    children=[
283                        toga.Row(
284                            children=[self.input_types, toga.Label("➾"), self.output_types, self.task_stack],
285                            style=Pack(align_items="end", gap=5),
286                        ),
287                        toga.Row(
288                            children=[self.model_stack, toga.Label("↪︎")],  # , live_stats
289                            style=Pack(align_items="end", text_direction="rtl", gap=5),
290                        ),
291                    ],
292                    style=Pack(vertical_align_items="center", gap=5, justify_content="end", align_items="end"),
293                ),
294                toga.Row(
295                    children=[
296                        toga.Column(children=[self.character_stats, self.token_stats, self.time_stats]),
297                        toga.Column(
298                            children=[
299                                toga.Button("▶︎", on_press=self.ticker, style=Pack(width=30, height=20, font_size="12")),
300                                toga.Button("⧉", on_press=self.copy_reply, style=Pack(width=30, height=20, font_size=15, vertical_align_items="start")),
301                            ],
302                            style=Pack(gap=5),
303                        ),
304                        toga.Column(
305                            children=[
306                                toga.Button(
307                                    """📎
308                                _""",
309                                    on_press=self.attach_file,
310                                    style=Pack(width=30, height=20, font_size="10", align_items="start", justify_content="start"),
311                                ),
312                                toga.Button("⌫", on_press=self.empty_prompt, style=Pack(width=30, height=20, font_size="14")),
313                            ],
314                            style=Pack(font_size="15", gap=5),
315                        ),
316                    ],
317                    style=Pack(vertical_align_items="center", gap=5, justify_content="start", align_items="start"),
318                ),
319            ],
320            style=Pack(margin=10, gap=5, vertical_align_items="center", justify_content="start", align_items="start"),
321        )
322        self.status_log = toga.Label(f"{inspect_history()}")  # show llm history
323        self.status_tab = toga.OptionItem(text="|  Connecting...", content=self.status_log, enabled=False)
324        # self.response_array = toga.Box(children=[self.response_panel], style=Pack(flex=1))
325        # self.endless_response = toga.ScrollContainer(content=self.response_array)
326        resize_area = toga.SplitContainer(
327            content=[
328                toga.OptionContainer(
329                    content=[
330                        ("Output", self.response_panel),
331                        ("Graph", self.browser_panel),
332                        self.status_tab,
333                    ],
334                    on_select=self.switch_tabs,
335                    style=Pack(background_color="#000000", flex=2),
336                    id="tab_panel",
337                ),
338                toga.Row(
339                    children=[
340                        toga.Column(justify_content="start", style=Pack(flex=0.33)),
341                        toga.Box(children=[self.message_panel], style=Pack(flex=1)),
342                        toga.Column(style=Pack(flex=0.33, justify_content="start")),
343                    ]
344                ),
345            ],
346            direction=Direction.HORIZONTAL,
347            style=Pack(flex=3),
348        )
349
350        self.final_layout = toga.Column(children=[status_bar, resize_area], style=Pack(background_color=self.bg_text, flex=1))
351
352    def initialize_layout(self) -> None:
353        """Create the layout of the application."""
354        self.main_window.content = self.final_layout
355
356    def startup(self) -> None:
357        """Startup Logic. Initialize widgets and layout, then asynchronous tasks for populating datagets"""
358        self.main_window = toga.MainWindow()
359        self.model_stream = ModelStream()
360        self.task_stream = TaskStream()
361        self.token_stream = TokenStream()
362        self.token_stream = TokenStream()
363
364        start = toga.Command(
365            self.ticker,
366            text="Start",
367            tooltip="Run the current available prompts.",
368            shortcut=Key.MOD_1 + Key.ENTER,
369            group=toga.Group.APP,
370            section=-1,
371        )
372        attach = toga.Command.standard(
373            self,
374            toga.Command.OPEN,
375            text="Attach File...",
376            tooltip="Attach a file to the prompt.",
377            shortcut=Key.MOD_1 + Key.O,
378            action=self.attach_file,
379            group=toga.Group.APP,
380            section=0,
381        )
382        copy_reply = toga.Command(
383            self.copy_reply,
384            text="Copy Response",
385            tooltip="Copy the response provided by the system",
386            group=toga.Group.APP,
387            section=0,
388        )
389        clear = toga.Command(
390            self.empty_prompt,
391            text="Clear Prompt",
392            tooltip="Empty the user prompt field.",
393            shortcut=Key.MOD_3 + Key.BACKSPACE,
394            group=toga.Group.APP,
395            section=1,
396        )
397        stop = toga.Command(
398            self.halt,
399            text="Stop",
400            tooltip="Cancel the current sequence generation.",
401            shortcut=Key.MOD_1 + Key.ESCAPE,  #
402            group=toga.Group.APP,
403            section=1,
404        )
405        self.commands.add(start, attach, copy_reply, clear, stop)
406
407        self.initialize_inputs()
408        self.initialize_static()
409        self.initialize_layout()
410        asyncio.create_task(self.model_graph())
411        asyncio.create_task(self.token_estimate(self))
412        asyncio.create_task(self.populate_in_types())
413        asyncio.create_task(self.populate_out_types())
414        asyncio.create_task(self.populate_model_stack())
415        asyncio.create_task(self.populate_task_stack())
416        self.main_window.show()
417        self.status_display = self.status_tab
418        self.bg = self.bg_graph
419        self.status_text_prefix = "|  "
420
421        asyncio.create_task(self.switch_tabs())
422
423
424def main(url: str = "http://127.0.0.1:8188"):
425    """The entry point for the application."""
426    app = Interface(
427        formal_name="Shadowbox",
428        app_id="org.darkshapes.shadowbox",
429        app_name="sdbx",
430        author="Darkshapes",
431        home_page="https://darkshapes.org",
432        description=" A generative AI instrument. ",
433    )
434
435    app.icon = toga.Icon(path="resources/anomaly_128x")
436    try:
437        app.main_loop()
438    except Exception as error_log:
439        print(error_log)