supporting custom python code compatible with chaining widgets

Author: xtianpoli

ipyprogressivis/js/src/commands.js

@@ -93,7 +93,7 @@ export function setBackup(nbtracker, backupstring) {
   backupCell.model.sharedModel.setMetadata("progressivis_backup", backupstring);
 }
 
-export function createStageCells(nbtracker, tag, md, code) {
+export function createStageCells(nbtracker, tag, md, code, rw, run) {
   var crtWidget = nbtracker.currentWidget;
   var notebook = crtWidget.content;
   var tag = tag.toString();
@@ -128,9 +128,11 @@ export function createStageCells(nbtracker, tag, md, code) {
   });
   notebook.activeCellIndex = i + 1;
   var cell = notebook.widgets[i + 1];
-  NotebookActions.run(notebook, crtWidget.sessionContext);
+  if(run){
+      NotebookActions.run(notebook, crtWidget.sessionContext);
+  }
   cell.model.sharedModel.setMetadata("trusted", true);
-  cell.model.sharedModel.setMetadata("editable", false);
+  cell.model.sharedModel.setMetadata("editable", rw);
   cell.model.sharedModel.setMetadata("deletable", false);
   cell.model.sharedModel.setMetadata("progressivis_tag", tag);
 }

ipyprogressivis/js/src/labplugin.js

