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 |
'' |
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)