zodiac.toga.palette 

  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
  4from typing import Callable
  5import toga
  6
  7
  8class CommandPalette:
  9    async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> toga.Widget:
 10        """Process and synthesize input data based on selected model.\n
 11        :param widget: The UI widget that triggered this action, typically used for state management.\n
 12        :type widget: toga.widgets
 13        :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
 14        :type external: bool"""
 15        from zodiac.toga.signatures import ready_predictor
 16
 17        self.response_panel.value += f"{os.path.basename(self.registry_entry.model)} :\n"
 18
 19        await self.token_stream.set_tokenizer(self.registry_entry)
 20        prompts = {}
 21        if self.message_panel.value:
 22            cache = False
 23            prompts.setdefault("text", self.message_panel.value)
 24        stream = True if self.output_types.value == "text" else False
 25        # prompts.setdefault("audio",[0]) if else []
 26        # prompts.setdefault("image",[]) if image": []
 27
 28        if stream:
 29            context_data, predictor_data = await ready_predictor(self.registry_entry, dspy_stream=stream, async_stream=stream, cache=cache)
 30            await self.stream_text(prompts, context_data, predictor_data)
 31        else:
 32            await self.generate_media(prompts, self.registry_entry)  # context_data, predictor_data)
 33        return widget
 34
 35    async def stream_text(self, prompts, context_data, predictor_data):
 36        from zodiac.toga.signatures import Predictor
 37        from litellm.types.utils import ModelResponseStream  # StatusStreamingCallback
 38        from dspy.streaming import StatusMessage, StreamResponse
 39
 40        self.response_panel.scroll_to_bottom()
 41        with dspy_context(**context_data):
 42            self.program = streamify(Predictor(), **predictor_data)
 43            async for prediction in self.program(question=prompts["text"]):
 44                if isinstance(prediction, ModelResponseStream) and prediction["choices"][0]["delta"]["content"]:
 45                    self.response_panel.value += prediction["choices"][0]["delta"]["content"]
 46                elif isinstance(prediction, StreamResponse) or hasattr(prediction, "chunk"):
 47                    self.response_panel.value += str(prediction.chunk)
 48                elif isinstance(prediction, Prediction) or hasattr(prediction, "answer"):
 49                    self.response_panel.value += str(prediction.answer)
 50                elif isinstance(prediction, StatusMessage) or hasattr(prediction, "message"):
 51                    self.status_display.text = self.status_text_prefix + str(prediction.message)
 52        self.response_panel.value += "\n--\n\n"
 53        return prediction
 54
 55    async def generate_media(self, prompts, registry_entry) -> None:  # , predictor_data
 56        from nnll.tensor_pipe.construct_pipe import ConstructPipeline
 57        from nnll.tensor_pipe.inference import run_inference
 58        from zodiac.streams.class_stream import best_package
 59        from zodiac.providers.constants import MIR_DB
 60
 61        pkg_data = await best_package(pkg_data=registry_entry)
 62        constructor = ConstructPipeline()
 63        pipe_data = constructor.create_pipeline(registry_entry, pkg_data, MIR_DB)
 64        return run_inference(pipe_data, prompts)
 65
 66        # from zodiac.toga.signatures import Predictor
 67
 68        # return prediction
 69
 70    async def halt(self, widget, **kwargs) -> None:
 71        """Stop processing prompt\n
 72        :param widget: The calling widget object"""
 73        if not self.program.done():
 74            import gc
 75
 76            del self.program
 77            gc.collect()
 78            self.status_display.text = self.status_text_prefix + "Cancelled."
 79
 80    async def empty_prompt(self, widget, **kwargs) -> None:
 81        """Clears the prompt input area.
 82        :param widget: Triggering widget"""
 83        self.message_panel.value = ""
 84
 85    async def copy_reply(self, widget, **kwargs) -> None:
 86        """Push the reply into the clipboard
 87        :param widget: Triggering widget"""
 88        import pyperclip
 89
 90        pyperclip.copy(self.response_panel.value)
 91        self.status_display.text = self.status_text_prefix + self.status_info[7]
 92
 93    async def attach_file(self, widget, **kwargs) -> None:
 94        """Attaches a file's contents to the prompt area.
 95        :param widget: Triggering widget"""
 96        import json
 97
 98        try:
 99            file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