@@ -87,7 +87,7 @@ export const progressivisPlugin = {
       label: "Create stage cells",
       caption: "Create stage cells",
       execute: (args) => {
-        cmds.createStageCells(nbtracker, args.tag, args.md, args.code);
+          cmds.createStageCells(nbtracker, args.tag, args.md, args.code, args.rw, args.run);
       },
     });
     const TalkerView = class extends DOMWidgetView {

ipyprogressivis/widgets/chaining/__init__.py

@@ -13,6 +13,8 @@
 from .histogram import HistogramW
 from .iscaler import ScalerW
 from .any_vega import AnyVegaW
+from .code_cell import CodeCellW
+
 __all__ = [
     "Constructor",
     "DescStatsW",
@@ -27,5 +29,6 @@
     "ScalerW",
     "FacadeCreatorW",
     "HeatmapW",
-    "AnyVegaW"
+    "AnyVegaW",
+    "CodeCellW"
     ]

ipyprogressivis/widgets/chaining/code_cell.py

@@ -0,0 +1,13 @@
+from .utils import VBoxTyped, TypedBase, stage_register
+import ipywidgets as ipw
+
+
+class CodeCellW(VBoxTyped):
+    class Typed(TypedBase):
+        dongle: ipw.Label
+
+    def initialize(self) -> None:
+        self.c_.dongle = ipw.Label("Chaining ...")
+
+
+stage_register["Python"] = CodeCellW

ipyprogressivis/widgets/chaining/constructor.py

@@ -6,10 +6,11 @@
     make_button,
     set_dag,
     _Dag,
+    Proxy,
     DAGWidget,
     RootVBox,
     TypedBox,
-    NodeVBox,
+    NodeCarrier,
     TypedBase,
     get_widget_by_id,
     get_widget_by_key,
@@ -58,6 +59,7 @@ class Typed(TypedBase):
         record_ck: ipw.Checkbox
         csv: Optional[ipw.HBox]
         parquet: Optional[ipw.HBox]
+        custom: Optional[ipw.HBox]
         replay: ipw.Button
 
     last_created = None
@@ -126,6 +128,8 @@ def _start_scheduler_cb(self, btn: ipw.Button) -> None:
         self.child.csv = self.make_loader_box(ftype="csv", disabled=self._locked())
         self.child.parquet = self.make_loader_box(ftype="parquet",
                                                   disabled=self._locked())
+        self.child.custom = self.make_loader_box(ftype="custom",
+                                                 disabled=self._locked())
         self._arch_list = [b642json(elt)
                            for elt in bunpack(self._backup.value)
                            ] if self._backup.value else []
@@ -143,6 +147,8 @@ def _play_mode_cb(self, change: dict[str, AnyType]) -> None:
             wg.disabled = is_replay
         for wg in self.child.parquet.children:
             wg.disabled = is_replay
+        for wg in self.child.custom.children:
+            wg.disabled = is_replay
 
         if is_replay:
             self.child.record_ck.value = False
@@ -160,6 +166,7 @@ def _replay_cb(self, btn: ipw.Button) -> None:
         btn.disabled = True
         self.child.csv.children[-1].disabled = True
         self.child.parquet.children[-1].disabled = True
+        self.child.custom.children[-1].disabled = True
         self.child.record_ck.value = False
         self.child.record_ck.disabled = True
         self.child.play_mode_radio.disabled = True
@@ -170,13 +177,18 @@ def _replay_cb(self, btn: ipw.Button) -> None:
         replay_next(self)
 
     @staticmethod
-    def widget_by_id(key: int) -> NodeVBox:
+    def widget_by_id(key: int) -> NodeCarrier:
         return get_widget_by_id(key)
 
     @staticmethod
-    def widget(key: str, num: int = 0) -> NodeVBox:
+    def widget(key: str, num: int = 0) -> NodeCarrier:
         return get_widget_by_key(key, num)
 
+    @staticmethod
+    def proxy(key: str, num: int = 0) -> Proxy:
+        widget = get_widget_by_key(key, num)
+        return Proxy(widget)
+
     @property
     def dom_id(self) -> str:
         return f"prog_{id(self)}"

ipyprogressivis/widgets/chaining/custom_loader.py

@@ -0,0 +1,10 @@
+from .utils import VBoxTyped, TypedBase
+import ipywidgets as ipw
+
+
+class CustomLoaderW(VBoxTyped):
+    class Typed(TypedBase):
+        dongle: ipw.Label
+
+    def initialize(self) -> None:
+        self.c_.dongle = ipw.Label("Custom loader")

ipyprogressivis/widgets/chaining/utils.py

@@ -7,7 +7,7 @@
 import ipywidgets as ipw
 from progressivis.table.dshape import dataframe_dshape
 from progressivis.vis import DataShape
-from progressivis.core import Sink
+from progressivis.core import Sink, Scheduler
 from progressivis.core import Module
 from progressivis.table.table_facade import TableFacade
 from progressivis.core.utils import normalize_columns
@@ -69,6 +69,47 @@ class Header:
     modules_out: Sidecar
 
 
+class Proxy:
+    def __init__(self, carrier: "NodeCarrier") -> None:
+        self.__carrier = carrier
+
+    @property
+    def input_module(self) -> ModuleOrFacade | None:
+        if self.__carrier is PARAMS["constructor"]:
+            return None
+        return self.__carrier._input_module
+
+    @property
+    def input_slot(self) -> str | None:
+        if self.__carrier is PARAMS["constructor"]:
+            return None
+        return self.__carrier._input_slot
+
+    @property
+    def input_dtypes(self) -> dict[str, str] | None:
+        if self.__carrier is PARAMS["constructor"]:
+            return None
+        return self.__carrier._dtypes
+
+    @property
+    def scheduler(self) -> Scheduler:
+        if self.__carrier is PARAMS["constructor"]:
+            assert PARAMS["constructor"] is not None
+            return cast(Scheduler, PARAMS["constructor"].scheduler)
+        assert self.input_module is not None
+        return self.input_module.scheduler()
+
+    def resume(self, output_module: ModuleOrFacade,
+               output_slot: str = "result",
+               output_dtypes: dict[str, str] | None = None,
+               freeze: bool = False) -> "NodeCarrier":
+        self.__carrier._output_module = output_module
+        self.__carrier._output_slot = output_slot
+        self.__carrier._output_dtypes = output_dtypes
+        self.__carrier.make_chaining_box()
+        return self.__carrier
+
+
 def get_header() -> Header:
     """
     NB: call this function ONLY from the first cell of the notebook!!
@@ -357,11 +398,11 @@ def make_button(
 
 
 stage_register: Dict[str, AnyType] = {}
-parent_widget: Optional["NodeVBox"] = None
+parent_widget: Optional["NodeCarrier"] = None
 parent_dtypes: Optional[Dict[str, str]] = None
 key_by_id: Dict[int, Tuple[str, int]] = {}
-widget_by_id: Dict[int, "NodeVBox"] = {}
-widget_by_key: Dict[Tuple[str, int], "NodeVBox"] = {}
+widget_by_id: Dict[int, "NodeCarrier"] = {}
+widget_by_key: Dict[Tuple[str, int], "NodeCarrier"] = {}
 widget_numbers: Dict[str, int] = defaultdict(int)
 recording_state: bool = False
 
@@ -412,13 +453,16 @@ def create_loader_widget(
     ctx = dict(parent=obj, dtypes=dtypes, input_module=obj._output_module, dag=dag)
     from .csv_loader import CsvLoaderW
     from .parquet_loader import ParquetLoaderW
+    from .custom_loader import CustomLoaderW
 
-    loader: Union[CsvLoaderW, ParquetLoaderW]
+    loader: CsvLoaderW | ParquetLoaderW | CustomLoaderW
     if ftype == "csv":
         loader = CsvLoaderW()
-    else:
-        assert ftype == "parquet"
+    elif ftype == "parquet":
         loader = ParquetLoaderW()
+    else:
+        assert ftype == "custom"
+        loader = CustomLoaderW()
     if frozen is not None:
         loader.frozen_kw = frozen
     stage = NodeCarrier(ctx, loader)
@@ -435,11 +479,11 @@ def create_loader_widget(
     return stage
 
 
-def get_widget_by_id(key: int) -> "NodeVBox":
+def get_widget_by_id(key: int) -> "NodeCarrier":
     return widget_by_id[key]
 
 
-def get_widget_by_key(key: str, num: int) -> "NodeVBox":
+def get_widget_by_key(key: str, num: int) -> "NodeCarrier":
     return widget_by_key[(key, num)]
 
 
@@ -453,7 +497,7 @@ def set_recording_state(val: bool) -> None:
 
 
 def _make_btn_start_loader(
-        obj: "NodeVBox", ftype: str, alias: WidgetType, frozen: AnyType = None
+        obj: "NodeCarrier", ftype: str, alias: WidgetType, frozen: AnyType = None
 ) -> Callable[..., None]:
     def _cbk(btn: ipw.Button) -> None:
         global parent_widget
@@ -468,7 +512,7 @@ def _cbk(btn: ipw.Button) -> None:
 
 
 def replay_start_loader(
-    obj: "NodeVBox", ftype: str, alias: str, frozen: AnyType | None = None
+    obj: "NodeCarrier", ftype: str, alias: str, frozen: AnyType | None = None
 ) -> None:
     global parent_widget
     parent_widget = obj
@@ -477,7 +521,7 @@ def replay_start_loader(
 
 
 def replay_new_stage(
-    obj: "NodeVBox", title: str, frozen: AnyType | None = None
+    obj: "NodeCarrier", title: str, frozen: AnyType | None = None
 ) -> None:
     class _FakeSel:
         value: str
@@ -675,6 +719,58 @@ def get_previous(obj: "ChainingWidget") -> "ChainingWidget":
 
 new_stage_cell_0 = "Constructor.widget('{key}'){end}"
 new_stage_cell = "Constructor.widget('{key}', {num}){end}"
+new_stage_cell_code = ("proxy = Constructor.proxy('{key}', {num})\n"
+                       "# proxy object provides the following attributes:\n"
+                       "#  input_module: Module | TableFacade \n"
+                       "#  input_slot: str \n"
+                       "#  input_dtypes: dict[str, str] | None\n"
+                       "#  scheduler: Scheduler\n"
+                       "# Warning: keep the code above unchanged\n"
+                       "# Put your own code here\n"
+                       "...\n"
+                       "...\n"
+                       "...\n"
+                       "# fill in the following variables:\n"
+                       "output_module: 'Module | TableFacade' = ...\n"
+                       "output_slot: str = 'result'\n"
+                       "output_dtypes: dict[str, str] | None = None\n"
+                       "freeze: bool = False\n"
+                       "# Warning: keep the code below unchanged\n"
+                       "proxy.resume(output_module, output_slot, output_dtypes, freeze)"
+                       "{end}"
+                       )
+
+new_loader_cell_code = ("proxy = Constructor.proxy('{key}', {num})\n"
+                        "scheduler = proxy.scheduler\n"
+                        "# Warning: keep the code above unchanged\n"
+                        "# Put your own imports here\n"
+                        "... \n"
+                        "with scheduler:\n"
+                        "    # Put your own code here\n"
+                        "    ...\n"
+                        "    ...\n"
+                        "    # fill in the following variables:\n"
+                        "    output_module: 'Module | TableFacade' = ...\n"
+                        "    output_slot: str = 'result'\n"
+                        "    output_dtypes: dict[str, str] | None = None\n"
+                        "    freeze: bool = False\n"
+                        "    # Warning: keep the code below unchanged\n"
+                        "    display(proxy.resume(output_module, output_slot,"
+                        " output_dtypes, freeze))"
+                        "{end}"
+                        )
+
+
+def get_stage_cell(key: str, num: int, end: str) -> tuple[str, bool, bool]:
+    if key == "Python":
+        return new_stage_cell_code.format(key=key, num=num, end=end), True, False
+    return new_stage_cell.format(key=key, num=num, end=end), False, True
+
+
+def get_loader_cell(key: str, ftype: str, num: int, end: str) -> tuple[str, bool, bool]:
+    if ftype == "custom":
+        return new_loader_cell_code.format(key=key, num=num, end=end), True, False
+    return new_stage_cell.format(key=key, num=num, end=end), False, True
 
 
 def add_new_stage(parent: "ChainingWidget", title: str, frozen: AnyType = None) -> None:
@@ -686,8 +782,8 @@ def add_new_stage(parent: "ChainingWidget", title: str, frozen: AnyType = None)
     if frozen is not None:
         end = ".run()"
     md = "## " + title + (f"[{n}]" if n else "")
-    code = new_stage_cell.format(key=title, num=n, end=end)
-    labcommand("progressivis:create_stage_cells", tag=tag, md=md, code=code)
+    code, rw, run = get_stage_cell(key=title, num=n, end=end)
+    labcommand("progressivis:create_stage_cells", tag=tag, md=md, code=code, rw=rw, run=run)
     add_to_record(dict(title=title, parent=parent_key))
 
 
@@ -703,14 +799,10 @@ def add_new_loader(
         end = ".run();"
     if alias:
         md = f"## {alias}"
-        code = new_stage_cell_0.format(key=alias, end=end)
     else:
         md = "## " + title + (f"[{n}]" if n else "")
-        if n:
-            code = new_stage_cell.format(key=title, num=n, end=end)
-        else:
-            code = new_stage_cell_0.format(key=title, end=end)
-    labcommand("progressivis:create_stage_cells", tag=tag, md=md, code=code)
+    code, rw, run = get_loader_cell(key=alias or title, ftype=ftype, num=n, end=end)
+    labcommand("progressivis:create_stage_cells", tag=tag, md=md, code=code, rw=rw, run=run)
     add_to_record(dict(ftype=ftype, alias=alias))