Skip to content

helpers

delete_expressions_from_notebook(expressions, file_path)

Delete code lines from the notebook that match the regular expressions

Parameters:

Name Type Description Default
expressions List[str]

List of regular expressions to match

required
file_path str

Path to the notebook

required
Source code in aicrowd/notebook/helpers.py
def delete_expressions_from_notebook(
    expressions: List[str], file_path: str
) -> Dict[str, Any]:
    """
    Delete code lines from the notebook that match the regular expressions

    Args:
        expressions: List of regular expressions to match
        file_path: Path to the notebook
    """
    nb = read_notebook(file_path)

    # TODO: Duplicated code, refactor
    for _cell in nb["cells"]:
        # Match the lines only in code blocks
        if _cell["cell_type"] != "code":
            continue
        source_code = []
        for _code_line in _cell["source"]:
            matched = False
            for _expr in expressions:
                if re.search(_expr, _code_line):
                    log.debug("Matched line to remove: %s", _code_line)
                    matched = True
                    break
            if not matched:
                source_code.append(_code_line)
        _cell["source"] = source_code
    return nb

get_colab_notebook_via_kernel(timeout_sec=180)

Get colab notebook from colab frontend via messaging

Source code in aicrowd/notebook/helpers.py
def get_colab_notebook_via_kernel(
    timeout_sec: int = 180,
) -> Union[Tuple[Dict[str, Any], str], Tuple[None, None]]:
    """
    Get colab notebook from colab frontend via messaging
    """
    _message = import_module("google.colab._message")
    notebook = _message.blocking_request("get_ipynb", timeout_sec=timeout_sec)
    if notebook is None:
        return None, None
    return notebook["ipynb"], notebook["ipynb"]["metadata"]["colab"]["name"]

get_jupyter_server_endpoint()

Returns the jupyter server endpoint along with the token

Returns:

Type Description
Tuple[str, str]

Jupyter server endpoint and token

Source code in aicrowd/notebook/helpers.py
def get_jupyter_server_endpoint() -> Tuple[str, str]:
    """
    Returns the jupyter server endpoint along with the token

    Returns:
        Jupyter server endpoint and token
    """
    server_info = get_jupyter_server_info()
    return (
        server_info.get("url", get_default_jupyter_api_session_host()),
        server_info.get("token", ""),
    )

get_jupyter_server_info()

Returns the latest jupyter server info

Source code in aicrowd/notebook/helpers.py
def get_jupyter_server_info() -> Dict[str, str]:
    """
    Returns the latest jupyter server info
    """
    try:
        return get_jupyter_lab_server_info()
    except exceptions.NotebookAppImportException:
        log.debug("Found no active kernels from jupyter lab")
        return get_jupyter_notebook_server_info()

get_notebook(notebook_path=None)

Get notebook JSON or notebook file path and, notebook name.

If notebook path is provided then returns the notebook path along with the notebook name else based on the kernel environment, returns notebook and notebook name. If the environment is Google colaboratory and if the code is running inside kernel, fetch notebook from frontend else mount google drive and get notebook path. If the environment is Jupyter notebook or Jupyterlab, get notebook path from disk. If the code is running inside kernel, save the notebook else show warning message.

Parameters:

Name Type Description Default
notebook_path str

Path of the notebook

None

Returns:

Type Description
Tuple[Union[Dict[str, Any], str], str]

notebook: A JSON dictionary containing notebook contents or Path to the notebook file notebook_name: Name of the notebook

Source code in aicrowd/notebook/helpers.py
def get_notebook(notebook_path: str = None) -> Tuple[Union[Dict[str, Any], str], str]:
    """Get notebook JSON or notebook file path and, notebook name.

    If notebook path is provided then returns the notebook path along with the notebook name else based on the kernel
    environment, returns notebook and notebook name. If the environment is Google colaboratory and if the code is
    running inside kernel, fetch notebook from frontend else mount google drive and get notebook path. If the
    environment is Jupyter notebook or Jupyterlab, get notebook path from disk. If the code is running inside kernel,
    save the notebook else show warning message.

    Args:
        notebook_path: Path of the notebook

    Returns:
        notebook: A JSON dictionary containing notebook contents or Path to the notebook file
        notebook_name: Name of the notebook
    """
    if notebook_path:
        return notebook_path, os.path.basename(notebook_path)
    if is_google_colab_env():
        log.debug("Detected google colab environment")
        return get_colab_notebook()
    return get_saved_jupyter_notebook()

save_jupyter_notebook(file_path, timeout_sec=5)

Save jupyter notebook

Use kernel comms messaging to save the notebook and wait until the notebook is saved or timeout occurs.

Parameters:

Name Type Description Default
file_path str

Notebook file path to be saved

required
timeout_sec int

Time in seconds to wait until file is getting saved

5
Source code in aicrowd/notebook/helpers.py
def save_jupyter_notebook(file_path: str, timeout_sec: int = 5):
    """Save jupyter notebook

    Use kernel comms messaging to save the notebook and wait until the notebook is saved or timeout occurs.

    Args:
        file_path: Notebook file path to be saved
        timeout_sec: Time in seconds to wait until file is getting saved
    """
    ipython_display = import_module("IPython.display")
    log.debug("Saving notebook: %s", file_path)
    deadline = time.time() + timeout_sec
    start_md5 = hashlib.md5(open(file_path, "rb").read()).hexdigest()
    ipython_display.display(
        ipython_display.Javascript("IPython.notebook.save_checkpoint();")
    )
    current_md5 = start_md5
    while (start_md5 == current_md5) and (time.time() < deadline):
        log.debug("Waiting for the notebook to finish saving")
        time.sleep(1)
        current_md5 = hashlib.md5(open(file_path, "rb").read()).hexdigest()

show_warning_to_use_magic_command(reason='')

Show warning message to use %aicrowd magic command with complete code

Some features won't work with !aicrowd. %aicrowd magic command can be used to save notebook inside jupyter notebook/jupyterLab environment and also to get notebook directly from frontend without mounting the drive in colab environment.

Parameters:

Name Type Description Default
reason str

A string which describes the reason to use %aicrowd command over !aicrowd

''
Source code in aicrowd/notebook/helpers.py
def show_warning_to_use_magic_command(reason: str = "") -> None:
    """Show warning message to use `%aicrowd` magic command with complete code

    Some features won't work with !aicrowd. `%aicrowd` magic command can be used to save
    notebook inside jupyter notebook/jupyterLab environment and also to get notebook directly from frontend without
    mounting the drive in colab environment.

    Args:
        reason: A string which describes the reason to use `%aicrowd` command over `!aicrowd`
    """
    description = (
        f"`%aicrowd` magic command can be used to save the notebook inside jupyter notebook/jupyterLab environment "
        f"and also to get the notebook directly from the frontend without mounting the drive in colab environment. "
        f"You can use magic command {reason} and submit using the code below:\n "
    )
    code = "\n".join(["%load_ext aicrowd.magic", "%aicrowd " + " ".join(sys.argv[1:])])
    warnings.warn(description + code)