100            self.status_display.text = f"Read. {file_path_named}"
101            if file_path_named is not None:
102                from nnll.metadata.json_io import read_json_file
103
104                file_contents = read_json_file(file_path_named)
105                self.message_panel.scroll_to_bottom()
106                self.message_panel.value = json.dumps(file_contents)
107                self.status_display.text = self.status_text_prefix + self.status_info[6]
108            else:
109                self.status_display.text = self.status_text_prefix + self.status_info[4]
110        except (ValueError, json.JSONDecodeError):
111            self.status_display.text = self.status_text_prefix + self.status_info[5]
112
113    async def reset_position(self, widget, **kwargs) -> None:
114        """Scrolls text panel to bottom after content update.
115        :param widget: text panel widget
116        """
117        setattr(self, "position_counter", getattr(self, "position_counter", 0) + 1)
118        if max(self.scroll_buffer, self.position_counter) >= self.scroll_buffer:
119            self.position_counter = 0
120            widget.scroll_to_bottom()
121
122    async def on_select_handler(self, widget, **kwargs) -> None:
123        """React to input/output choice\n
124        :param widget: The widget that triggered the event."""
125        selection = widget.value
126        if self.model_stream._graph.registry_entries is not None:
127            self.registry_entry = next(iter(registry["entry"] for registry in self.model_stream._graph.registry_entries if selection in registry["entry"].model))
128            await self.populate_task_stack()
129            await self.token_stream.set_tokenizer(self.registry_entry)
130        else:
131            self.registry_entry = "No model..."
132
133    async def model_graph(self):
134        """Builds the model graph."""
135        await self.model_stream.model_graph()
136
137    async def populate_in_types(self) -> None:
138        """Builds the input types selection."""
139        in_edge_names = await self.model_stream.show_edges()
140        self.input_types.items = in_edge_names
141
142    async def populate_out_types(self) -> None:
143        """Builds the output types selection."""
144        out_edges = await self.model_stream.show_edges(target=True)
145        self.output_types.items = out_edges
146
147    async def populate_model_stack(self, widget: toga.Widget = None, **kwargs) -> None:
148        """Builds the model stack selection dropdown."""
149
150        await self.model_stream.clear()
151        if self.input_types.value and self.output_types.value:
152            models = await self.model_stream.trace_models(self.input_types.value, self.output_types.value)
153            self.model_stack.items = models  # [model[0][:20] for model in models if len(model[0]) > 20]
154            await self.token_estimate(widget=self.message_panel)
155
156    async def populate_task_stack(self, widget: toga.Widget = None, **kwargs) -> None:
157        """Builds the task stack selection dropdown."""
158        selection = self.model_stack.value
159        if self.model_stream._graph.registry_entries:
160            registry_entry = next(
161                iter(
162                    registry["entry"]  # formatting
163                    for registry in self.model_stream._graph.registry_entries  # formatting
164                    if selection in registry["entry"].model
165                )
166            )
167        else:
168            registry_entry = "No models..."
169        await self.task_stream.set_filter_type(self.input_types.value, self.output_types.value)
170        if registry_entry and not isinstance(registry_entry, str):
171            tasks = await self.task_stream.filter_tasks(registry_entry)
172        else:
173            tasks = ""
174
175        self.task_stack.items = tasks
176
177    async def switch_tabs(self, widget: toga.Widget = None, **kwargs) -> None:
178        """Switches between text and graph tabs.
179        :param widget: The triggering widget (optional), defaults to None"""
180        self.browser_panel.evaluate_javascript("location.reload();")
181        self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
182        self.final_layout.style.background_color = self.bg
183        self.final_layout.refresh()
184        self.status_display.text += self.status_info[0]
185        await self.ping_server(widget=self.status_display)
186
187    async def ping_server(self, widget: toga.Widget, **kwargs) -> toga.Widget:
188        self.browser_panel.url = self.graph_server
189        try:
190            request = requests.get(self.graph_server, timeout=(3, 3))
191            if request is not None:
192                if hasattr(request, "status_code"):
193                    status = request.status_code
194                if (hasattr(request, "ok") and request.ok) or (hasattr(request, "reason") and request.reason == "OK"):
195                    await self.active_server()
196                elif hasattr(request, "json"):
197                    status = request.json()
198                    if status.get("result") == "OK":
199                        await self.active_server()
200                else:
201                    self.browser_panel.url = self.graph_disabled
202                    await self.active_server(False)
203            else:
204                self.browser_panel.url = self.graph_disabled
205                await self.active_server(False)
206        except (ConnectTimeout, ConnectionError, ConnectionRefusedError, MaxRetryError, NewConnectionError, OSError):
207            await self.active_server(False)
208            pass
209        return widget
210
211    async def active_server(self, enabled: bool = True):
212        if not enabled:
213            status_info = self.status_info[1]
214            self.browser_panel.url = self.graph_disabled
215        else:
216            status_info = self.status_info[2]
217            self.browser_panel.url = self.graph_server
218        for info in self.status_info:
219            self.status_display.text = self.status_display.text.replace(info, "")
220        self.status_display.text += status_info
class CommandPalette: 
  9class CommandPalette:
 10    async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> toga.Widget:
 11        """Process and synthesize input data based on selected model.\n
 12        :param widget: The UI widget that triggered this action, typically used for state management.\n
 13        :type widget: toga.widgets
 14        :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
 15        :type external: bool"""
 16        from zodiac.toga.signatures import ready_predictor
 17
 18        self.response_panel.value += f"{os.path.basename(self.registry_entry.model)} :\n"
 19
 20        await self.token_stream.set_tokenizer(self.registry_entry)
 21        prompts = {}
 22        if self.message_panel.value:
 23            cache = False
 24            prompts.setdefault("text", self.message_panel.value)
 25        stream = True if self.output_types.value == "text" else False
 26        # prompts.setdefault("audio",[0]) if else []
 27        # prompts.setdefault("image",[]) if image": []
 28
 29        if stream:
 30            context_data, predictor_data = await ready_predictor(self.registry_entry, dspy_stream=stream, async_stream=stream, cache=cache)
 31            await self.stream_text(prompts, context_data, predictor_data)
 32        else:
 33            await self.generate_media(prompts, self.registry_entry)  # context_data, predictor_data)
 34        return widget
 35
 36    async def stream_text(self, prompts, context_data, predictor_data):
 37        from zodiac.toga.signatures import Predictor
 38        from litellm.types.utils import ModelResponseStream  # StatusStreamingCallback
 39        from dspy.streaming import StatusMessage, StreamResponse
 40
 41        self.response_panel.scroll_to_bottom()
 42        with dspy_context(**context_data):
 43            self.program = streamify(Predictor(), **predictor_data)
 44            async for prediction in self.program(question=prompts["text"]):
 45                if isinstance(prediction, ModelResponseStream) and prediction["choices"][0]["delta"]["content"]:
 46                    self.response_panel.value += prediction["choices"][0]["delta"]["content"]
 47                elif isinstance(prediction, StreamResponse) or hasattr(prediction, "chunk"):
 48                    self.response_panel.value += str(prediction.chunk)
 49                elif isinstance(prediction, Prediction) or hasattr(prediction, "answer"):
 50                    self.response_panel.value += str(prediction.answer)
 51                elif isinstance(prediction, StatusMessage) or hasattr(prediction, "message"):
 52                    self.status_display.text = self.status_text_prefix + str(prediction.message)
 53        self.response_panel.value += "\n--\n\n"
 54        return prediction
 55
 56    async def generate_media(self, prompts, registry_entry) -> None:  # , predictor_data
 57        from nnll.tensor_pipe.construct_pipe import ConstructPipeline
 58        from nnll.tensor_pipe.inference import run_inference
 59        from zodiac.streams.class_stream import best_package
 60        from zodiac.providers.constants import MIR_DB
 61
 62        pkg_data = await best_package(pkg_data=registry_entry)
 63        constructor = ConstructPipeline()
 64        pipe_data = constructor.create_pipeline(registry_entry, pkg_data, MIR_DB)
 65        return run_inference(pipe_data, prompts)
 66
 67        # from zodiac.toga.signatures import Predictor
 68
 69        # return prediction
 70
 71    async def halt(self, widget, **kwargs) -> None:
 72        """Stop processing prompt\n
 73        :param widget: The calling widget object"""
 74        if not self.program.done():
 75            import gc
 76
 77            del self.program
 78            gc.collect()
 79            self.status_display.text = self.status_text_prefix + "Cancelled."
 80
 81    async def empty_prompt(self, widget, **kwargs) -> None:
 82        """Clears the prompt input area.
 83        :param widget: Triggering widget"""
 84        self.message_panel.value = ""
 85
 86    async def copy_reply(self, widget, **kwargs) -> None:
 87        """Push the reply into the clipboard
 88        :param widget: Triggering widget"""
 89        import pyperclip
 90
 91        pyperclip.copy(self.response_panel.value)
 92        self.status_display.text = self.status_text_prefix + self.status_info[7]
 93
 94    async def attach_file(self, widget, **kwargs) -> None:
 95        """Attaches a file's contents to the prompt area.
 96        :param widget: Triggering widget"""
 97        import json
 98
 99        try:
