1# # SPDX-License-Identifier: MPL-2.0 AND LicenseRef-Commons-Clause-License-Condition-1.0
2# # <!-- // /* d a r k s h a p e s */ -->
3
4# import os
5# import asyncio
6# from typing import Callable, Optional
7
8# import toga
9# from toga import Key
10# from zodiac.streams.model_stream import ModelStream
11# from zodiac.streams.task_stream import TaskStream
12# from zodiac.streams.token_stream import TokenStream
13
14
15# class DynamicApp:
16# character_stats = None
17
18# def __init__(self):
19# self.registry_entry = None
20# self.model_source = ModelStream()
21# self.task_source = TaskStream()
22# self.token_source = TokenStream()
23
24# async def initialize(self):
25# asyncio.create_task(self.token_estimate(self))
26# asyncio.create_task(self.model_graph())
27# asyncio.create_task(self.populate_in_types())
28# asyncio.create_task(self.populate_out_types())
29# asyncio.create_task(self.populate_model_stack())
30# asyncio.create_task(self.populate_task_stack())
31
32# async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> None:
33# """Process and synthesize input data based on selected model.\n
34# :param widget: The UI widget that triggered this action, typically used for state management.\n
35# :type widget: toga.widgets
36# :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
37# :type external: bool"""
38# from zodiac.toga.signatures import text_qa_stream
39# from dspy.streaming import StatusMessage
40
41# prompts = {"text": self.message_panel.value, "audio": [0], "image": []}
42# # self.status.text = "Processing..."
43# async for chunk in text_qa_stream(registry_entry=self.registry_entry, prompt=prompts["text"]):
44# if chunk and isinstance(chunk, StatusMessage):
45# self.status.text = chunk.message
46# elif chunk:
47# self.response_panel.value += chunk
48
49# async def empty_prompt(self, widget, **kwargs) -> None:
50# self.message_panel.value = ""
51
52# async def halt(self, widget, **kwargs) -> None:
53# """Stop processing prompt\n
54# :param widget: The calling widget object"""
55# self.status.text = "Cancelled."
56
57# async def include_file(self, widget, **kwargs) -> None:
58# import json
59
60# try:
61# file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
62# self.status.text = f"Read. {file_path_named}"
63# if file_path_named is not None:
64# from nnll.metadata.json_io import read_json_file
65
66# file_contents = read_json_file(file_path_named)
67# self.message_panel.scroll_to_bottom()
68# self.message_panel.value = json.dumps(file_contents)
69# self.status.text = f"Attached {os.path.basename(file_path_named)}."
70# else:
71# self.status.text = "No file. "
72# except (ValueError, json.JSONDecodeError):
73# self.status.text = "Read failed... "
74
75# async def reset_position(self, widget, **kwargs) -> None:
76# widget.scroll_to_bottom()
77
78# async def on_select_handler(self, widget, **kwargs):
79# """React to input/output choice\n
80# :param widget: The widget that triggered the event."""
81# selection = widget.value
82# registry_entry = next(iter(registry["entry"] for registry in self.model_source._graph.registry_entries if selection in registry["entry"].model))
83# self.registry_entry = registry_entry
84# await self.populate_task_stack()
85# await self.token_source.set_tokenizer(registry_entry)
86
87# async def model_graph(self):
88# """Builds the model graph."""
89# await self.model_source.model_graph()
90
91# async def token_estimate(self, widget, **kwargs):
92# token_count, character_count = await self.token_source.token_count(widget.value)
93# self.character_stats.text = "{:02}".format(character_count) + "".join(self.formatted_units[0])
94# self.token_stats.text = "{:02}".format(token_count) + "".join(self.formatted_units[1])
95# self.time_stats.text = "{:02}".format(0.0) + "".join(self.formatted_units[2])
96
97# async def populate_in_types(self):
98# """Builds the input types selection."""
99
100# in_edge_names = await self.model_source.show_edges()
101# self.input_types.items = in_edge_names
102
103# async def populate_out_types(self):
104# """Builds the output types selection."""
105
106# out_edges = await self.model_source.show_edges(target=True)
107# self.output_types.items = out_edges
108
109# async def populate_model_stack(self, widget: Optional[Callable] = None):
110# """Builds the model stack selection dropdown."""
111
112# await self.model_source.clear()
113# if self.input_types.value and self.output_types.value:
114# models = await self.model_source.trace_models(self.input_types.value, self.output_types.value)
115# self.model_stack.items = models # [model[0][:20] for model in models if len(model[0]) > 20]
116
117# async def populate_task_stack(self, widget: Optional[Callable] = None):
118# """Builds the task stack selection dropdown."""
119# selection = self.model_stack.value
120# registry_entry = next(
121# iter(
122# registry["entry"] # formatting
123# for registry in self.model_source._graph.registry_entries # formatting
124# if selection in registry["entry"].model
125# )
126# )
127# await self.task_source.set_filter_type(self.input_types.value, self.output_types.value)
128# tasks = await self.task_source.trace_tasks(registry_entry)
129
130# self.task_stack.items = tasks
131
132# async def switch_tabs(self, widget: Optional[Callable] = None):
133# self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
134# self.final_layout.style.background_color = self.bg
135# self.final_layout.refresh()
136# self.browser_panel.evaluate_javascript("location.reload();")
137
138# async def add_commands(self, parent):
139# # control_group = toga.Group("Controls", order=40)
140# start = toga.Command(
141# self.ticker,
142# text="Start",
143# tooltip="Run the current available prompts.",
144# shortcut=Key.MOD_1 + Key.ENTER,
145# group=toga.Group.APP,
146# section=-1,
147# )
148# stop = toga.Command(
149# self.halt,
150# text="Stop",
151# tooltip="Cancel the current sequence generation.",
152# shortcut=Key.ESCAPE,
153# group=toga.Group.APP,
154# section=0,
155# )
156# attach = toga.Command.standard(
157# self,
158# toga.Command.OPEN,
159# text="Attach File...",
160# tooltip="Attach a file to the prompt.",
161# shortcut=Key.MOD_1 + Key.O,
162# action=self.include_file,
163# group=toga.Group.APP,
164# section=1,
165# )
166# clear = toga.Command(
167# self.empty_prompt,
168# text="Clear Prompt",
169# tooltip="Empty the prompt field.",
170# shortcut=Key.MOD_3 + Key.BACKSPACE,
171# group=toga.Group.APP,
172# section=2,
173# )
174# parent.commands.add(start, stop, attach, clear)
175
176# # async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> None:
177# # """Process and synthesize input data based on selected model.\n
178# # :param widget: The UI widget that triggered this action, typically used for state management.\n
179# # :type widget: toga.widgets
180# # :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
181# # :type external: bool"""
182# # from zodiac.toga.signatures import text_qa_stream
183# # from dspy.streaming import StatusMessage
184
185# # prompts = {"text": self.message_panel.value, "audio": [0], "image": []}
186# # # self.status.text = "Processing..."
187# # async for chunk in text_qa_stream(registry_entry=self.registry_entry, prompt=prompts["text"]):
188# # if chunk and isinstance(chunk, StatusMessage):
189# # self.status.text = chunk.message
190# # elif chunk:
191# # self.response_panel.value += chunk
192
193# # async def empty_prompt(self, widget, **kwargs) -> None:
194# # self.message_panel.value = ""
195
196# # async def halt(self, widget, **kwargs) -> None:
197# # """Stop processing prompt\n
198# # :param widget: The calling widget object"""
199# # self.status.text = "Cancelled."
200
201# # async def include_file(self, widget, **kwargs) -> None:
202# # import json
203
204# # try:
205# # file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
206# # self.status.text = f"Read. {file_path_named}"
207# # if file_path_named is not None:
208# # from nnll.metadata.json_io import read_json_file
209
210# # file_contents = read_json_file(file_path_named)
211# # self.message_panel.scroll_to_bottom()
212# # self.message_panel.value = json.dumps(file_contents)
213# # self.status.text = f"Attached {os.path.basename(file_path_named)}."
214# # else:
215# # self.status.text = "No file. "
216# # except (ValueError, json.JSONDecodeError):
217# # self.status.text = "Read failed... "
218
219# # async def reset_position(self, widget, **kwargs) -> None:
220# # self.widget.scroll_to_bottom()
221
222# # async def on_select_handler(self, widget, **kwargs):
223# # """React to input/output choice\n
224# # :param widget: The widget that triggered the event."""
225# # selection = widget.value
226# # registry_entry = next(iter(registry["entry"] for registry in self.model_source._graph.registry_entries if selection in registry["entry"].model))
227# # self.registry_entry = registry_entry
228# # await self.populate_task_stack()
229# # await self.token_source.set_tokenizer(registry_entry)
230
231# # async def token_estimate(self, widget, **kwargs):
232# # token_count, character_count = await self.token_source.token_count(widget.value)
233# # self.static_interface.character_stats.text = "{:02}".format(character_count) + "".join(self.static_interface.formatted_units[0])
234# # self.static_interface.token_stats.text = "{:02}".format(token_count) + "".join(self.static_interface.formatted_units[1])
235# # self.static_interface.time_stats.text = "{:02}".format(0.0) + "".join(self.static_interface.formatted_units[2])
236
237# # async def populate_in_types(self):
238# # """Builds the input types selection."""
239
240# # in_edge_names = await self.model_source.show_edges()
241# # self.input_types.items = in_edge_names
242
243# # async def populate_out_types(self):
244# # """Builds the output types selection."""
245
246# # out_edges = await self.model_source.show_edges(target=True)
247# # self.output_types.items = out_edges
248
249# # async def populate_model_stack(self, widget: Optional[Callable] = None):
250# # """Builds the model stack selection dropdown."""
251
252# # await self.model_source.clear()
253# # if self.input_types.value and self.output_types.value:
254# # models = await self.model_source.trace_models(self.input_types.value, self.output_types.value)
255# # self.model_stack.items = models # [model[0][:20] for model in models if len(model[0]) > 20]
256
257# # async def populate_task_stack(self, widget: Optional[Callable] = None):
258# # """Builds the task stack selection dropdown."""
259# # selection = self.model_stack.value
260# # registry_entry = next(
261# # iter(
262# # registry["entry"] # formatting
263# # for registry in self.model_source._graph.registry_entries # formatting
264# # if selection in registry["entry"].model
265# # )
266# # )
267# # await self.task_source.set_filter_type(self.input_types.value, self.output_types.value)
268# # tasks = await self.task_source.trace_tasks(registry_entry)
269
270# # self.task_stack.items = tasks
271
272# # async def switch_tabs(self, widget: Optional[Callable] = None):
273# # self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
274# # self.final_layout.style.background_color = self.bg
275# # self.final_layout.refresh()
276# # self.browser_panel.evaluate_javascript("location.reload();")