Submit notebook context to TextCortex (#14)

* Moved exec(code) into run_dialog()

* wip, added back poetry

* introduced icortex history variable

* Introduced ICortexHistory, ServiceInteraction. Migrated setup to poetry

* wip

* Minor

* Bumped version

Author: osolmaz

.gitignore

@@ -135,8 +135,6 @@ cache
 *.zip
 cache.json
 Untitled*.ipynb
-pyproject.toml
-poetry.lock
 tmp/
 lab/
 *.csv

icortex/__init__.py

@@ -1,4 +1,7 @@
 from icortex.kernel import ICortexKernel, print_help, get_icortex_kernel
-from icortex.cli import set_icortex_service, eval_cli
-from icortex.exec import eval_prompt
+
+# from icortex.cli import set_icortex_service, eval_cli
 import icortex.services
+import importlib.metadata
+
+__version__ = importlib.metadata.version("icortex")

icortex/cli.py

@@ -2,7 +2,6 @@
 import shlex
 import sys
 import argparse
-from icortex.kernel import get_icortex_kernel
 from jupyter_console.app import ZMQTerminalIPythonApp
 from icortex.services import get_available_services
 from icortex.defaults import DEFAULT_ICORTEX_CONFIG_PATH
@@ -150,14 +149,14 @@ def get_parser(prog=None):
     return parser, parser_service
 
 
-def set_icortex_service(config_path=DEFAULT_ICORTEX_CONFIG_PATH):
-    kernel = get_icortex_kernel()
-    if kernel is not None:
-        return ICortexConfig(DEFAULT_ICORTEX_CONFIG_PATH).set_service()
-    return False
+# def set_icortex_service(kernel, config_path=DEFAULT_ICORTEX_CONFIG_PATH):
 
+#     if kernel is not None:
+#         return ICortexConfig(DEFAULT_ICORTEX_CONFIG_PATH).set_service()
+#     return False
 
-def main(argv=None, prog=None):
+
+def main(argv=None, prog=None, kernel=None):
     if argv is None:
         argv = sys.argv[1:]
 
@@ -195,6 +194,7 @@ def main(argv=None, prog=None):
     elif args.command == "help":
         parser.print_help()
     elif args.command == "shell" or args.command is None:
+        from icortex.kernel import get_icortex_kernel
         kernel = get_icortex_kernel()
         if kernel is None:
             ZMQTerminalICortexApp.launch_instance()

icortex/config.py

@@ -3,14 +3,17 @@
 
 from icortex.services import get_available_services, get_service
 from icortex.helper import yes_no_input, prompt_input
-from icortex.kernel import get_icortex_kernel
 from icortex.services.service_base import ServiceVariable
 
 
 class ICortexConfig:
     def __init__(self, path: str):
         self.path = path
         self.read_config()
+        self.kernel = None
+
+    def set_kernel(self, kernel):
+        self.kernel = kernel
 
     def get_service_name(self):
         if "service" in self.dict:
@@ -45,8 +48,8 @@ def set_service_var(self, var_name: str, var_value) -> bool:
         success = self.write_config()
         if success:
             print(f"Set variable {var_name} to {cast_value}.")
-            kernel = get_icortex_kernel()
-            if kernel is not None:
+            # kernel = get_icortex_kernel()
+            if self.kernel is not None:
                 self.set_service()
             return True
         else:
@@ -73,17 +76,17 @@ def set_service_config(self, service_name: str, hard_init=False) -> bool:
         success = self.write_config()
         if success:
             print(f"Set service to {service_name} successfully.")
-            kernel = get_icortex_kernel()
-            if kernel is not None:
+            # kernel = get_icortex_kernel()
+            if self.kernel is not None:
                 self.set_service()
             return True
         else:
             raise Exception("Could not write configuration file")
 
     def set_service(self):
         # TODO: pass the --config flag from icortex somehow
-        kernel = get_icortex_kernel()
-        if kernel is None:
+        # kernel = get_icortex_kernel()
+        if self.kernel is None:
             return False
 
         if not self.dict:
@@ -102,7 +105,7 @@ def set_service(self):
         service_config = self.dict[service_name]
         service_class = get_service(service_name)
 
-        kernel.set_service(service_class(service_config))
+        self.kernel.set_service(service_class(service_config))
         return True
 
     def ask_which_service(self) -> str:

icortex/context.py

@@ -0,0 +1,85 @@
+import typing as t
+import importlib.metadata
+from copy import deepcopy
+import platform
+from icortex.defaults import DEFAULT_HISTORY_VAR
+
+icortex_version = importlib.metadata.version("icortex")
+
+INITIAL_HISTORY_VAL = {
+    "metadata": {
+        "kernelspec": {
+            "display_name": "ICortex (Python 3)",
+            "language": "icortex",
+            "name": "icortex",
+        },
+        "language_info": {
+            "pygments_lexer": "icortex",
+            # "codemirror_mode": {"name": "ipython", "version": 3},
+            "file_extension": ".icx",
+            "mimetype": "text/x-icortex",
+            "name": "icortex",
+            # "nbconvert_exporter": "python",
+            "pygments_lexer": "icortex",
+            "version": icortex_version,
+            "python_version": platform.python_version(),
+        },
+    },
+    "nbformat": 4,
+    "nbformat_minor": 5,
+    "cells": [],
+}
+
+
+class ICortexHistory:
+    """Interface to construct a history variable in globals for storing
+    notebook context.
+    The constructed dict maps to JSON, and the schema is compatible
+    with the Jupyter notebook format:
+    https://nbformat.readthedocs.io/en/latest/format_description.html"""
+
+    def __init__(self, scope: t.Dict[str, t.Any]):
+        self.scope = scope
+        self._check_init()
+
+    def _check_init(self):
+        if DEFAULT_HISTORY_VAR not in self.scope:
+            self.scope[DEFAULT_HISTORY_VAR] = deepcopy(INITIAL_HISTORY_VAL)
+
+        self._dict = self.scope[DEFAULT_HISTORY_VAR]
+
+    def get_dict(self):
+        return deepcopy(self._dict)
+
+    def add_code(self, code: str, outputs: t.List[t.Any]):
+        self._check_init()
+
+        ret = {
+            "cell_type": "code",
+            "metadata": {},
+            "source": code,
+            "outputs": outputs,
+        }
+
+        self._dict["cells"].append(ret)
+        return ret
+
+    def add_prompt(
+        self,
+        prompt: str,
+        outputs: t.List[t.Any],
+        service_interaction: t.Dict[str, t.Any],
+    ):
+        self._check_init()
+
+        ret = {
+            "cell_type": "code",
+            "metadata": {
+                "service": service_interaction,
+            },
+            "source": prompt,
+            "outputs": outputs,
+        }
+
+        self._dict["cells"].append(ret)
+        return ret

icortex/defaults.py

@@ -1,8 +1,9 @@
-
 # Default parameters
 DEFAULT_ICORTEX_CONFIG_PATH = "icortex.toml"
 DEFAULT_CACHE_PATH = "cache.json"
 DEFAULT_REGENERATE = False
 DEFAULT_AUTO_INSTALL_PACKAGES = False
+DEFAULT_AUTO_EXECUTE = False
 DEFAULT_QUIET = False
 DEFAULT_SERVICE = "textcortex"
+DEFAULT_HISTORY_VAR = "_icortex_history"

icortex/exec.py

@@ -1,28 +1,33 @@
-def unescape(s):
+from pygments import highlight
+from pygments.formatters import Terminal256Formatter
+from pygments.lexers import PythonLexer
+
+
+def unescape(s) -> str:
     return s.encode("utf-8").decode("unicode_escape")
 
 
-def is_prompt(input: str):
+def is_prompt(input: str) -> bool:
     return input.strip()[0] == "/"
 
 
-def is_cli(input: str):
+def is_cli(input: str) -> bool:
     return input.strip()[:2] == "//"
 
 
-def escape_quotes(s: str):
+def escape_quotes(s: str) -> str:
     return s.replace('"', r"\"").replace("'", r"\'")
 
 
-def extract_prompt(input: str):
+def extract_prompt(input: str) -> str:
     return input.strip()[1:].strip()
 
 
-def extract_cli(input: str):
+def extract_cli(input: str) -> str:
     return input.strip()[2:].strip()
 
 
-def yes_no_input(message: str, default=True):
+def yes_no_input(message: str, default=True) -> bool:
     if default:
         message += " [Y/n]"
     else:
@@ -45,3 +50,7 @@ def prompt_input(message: str, type=str, default=None):
         return default
     else:
         return type(user_input)
+
+
+def highlight_python(code: str):
+    return highlight(code, PythonLexer(), Terminal256Formatter())