100            file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
101            self.status_display.text = f"Read. {file_path_named}"
102            if file_path_named is not None:
103                from nnll.metadata.json_io import read_json_file
104
105                file_contents = read_json_file(file_path_named)
106                self.message_panel.scroll_to_bottom()
107                self.message_panel.value = json.dumps(file_contents)
108                self.status_display.text = self.status_text_prefix + self.status_info[6]
109            else:
110                self.status_display.text = self.status_text_prefix + self.status_info[4]
111        except (ValueError, json.JSONDecodeError):
112            self.status_display.text = self.status_text_prefix + self.status_info[5]
113
114    async def reset_position(self, widget, **kwargs) -> None:
115        """Scrolls text panel to bottom after content update.
116        :param widget: text panel widget
117        """
118        setattr(self, "position_counter", getattr(self, "position_counter", 0) + 1)
119        if max(self.scroll_buffer, self.position_counter) >= self.scroll_buffer:
120            self.position_counter = 0
121            widget.scroll_to_bottom()
122
123    async def on_select_handler(self, widget, **kwargs) -> None:
124        """React to input/output choice\n
125        :param widget: The widget that triggered the event."""
126        selection = widget.value
127        if self.model_stream._graph.registry_entries is not None:
128            self.registry_entry = next(iter(registry["entry"] for registry in self.model_stream._graph.registry_entries if selection in registry["entry"].model))
129            await self.populate_task_stack()
130            await self.token_stream.set_tokenizer(self.registry_entry)
131        else:
132            self.registry_entry = "No model..."
133
134    async def model_graph(self):
135        """Builds the model graph."""
136        await self.model_stream.model_graph()
137
138    async def populate_in_types(self) -> None:
139        """Builds the input types selection."""
140        in_edge_names = await self.model_stream.show_edges()
141        self.input_types.items = in_edge_names
142
143    async def populate_out_types(self) -> None:
144        """Builds the output types selection."""
145        out_edges = await self.model_stream.show_edges(target=True)
146        self.output_types.items = out_edges
147
148    async def populate_model_stack(self, widget: toga.Widget = None, **kwargs) -> None:
149        """Builds the model stack selection dropdown."""
150
151        await self.model_stream.clear()
152        if self.input_types.value and self.output_types.value:
153            models = await self.model_stream.trace_models(self.input_types.value, self.output_types.value)
154            self.model_stack.items = models  # [model[0][:20] for model in models if len(model[0]) > 20]
155            await self.token_estimate(widget=self.message_panel)
156
157    async def populate_task_stack(self, widget: toga.Widget = None, **kwargs) -> None:
158        """Builds the task stack selection dropdown."""
159        selection = self.model_stack.value
160        if self.model_stream._graph.registry_entries:
161            registry_entry = next(
162                iter(
163                    registry["entry"]  # formatting
164                    for registry in self.model_stream._graph.registry_entries  # formatting
165                    if selection in registry["entry"].model
166                )
167            )
168        else:
169            registry_entry = "No models..."
170        await self.task_stream.set_filter_type(self.input_types.value, self.output_types.value)
171        if registry_entry and not isinstance(registry_entry, str):
172            tasks = await self.task_stream.filter_tasks(registry_entry)
173        else:
174            tasks = ""
175
176        self.task_stack.items = tasks
177
178    async def switch_tabs(self, widget: toga.Widget = None, **kwargs) -> None:
179        """Switches between text and graph tabs.
180        :param widget: The triggering widget (optional), defaults to None"""
181        self.browser_panel.evaluate_javascript("location.reload();")
182        self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
183        self.final_layout.style.background_color = self.bg
184        self.final_layout.refresh()
185        self.status_display.text += self.status_info[0]
186        await self.ping_server(widget=self.status_display)
187
188    async def ping_server(self, widget: toga.Widget, **kwargs) -> toga.Widget:
189        self.browser_panel.url = self.graph_server
190        try:
191            request = requests.get(self.graph_server, timeout=(3, 3))
192            if request is not None:
193                if hasattr(request, "status_code"):
194                    status = request.status_code
195                if (hasattr(request, "ok") and request.ok) or (hasattr(request, "reason") and request.reason == "OK"):
196                    await self.active_server()
197                elif hasattr(request, "json"):
198                    status = request.json()
199                    if status.get("result") == "OK":
200                        await self.active_server()
201                else:
202                    self.browser_panel.url = self.graph_disabled
203                    await self.active_server(False)
204            else:
205                self.browser_panel.url = self.graph_disabled
206                await self.active_server(False)
207        except (ConnectTimeout, ConnectionError, ConnectionRefusedError, MaxRetryError, NewConnectionError, OSError):
208            await self.active_server(False)
209            pass
210        return widget
211
212    async def active_server(self, enabled: bool = True):
213        if not enabled:
214            status_info = self.status_info[1]
215            self.browser_panel.url = self.graph_disabled
216        else:
217            status_info = self.status_info[2]
218            self.browser_panel.url = self.graph_server
219        for info in self.status_info:
220            self.status_display.text = self.status_display.text.replace(info, "")
221        self.status_display.text += status_info
async def ticker( self, widget: Callable, external: bool = False, **kwargs) -> toga.widgets.base.Widget: 
10    async def ticker(self, widget: Callable, external: bool = False, **kwargs) -> toga.Widget:
11        """Process and synthesize input data based on selected model.\n
12        :param widget: The UI widget that triggered this action, typically used for state management.\n
13        :type widget: toga.widgets
14        :param external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False
15        :type external: bool"""
16        from zodiac.toga.signatures import ready_predictor
17
18        self.response_panel.value += f"{os.path.basename(self.registry_entry.model)} :\n"
19
20        await self.token_stream.set_tokenizer(self.registry_entry)
21        prompts = {}
22        if self.message_panel.value:
23            cache = False
24            prompts.setdefault("text", self.message_panel.value)
25        stream = True if self.output_types.value == "text" else False
26        # prompts.setdefault("audio",[0]) if else []
27        # prompts.setdefault("image",[]) if image": []
28
29        if stream:
30            context_data, predictor_data = await ready_predictor(self.registry_entry, dspy_stream=stream, async_stream=stream, cache=cache)
31            await self.stream_text(prompts, context_data, predictor_data)
32        else:
33            await self.generate_media(prompts, self.registry_entry)  # context_data, predictor_data)
34        return widget

