When configuring the input, a default input with valid values is gene…

…rated and the input file may have multiple inputs defined.

Also added support for using pydev to run python (and not just microsoft.python).

sema4ai-python-ls-core/src/sema4ai_ls_core/basic.py

@@ -18,16 +18,17 @@
 import os
 import sys
 import threading
+from collections.abc import Callable
 from concurrent.futures import Future
 from contextlib import contextmanager
 from dataclasses import dataclass
 from functools import lru_cache
-from typing import Any, Tuple, TypeVar
-from collections.abc import Callable
+from typing import Any, Optional, TypeVar
 
 from sema4ai_ls_core.core_log import get_logger
 from sema4ai_ls_core.jsonrpc.exceptions import JsonRpcRequestCancelled
 from sema4ai_ls_core.options import DEFAULT_TIMEOUT
+from sema4ai_ls_core.protocols import IMonitor
 
 PARENT_PROCESS_WATCH_INTERVAL = 3  # 3 s
 
@@ -395,7 +396,12 @@ class ProcessRunResult:
 
 
 def launch_and_return_future(
-    cmd, environ, cwd, timeout=100, stdin: bytes = b"\n"
+    cmd,
+    environ,
+    cwd,
+    timeout=100,
+    stdin: bytes = b"\n",
+    monitor: Optional[IMonitor] = None,
 ) -> "Future[ProcessRunResult]":
     import subprocess
 
@@ -469,6 +475,14 @@ def report_output():
         except BaseException as e:
             future.set_exception(e)
 
+    if monitor is not None:
+
+        def on_monitor_cancelled():
+            if process.poll() is None:  # i.e.: still running.
+                process.kill()
+
+        monitor.add_listener(on_monitor_cancelled)
+
     threading.Thread(target=report_output, daemon=True).start()
     return future
 

sema4ai-python-ls-core/src/sema4ai_ls_core/jsonrpc/monitor.py

@@ -1,5 +1,5 @@
-from sema4ai_ls_core.protocols import IMonitor, IMonitorListener
 from sema4ai_ls_core.core_log import get_logger
+from sema4ai_ls_core.protocols import IMonitor, IMonitorListener
 
 log = get_logger(__name__)
 
@@ -10,7 +10,7 @@ def __init__(self, title: str = ""):
         self._cancelled: bool = False
         self._listeners: tuple[IMonitorListener, ...] = ()
 
-    def add_listener(self, listener):
+    def add_listener(self, listener: IMonitorListener):
         if self._cancelled:
             listener()
         else:

sema4ai-python-ls-core/src/sema4ai_ls_core/protocols.py

@@ -1,19 +1,9 @@
 import sys
 import threading
 import typing
-from enum import Enum
-from typing import (
-    Any,
-    Dict,
-    Generic,
-    List,
-    Optional,
-    Tuple,
-    Type,
-    TypeVar,
-    Union,
-)
 from collections.abc import Callable, Iterable, Mapping
+from enum import Enum
+from typing import Any, Dict, Generic, List, Optional, Tuple, Type, TypeVar, Union
 
 if typing.TYPE_CHECKING:
     # This would lead to a circular import, so, do it only when type-checking.
