File size: 4,623 Bytes
a35fa4d 9f68e0d a35fa4d 9f68e0d a35fa4d 9f68e0d a35fa4d 9f68e0d a35fa4d 9f68e0d a35fa4d 9f68e0d a35fa4d 78dfff8 9f68e0d a35fa4d 9f68e0d 78dfff8 a35fa4d 78dfff8 a35fa4d 78dfff8 |
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 |
import re
from typing import List
import gradio as gr
from components import all_tasks, Input, MAX_INPUTS, MAX_TASKS, Task
def add_input(*visibility):
for i, visible in enumerate(visibility, 1):
if not bool(visible):
return (
[gr.Textbox.update(visible=True)] * i
+ [gr.Textbox.update(visible=False, value="")] * (MAX_INPUTS - i)
+ [1] * i
+ [0] * (MAX_INPUTS - i)
)
def remove_input(*visibility):
for i, visible in reversed(list(enumerate(visibility, 1))):
if bool(visible):
return (
[gr.Textbox.update(visible=True)] * (i - 1)
+ [gr.Textbox.update(visible=False, value="")] * (MAX_INPUTS - i + 1)
+ [1] * (i - 1)
+ [0] * (MAX_INPUTS - i + 1)
)
def _is_task_row_fully_invisible(row: List[int]) -> bool:
print(row)
for visible in row:
if bool(visible):
return False
return True
def add_task(index, *visibility):
visibility = list(visibility)
n_avail_tasks = len(Task.AVAILABLE_TASKS)
for i in range(MAX_TASKS):
start_row = i * n_avail_tasks
is_row_invisible = _is_task_row_fully_invisible(
visibility[start_row : start_row + n_avail_tasks]
)
if is_row_invisible:
unchanged_up_to = start_row + index
return (
[gr.Box.update()] * unchanged_up_to
+ [gr.Box.update(visible=True)]
+ [gr.Box.update()] * (len(visibility) - unchanged_up_to - 1)
+ [gr.Number.update()] * unchanged_up_to
+ [1]
+ [gr.Number.update()] * (len(visibility) - unchanged_up_to - 1)
)
def remove_task(*visibility):
visibility = list(visibility)
n_avail_tasks = len(Task.AVAILABLE_TASKS)
print(visibility, n_avail_tasks)
for i in range(MAX_TASKS):
start_row = i * n_avail_tasks
is_row_invisible = _is_task_row_fully_invisible(
visibility[start_row : start_row + n_avail_tasks]
)
print(start_row, start_row + n_avail_tasks, is_row_invisible)
if is_row_invisible:
unchanged_up_to = start_row - n_avail_tasks
return (
[gr.Box.update()] * unchanged_up_to
+ [gr.Box.update(visible=False)] * (len(visibility) - unchanged_up_to)
+ [gr.Number.update()] * unchanged_up_to
+ [0] * (len(visibility) - unchanged_up_to)
)
def _clear_error():
return gr.HighlightedText.update(value=None, visible=False)
def execute_task(id_: int, prev_error_value, n_inputs, *vars_in_scope):
"""
Params:
- id_: This will tell us which task to execute.
- prev_error_value: I carry around whether there is an error in the execution, to be displayed at the end.
- n_inputs: How many inputs does this task have?
- vars: All variables in scope. This can be a) task inputs, input varaibles or previous task outputs.
"""
n_inputs = int(n_inputs)
task_inputs = vars_in_scope[:n_inputs]
input_vars = vars_in_scope[n_inputs:MAX_INPUTS]
task_outputs = vars_in_scope[MAX_INPUTS:]
non_empty_task_inputs = [ti for ti in task_inputs if ti]
# Put all defined variables into a dict, with names (except task inputs)
vars = {
f"{Input.VNAME}{i}": input_ for i, input_ in enumerate(input_vars) if input_
}
vars.update({f"{Task.VNAME}{i}": task for i, task in enumerate(task_outputs)})
# Get all variables referenced within the task inputs
prompt_vars = {v for ti in non_empty_task_inputs for v in re.findall("{(.*?)}", ti)}
# If there is an undefined variable referenced, HighlightedText will signal the error.
undefined_vars = prompt_vars - vars.keys()
if len(undefined_vars) > 0:
return None, gr.HighlightedText.update(
value=[
(
f"The following variables are being used before being defined :: {undefined_vars}. Please check your tasks.",
"ERROR",
)
],
visible=True,
)
error_update = gr.HighlightedText.update(
value=prev_error_value, visible=prev_error_value is not None
)
if non_empty_task_inputs:
# Execute the task logic
return (
all_tasks[id_].execute(*non_empty_task_inputs, vars),
error_update,
)
else:
# There is no actionf or this task.
return None, error_update
|