Process and synthesize input data based on selected model.

Parameters

  • widget: The UI widget that triggered this action, typically used for state management.

  • external: Indicates whether the processing should be handled externally (e.g., via clipboard), defaults to False

async def stream_text(self, prompts, context_data, predictor_data): 
36    async def stream_text(self, prompts, context_data, predictor_data):
37        from zodiac.toga.signatures import Predictor
38        from litellm.types.utils import ModelResponseStream  # StatusStreamingCallback
39        from dspy.streaming import StatusMessage, StreamResponse
40
41        self.response_panel.scroll_to_bottom()
42        with dspy_context(**context_data):
43            self.program = streamify(Predictor(), **predictor_data)
44            async for prediction in self.program(question=prompts["text"]):
45                if isinstance(prediction, ModelResponseStream) and prediction["choices"][0]["delta"]["content"]:
46                    self.response_panel.value += prediction["choices"][0]["delta"]["content"]
47                elif isinstance(prediction, StreamResponse) or hasattr(prediction, "chunk"):
48                    self.response_panel.value += str(prediction.chunk)
49                elif isinstance(prediction, Prediction) or hasattr(prediction, "answer"):
50                    self.response_panel.value += str(prediction.answer)
51                elif isinstance(prediction, StatusMessage) or hasattr(prediction, "message"):
52                    self.status_display.text = self.status_text_prefix + str(prediction.message)
53        self.response_panel.value += "\n--\n\n"
54        return prediction
async def generate_media(self, prompts, registry_entry) -> None: 
56    async def generate_media(self, prompts, registry_entry) -> None:  # , predictor_data
57        from nnll.tensor_pipe.construct_pipe import ConstructPipeline
58        from nnll.tensor_pipe.inference import run_inference
59        from zodiac.streams.class_stream import best_package
60        from zodiac.providers.constants import MIR_DB
61
62        pkg_data = await best_package(pkg_data=registry_entry)
63        constructor = ConstructPipeline()
64        pipe_data = constructor.create_pipeline(registry_entry, pkg_data, MIR_DB)
65        return run_inference(pipe_data, prompts)
66
67        # from zodiac.toga.signatures import Predictor
68
69        # return prediction
async def halt(self, widget, **kwargs) -> None: 
71    async def halt(self, widget, **kwargs) -> None:
72        """Stop processing prompt\n
73        :param widget: The calling widget object"""
74        if not self.program.done():
75            import gc
76
77            del self.program
78            gc.collect()
79            self.status_display.text = self.status_text_prefix + "Cancelled."