@@ -909,7 +899,9 @@ def check_cancelled(self) -> None:
         """
 
     def add_listener(self, listener: IMonitorListener):
-        pass
+        """
+        Adds a listener that'll be called when the monitor is cancelled.
+        """
 
 
 class ActionResultDict(TypedDict):

sema4ai-python-ls-core/src/sema4ai_ls_core/unittest_tools/fixtures.py

@@ -25,7 +25,7 @@ def wait_for_test_condition(condition, msg=None, timeout=TIMEOUT, sleep=1 / 20.0
 
 
 @pytest.fixture
-def ws_root_path(tmpdir):
+def ws_root_path(tmpdir) -> str:
     return str(tmpdir.join("root"))
 
 

sema4ai/pyproject.toml

@@ -51,7 +51,7 @@ sema4ai-python-ls-core = { path = "../sema4ai-python-ls-core/", develop = true }
 fire = "*"
 
 robocorp-actions = "^0.2.1" # Just needed for testing.
-sema4ai-actions = "^0.10.0" # Just needed for testing.
+sema4ai-actions = "^1.0.1" # Just needed for testing.
 
 numpy = "<2"
 ruff = "^0.6.5"

sema4ai/src/sema4ai_code/compute_launch.py

@@ -1,5 +1,5 @@
 import os
-from typing import Any, Dict, List, Optional
+from typing import Any
 
 from sema4ai_ls_core import uris
 from sema4ai_ls_core.core_log import get_logger

sema4ai/src/sema4ai_code/robo/actions_form_data.py

@@ -0,0 +1,275 @@
+from enum import Enum
+from typing import Any, Dict, List, Union
+
+
+class InputPropertyType(Enum):
+    STRING = "string"
+    INTEGER = "integer"
+    NUMBER = "number"
+    BOOLEAN = "boolean"
+    ARRAY = "array"
+    OBJECT = "object"
+    ENUM = "enum"
+
+
+class InputProperty:
+    def __init__(
+        self,
+        title: str,
+        description: str,
+        type: InputPropertyType,
+        enum: List[str] | None = None,
+    ):
+        self.title = title
+        self.description = description
+        self.type = type
+        self.enum = enum
+
+
+PropertyFormDataType = Union[str, int, float, bool, List["PropertyFormData"]]
+
+
+class PropertyFormData:
+    def __init__(
+        self,
+        name: str,
+        prop: InputProperty,
+        required: bool,
+        value: PropertyFormDataType,
+        title: str | None = None,
+        options: List[str] | None = None,
+    ):
+        self.name = name
+        self.prop = prop
+        self.required = required
+        self.value = value
+        self.title = title
+        self.options = options
+
+    def value_as_str(self) -> str:
+        if self.prop.enum:
+            if self.prop.description:
+                return f"{self.prop.enum} ({self.prop.description})"
+            return f"{self.prop.enum}"
+
+        if self.prop.description:
+            return f"{self.prop.type.value}: {self.prop.description}"
+
+        return f"{self.prop.type.value}"
+
+    def __str__(self):
+        return f"{self.name} [{self.value_as_str()}]"
+
+    __repr__ = __str__
+
+
+def get_default_value(
+    property_type: str, enum: List[str] | None = None
+) -> PropertyFormDataType:
+    if property_type == "number":
+        return 0.0
+    elif property_type == "boolean":
+        return False
+    elif property_type == "integer":
+        return 0
+    elif property_type == "array":
+        return "[]"
+    elif property_type == "object":
+        return "{}"
+    else:
+        if enum:
+            return enum[0]
+        return ""
+
+
+def set_array_item_title(item: PropertyFormData) -> None:
+    new_title = item.prop.title
+    if new_title.endswith("*"):
+        new_title = new_title[:-1]
+    if not new_title.endswith(" (item)"):
+        new_title += " (item)"
+    item.prop.title = new_title
+
+
+def properties_to_form_data(
+    schema: Dict[str, Any], parents: List[str] | None = None
+) -> List[PropertyFormData]:
+    if parents is None:
+        parents = []
+
+    if "properties" not in schema:
+        return []
+
+    entries = []
+    for name, prop in schema["properties"].items():
+        property_name = parents + [name]
+
+        if "$ref" in prop:
+            continue
+
+        if "properties" in prop:
+            entries.extend(properties_to_form_data(prop, property_name))
+            continue
+
+        if "allOf" in prop:
+            first_child = prop["allOf"][0]
+            if "enum" in first_child:
+                entry = PropertyFormData(
+                    name=".".join(property_name),
+                    prop=InputProperty(
+                        title=prop.get("title", property_name[-1]),
+                        description=prop.get("description", ""),
+                        type=InputPropertyType.ENUM,
+                    ),
+                    required=name in schema.get("required", []),
+                    value=prop.get(
+                        "default", first_child["enum"][0] if first_child["enum"] else ""
+                    ),
+                    options=first_child["enum"],
+                )
+                entries.append(entry)
+            else:
+                for item in prop["allOf"]:
+                    entries.extend(properties_to_form_data(item, property_name))
+            continue
+
+        if isinstance(prop.get("type"), list) or prop.get("type") == "array":
+            row_entry = PropertyFormData(
+                name=".".join(property_name),
+                title=prop.get("title", property_name[-1]),
+                prop=InputProperty(
+                    title=prop.get("title", property_name[-1]),
+                    description=prop.get("description", ""),
+                    type=InputPropertyType.ARRAY,
+                ),
+                required=name in schema.get("required", []),
+                value=get_default_value(prop.get("type", "string")),
+            )
+
+            if "items" in prop:
+                if "properties" in prop["items"]:
+                    row_properties = properties_to_form_data(
+                        prop["items"], property_name + ["0"]
+                    )
+                    row_entry.value = row_properties
+                    entries.append(row_entry)
+                    entries.extend(row_properties)
+                else:
+                    enum = prop["items"].get("enum")
+                    row_property = PropertyFormData(
+                        name=f"{'.'.join(property_name)}.0",
+                        prop=InputProperty(
+                            title=prop.get("title", property_name[-1]),
+                            description=prop.get("description", ""),
+                            type=InputPropertyType(prop["items"].get("type", "string")),
+                            enum=enum,
+                        ),
+                        required=name in schema.get("required", []),
+                        value=get_default_value(
+                            prop["items"].get("type", "string"), enum
+                        ),
+                    )
+                    row_entry.value = [row_property]
+                    set_array_item_title(row_property)
+                    entries.extend([row_entry, row_property])
+            else:
+                entries.append(row_entry)
+            continue
+
+        enum = prop.get("enum")
+        entry = PropertyFormData(
+            name=".".join(property_name),
+            prop=InputProperty(
+                title=prop.get("title", property_name[-1]),
+                description=prop.get("description", ""),
+                type=InputPropertyType(prop.get("type", "string")),
+                enum=enum,
+            ),
+            required=name in schema.get("required", []),
+            value=get_default_value(prop.get("type", "string"), enum),
+        )
+
+        if schema.get("title") and entries:
+            entry.title = schema["title"]
+
+        entries.append(entry)
+
+    return entries
+
+
+Payload = Dict[str, Any]
+
+
+def form_data_to_payload(data: List[PropertyFormData]) -> Payload:
+    result: Payload = {}
+
+    for item in data:
+        levels = item.name.split(".")
+        property_name = levels[-1]
+
+        current_level = result
+        for level in levels[:-1]:
+            if level not in current_level:
+                current_level[level] = {}
+            current_level = current_level[level]
+
+        if item.prop.type == InputPropertyType.OBJECT:
+            current_level[property_name] = eval(str(item.value))
+        elif item.prop.type == InputPropertyType.ARRAY:
+            if property_name not in current_level:
+                current_level[property_name] = []
+        elif isinstance(current_level, list):
+            current_level.append(item.value)
+        else:
+            current_level[property_name] = item.value
+
+    return result
+
+
+def payload_to_form_data(
+    payload: Payload, form_data: List[PropertyFormData], path: str = ""
+) -> List[PropertyFormData]:
+    result: List[PropertyFormData] = []
+
+    for key, val in payload.items():
+        full_path = f"{path}.{key}" if path else key
+        if isinstance(val, dict):
+            result.extend(payload_to_form_data(val, form_data, full_path))
+        elif isinstance(val, list):
+            found_data = next(
+                (elem for elem in form_data if elem.name == full_path), None
+            )
+            if found_data:
+                result.append(found_data)
+            for index, elem_value in enumerate(val):
+                found_elem = next(
+                    (elem for elem in form_data if elem.name == f"{full_path}.{index}"),
+                    None,
+                )
+                if found_elem:
+                    result.append(
+                        PropertyFormData(**{**found_elem.__dict__, "value": elem_value})
+                    )
+                else:
+                    prev = next(
+                        (elem for elem in form_data if elem.name == f"{full_path}.0"),
+                        None,
+                    )
+                    if prev:
+                        result.append(
+                            PropertyFormData(
+                                **{
+                                    **prev.__dict__,
+                                    "value": elem_value,
+                                    "name": f"{full_path}.{index}",
+                                }
+                            )
+                        )
+        else:
+            found_data = next(
+                (elem for elem in form_data if elem.name == full_path), None
+            )
+            if found_data:
+                result.append(PropertyFormData(**{**found_data.__dict__, "value": val}))
+
+    return result