Stop processing prompt

Parameters
  • widget: The calling widget object
async def empty_prompt(self, widget, **kwargs) -> None: 
81    async def empty_prompt(self, widget, **kwargs) -> None:
82        """Clears the prompt input area.
83        :param widget: Triggering widget"""
84        self.message_panel.value = ""

Clears the prompt input area.

Parameters
  • widget: Triggering widget
async def copy_reply(self, widget, **kwargs) -> None: 
86    async def copy_reply(self, widget, **kwargs) -> None:
87        """Push the reply into the clipboard
88        :param widget: Triggering widget"""
89        import pyperclip
90
91        pyperclip.copy(self.response_panel.value)
92        self.status_display.text = self.status_text_prefix + self.status_info[7]

Push the reply into the clipboard

Parameters
  • widget: Triggering widget
async def attach_file(self, widget, **kwargs) -> None: 
 94    async def attach_file(self, widget, **kwargs) -> None:
 95        """Attaches a file's contents to the prompt area.
 96        :param widget: Triggering widget"""
 97        import json
 98
 99        try:
100            file_path_named = await self.main_window.dialog(toga.OpenFileDialog(title="Attach a file to the prompt"))
101            self.status_display.text = f"Read. {file_path_named}"
102            if file_path_named is not None:
103                from nnll.metadata.json_io import read_json_file
104
105                file_contents = read_json_file(file_path_named)
106                self.message_panel.scroll_to_bottom()
107                self.message_panel.value = json.dumps(file_contents)
108                self.status_display.text = self.status_text_prefix + self.status_info[6]
109            else:
110                self.status_display.text = self.status_text_prefix + self.status_info[4]
111        except (ValueError, json.JSONDecodeError):
112            self.status_display.text = self.status_text_prefix + self.status_info[5]

Attaches a file's contents to the prompt area.

Parameters
  • widget: Triggering widget
async def reset_position(self, widget, **kwargs) -> None: 
114    async def reset_position(self, widget, **kwargs) -> None:
115        """Scrolls text panel to bottom after content update.
116        :param widget: text panel widget
117        """
118        setattr(self, "position_counter", getattr(self, "position_counter", 0) + 1)
119        if max(self.scroll_buffer, self.position_counter) >= self.scroll_buffer:
120            self.position_counter = 0
121            widget.scroll_to_bottom()

Scrolls text panel to bottom after content update.

Parameters
  • widget: text panel widget
async def on_select_handler(self, widget, **kwargs) -> None: 
123    async def on_select_handler(self, widget, **kwargs) -> None:
124        """React to input/output choice\n
125        :param widget: The widget that triggered the event."""
126        selection = widget.value
127        if self.model_stream._graph.registry_entries is not None:
128            self.registry_entry = next(iter(registry["entry"] for registry in self.model_stream._graph.registry_entries if selection in registry["entry"].model))
129            await self.populate_task_stack()
130            await self.token_stream.set_tokenizer(self.registry_entry)
131        else:
132            self.registry_entry = "No model..."

React to input/output choice

Parameters
  • widget: The widget that triggered the event.
async def model_graph(self): 
134    async def model_graph(self):
135        """Builds the model graph."""
136        await self.model_stream.model_graph()

Builds the model graph.

async def populate_in_types(self) -> None: 
138    async def populate_in_types(self) -> None:
139        """Builds the input types selection."""
140        in_edge_names = await self.model_stream.show_edges()
141        self.input_types.items = in_edge_names

Builds the input types selection.

async def populate_out_types(self) -> None: 
143    async def populate_out_types(self) -> None:
144        """Builds the output types selection."""
145        out_edges = await self.model_stream.show_edges(target=True)
146        self.output_types.items = out_edges

Builds the output types selection.

async def populate_model_stack(self, widget: toga.widgets.base.Widget = None, **kwargs) -> None: 
148    async def populate_model_stack(self, widget: toga.Widget = None, **kwargs) -> None:
149        """Builds the model stack selection dropdown."""
150
151        await self.model_stream.clear()
152        if self.input_types.value and self.output_types.value:
153            models = await self.model_stream.trace_models(self.input_types.value, self.output_types.value)
154            self.model_stack.items = models  # [model[0][:20] for model in models if len(model[0]) > 20]
155            await self.token_estimate(widget=self.message_panel)

Builds the model stack selection dropdown.

async def populate_task_stack(self, widget: toga.widgets.base.Widget = None, **kwargs) -> None: 
157    async def populate_task_stack(self, widget: toga.Widget = None, **kwargs) -> None:
158        """Builds the task stack selection dropdown."""
159        selection = self.model_stack.value
160        if self.model_stream._graph.registry_entries:
161            registry_entry = next(
162                iter(
163                    registry["entry"]  # formatting
164                    for registry in self.model_stream._graph.registry_entries  # formatting
165                    if selection in registry["entry"].model
166                )
167            )
168        else:
169            registry_entry = "No models..."
170        await self.task_stream.set_filter_type(self.input_types.value, self.output_types.value)
171        if registry_entry and not isinstance(registry_entry, str):
172            tasks = await self.task_stream.filter_tasks(registry_entry)
173        else:
174            tasks = ""
175
176        self.task_stack.items = tasks

Builds the task stack selection dropdown.

async def switch_tabs(self, widget: toga.widgets.base.Widget = None, **kwargs) -> None: 
178    async def switch_tabs(self, widget: toga.Widget = None, **kwargs) -> None:
179        """Switches between text and graph tabs.
180        :param widget: The triggering widget (optional), defaults to None"""
181        self.browser_panel.evaluate_javascript("location.reload();")
182        self.bg = self.bg_graph if self.bg == self.bg_text else self.bg_text
183        self.final_layout.style.background_color = self.bg
184        self.final_layout.refresh()
185        self.status_display.text += self.status_info[0]
186        await self.ping_server(widget=self.status_display)

Switches between text and graph tabs.

Parameters
  • widget: The triggering widget (optional), defaults to None
async def ping_server( self, widget: toga.widgets.base.Widget, **kwargs) -> toga.widgets.base.Widget: 
188    async def ping_server(self, widget: toga.Widget, **kwargs) -> toga.Widget:
189        self.browser_panel.url = self.graph_server
190        try:
191            request = requests.get(self.graph_server, timeout=(3, 3))
192            if request is not None:
193                if hasattr(request, "status_code"):
194                    status = request.status_code
195                if (hasattr(request, "ok") and request.ok) or (hasattr(request, "reason") and request.reason == "OK"):
196                    await self.active_server()
197                elif hasattr(request, "json"):
198                    status = request.json()
199                    if status.get("result") == "OK":
200                        await self.active_server()
201                else:
202                    self.browser_panel.url = self.graph_disabled
203                    await self.active_server(False)
204            else:
205                self.browser_panel.url = self.graph_disabled
206                await self.active_server(False)
207        except (ConnectTimeout, ConnectionError, ConnectionRefusedError, MaxRetryError, NewConnectionError, OSError):
208            await self.active_server(False)
209            pass
210        return widget
async def active_server(self, enabled: bool = True): 
212    async def active_server(self, enabled: bool = True):
213        if not enabled:
214            status_info = self.status_info[1]
215            self.browser_panel.url = self.graph_disabled
216        else:
217            status_info = self.status_info[2]
218            self.browser_panel.url = self.graph_server
219        for info in self.status_info:
220            self.status_display.text = self.status_display.text.replace(info, "")
221        self.status_display.text += status_info