<filename>jupyterlab_extension/services.py
# -*- coding: utf-8 -*-
import json
import os
import re

import requests
from unicodedata import normalize

DATASETS_ENDPOINT = os.getenv("DATASETS_ENDPOINT", "http://datasets.platiagro:8080")
PROJECTS_ENDPOINT = os.getenv("PROJECTS_ENDPOINT", "http://projects.platiagro:8080")


def find_task_by_name(os_path):
    """
    Get the task name from the notebook using the PlatIAgro Projects API.

    Parameters
    ----------
    os_path : str

    Returns
    -------
    dict
        The task_id.

    Raises
    ------
    ConnectionError
        When the request did not succeed.
    HTTPError
        When the request did not succeed.
    """
    # only do this for Experiment.ipynb
    match = re.search(r"/tasks/(.*?)/Experiment.ipynb", os_path)

    if match:
        task_name = match.group(1)

        params = {'name': task_name}
        r = requests.get(f"{PROJECTS_ENDPOINT}/tasks", params=params)
        r.raise_for_status()
        tasks = r.json()['tasks']

        if len(tasks) > 0:
            for task in tasks:
                if task['name'] == task_name:

                    return task["uuid"]


def update_task(task_id, parameters=None) -> dict:
    """
    Updates a task from notebook using PlatIAgro Projects API.

    Parameters
    ----------
    task_id : str
    parameters : dict

    Returns
    -------
    dict
        The task details.

    Raises
    ------
    ConnectionError
        When the request did not succeed.
    HTTPError
        When the request did not succeed.
    """
    json = {}

    if parameters:
        json["parameters"] = parameters

    r = requests.patch(f"{PROJECTS_ENDPOINT}/tasks/{task_id}", json=json)
    r.raise_for_status()
    return r.json()


def create_dataset(file: bytes, filename: str = "file") -> dict:
    """Creates a dataset from a CSV file using PlatIAgro Datasets API.

    Args:
        file (bytes): file object.
        filename (str, optional): filename. Defaults to "file".

    Returns:
        dict: The dataset details: name, columns, and filename.

    Raises:
        ConnectionError: When the request did not succeed.
        HTTPError: When the request did not succeed.
    """
    files = {"file": (filename, file)}
    r = requests.post(f"{DATASETS_ENDPOINT}/datasets", files=files)
    r.raise_for_status()
    return r.json()


def generate_name(filename: str, attempt: int = 1, path: str = "/tmp/data") -> str:
    """Generates a dataset name from a given filename.

    Args:
        filename (str): source filename.
        path (str): path to check if name exist.
        attempt (int, optional): the current attempt of generating a new name.

    Returns:
        str: new generated dataset name.
    """
    name = normalize('NFKD', filename) \
        .encode('ASCII', 'ignore') \
        .replace(b' ', b'-') \
        .decode()

    if attempt > 1:
        name, extension = os.path.splitext(name)
        name = f"{name}-{attempt}{extension}"

    try:
        os.makedirs("/tmp/data", exist_ok=True)
        open(f"{path}/{name}")
    except FileNotFoundError:
        return name

    return generate_name(filename, attempt + 1)


def create_dataset_locally(file: bytes, filename: str = "file", name: str = "file", path: str = "/tmp/data"):
    """Creates a dataset from a CSV and writes locally.

    Args:
        file (bytes): file object content.
        filename (str, optional): original file name. Defaults to "file".
        name (str, optional): new file name generated by datasets API. Defaults to "file".
        path (str, optional): path to be writen. Defaults to "/tmp/data".

    Raises:
        OSError: When the writing did not succeed.
    """

    try:
        os.makedirs("/tmp/data", exist_ok=True)
        with open(f"{path}/{name}", 'wb') as csv_file:
            csv_file.write(file)

        return {"filename": filename, "name": name}
    except OSError as e:
        print(e)


def parse_parameters(experiment_notebook):
    """
    Parses and returns the parameters declared in a notebook.

    Parameters
    ----------
    experiment_notebook : dict

    Returns
    -------
    list:
        A list of parameters (name, default, type, label, description).
    """
    parameters = []
    cells = experiment_notebook.get("cells", [])
    for cell in cells:
        cell_type = cell["cell_type"]
        tags = cell["metadata"].get("tags", [])
        if cell_type == "code" and "parameters" in tags:
            source = cell["source"]

            parameters.extend(read_parameters_from_source(source))

    return parameters


def read_parameters_from_source(source):
    """
    Lists the parameters declared in source code.

    Parameters
    ----------
    source : list
        Source code lines.

    Returns
    -------
    list:
        A list of parameters (name, default, type, label, description).
    """
    parameters = []
    # Regex to capture a parameter declaration
    # Inspired by Google Colaboratory Forms
    # Example of a parameter declaration:
    # name = "value" #@param ["1st option", "2nd option"] {type:"string", label:"Foo Bar", description:"Foo Bar"}
    pattern = re.compile(r"^(\w+)\s*=\s*(.+)\s+#@param(?:(\s+\[.*\]))?(\s+\{.*\})")

    for line in source:
        match = pattern.search(line)
        if match:
            try:
                name = match.group(1)
                default = match.group(2)
                options = match.group(3)
                metadata = match.group(4)

                parameter = {"name": name}

                if default and default != "None":
                    if default in ["True", "False"]:
                        default = default.lower()
                    parameter["default"] = json.loads(default)

                if options:
                    parameter["options"] = json.loads(options)

                # adds quotes to metadata keys
                metadata = re.sub(r"(\w+):", r'"\1":', metadata)
                parameter.update(json.loads(metadata))

                parameters.append(parameter)
            except json.JSONDecodeError:
                pass

    return parameters

<filename>jupyterlab_extension/services.py
# -*- coding: utf-8 -*-
import json
import os
import re

import requests
from unicodedata import normalize

DATASETS_ENDPOINT = os.getenv("DATASETS_ENDPOINT", "http://datasets.platiagro:8080")
PROJECTS_ENDPOINT = os.getenv("PROJECTS_ENDPOINT", "http://projects.platiagro:8080")


def find_task_by_name(os_path):
    """
    Get the task name from the notebook using the PlatIAgro Projects API.

    Parameters
    ----------
    os_path : str

    Returns
    -------
    dict
        The task_id.

    Raises
    ------
    ConnectionError
        When the request did not succeed.
    HTTPError
        When the request did not succeed.
    """
    # only do this for Experiment.ipynb
    match = re.search(r"/tasks/(.*?)/Experiment.ipynb", os_path)

    if match:
        task_name = match.group(1)

        params = {'name': task_name}
        r = requests.get(f"{PROJECTS_ENDPOINT}/tasks", params=params)
        r.raise_for_status()
        tasks = r.json()['tasks']

        if len(tasks) > 0:
            for task in tasks:
                if task['name'] == task_name:

                    return task["uuid"]


def update_task(task_id, parameters=None) -> dict:
    """
    Updates a task from notebook using PlatIAgro Projects API.

    Parameters
    ----------
    task_id : str
    parameters : dict

    Returns
    -------
    dict
        The task details.

    Raises
    ------
    ConnectionError
        When the request did not succeed.
    HTTPError
        When the request did not succeed.
    """
    json = {}

    if parameters:
        json["parameters"] = parameters

    r = requests.patch(f"{PROJECTS_ENDPOINT}/tasks/{task_id}", json=json)
    r.raise_for_status()
    return r.json()


def create_dataset(file: bytes, filename: str = "file") -> dict:
    """Creates a dataset from a CSV file using PlatIAgro Datasets API.

    Args:
        file (bytes): file object.
        filename (str, optional): filename. Defaults to "file".

    Returns:
        dict: The dataset details: name, columns, and filename.

    Raises:
        ConnectionError: When the request did not succeed.
        HTTPError: When the request did not succeed.
    """
    files = {"file": (filename, file)}
    r = requests.post(f"{DATASETS_ENDPOINT}/datasets", files=files)
    r.raise_for_status()
    return r.json()


def generate_name(filename: str, attempt: int = 1, path: str = "/tmp/data") -> str:
    """Generates a dataset name from a given filename.

    Args:
        filename (str): source filename.
        path (str): path to check if name exist.
        attempt (int, optional): the current attempt of generating a new name.

    Returns:
        str: new generated dataset name.
    """
    name = normalize('NFKD', filename) \
        .encode('ASCII', 'ignore') \
        .replace(b' ', b'-') \
        .decode()

    if attempt > 1:
        name, extension = os.path.splitext(name)
        name = f"{name}-{attempt}{extension}"

    try:
        os.makedirs("/tmp/data", exist_ok=True)
        open(f"{path}/{name}")
    except FileNotFoundError:
        return name

    return generate_name(filename, attempt + 1)


def create_dataset_locally(file: bytes, filename: str = "file", name: str = "file", path: str = "/tmp/data"):
    """Creates a dataset from a CSV and writes locally.

    Args:
        file (bytes): file object content.
        filename (str, optional): original file name. Defaults to "file".
        name (str, optional): new file name generated by datasets API. Defaults to "file".
        path (str, optional): path to be writen. Defaults to "/tmp/data".

    Raises:
        OSError: When the writing did not succeed.
    """

    try:
        os.makedirs("/tmp/data", exist_ok=True)
        with open(f"{path}/{name}", 'wb') as csv_file:
            csv_file.write(file)

        return {"filename": filename, "name": name}
    except OSError as e:
        print(e)


def parse_parameters(experiment_notebook):
    """
    Parses and returns the parameters declared in a notebook.

    Parameters
    ----------
    experiment_notebook : dict

    Returns
    -------
    list:
        A list of parameters (name, default, type, label, description).
    """
    parameters = []
    cells = experiment_notebook.get("cells", [])
    for cell in cells:
        cell_type = cell["cell_type"]
        tags = cell["metadata"].get("tags", [])
        if cell_type == "code" and "parameters" in tags:
            source = cell["source"]

            parameters.extend(read_parameters_from_source(source))

    return parameters


def read_parameters_from_source(source):
    """
    Lists the parameters declared in source code.

    Parameters
    ----------
    source : list
        Source code lines.

    Returns
    -------
    list:
        A list of parameters (name, default, type, label, description).
    """
    parameters = []
    # Regex to capture a parameter declaration
    # Inspired by Google Colaboratory Forms
    # Example of a parameter declaration:
    # name = "value" #@param ["1st option", "2nd option"] {type:"string", label:"Foo Bar", description:"Foo Bar"}
    pattern = re.compile(r"^(\w+)\s*=\s*(.+)\s+#@param(?:(\s+\[.*\]))?(\s+\{.*\})")

    for line in source:
        match = pattern.search(line)
        if match:
            try:
                name = match.group(1)
                default = match.group(2)
                options = match.group(3)
                metadata = match.group(4)

                parameter = {"name": name}

                if default and default != "None":
                    if default in ["True", "False"]:
                        default = default.lower()
                    parameter["default"] = json.loads(default)

                if options:
                    parameter["options"] = json.loads(options)

                # adds quotes to metadata keys
                metadata = re.sub(r"(\w+):", r'"\1":', metadata)
                parameter.update(json.loads(metadata))

                parameters.append(parameter)
            except json.JSONDecodeError:
                pass

    return parameters

# -*- coding: utf-8 -*- Get the task name from the notebook using the PlatIAgro Projects API.      Parameters     ----------     os_path : str      Returns     -------     dict         The task_id.      Raises     ------     ConnectionError         When the request did not succeed.     HTTPError         When the request did not succeed. # only do this for Experiment.ipynb Updates a task from notebook using PlatIAgro Projects API.      Parameters     ----------     task_id : str     parameters : dict      Returns     -------     dict         The task details.      Raises     ------     ConnectionError         When the request did not succeed.     HTTPError         When the request did not succeed. Creates a dataset from a CSV file using PlatIAgro Datasets API.      Args:         file (bytes): file object.         filename (str, optional): filename. Defaults to "file".      Returns:         dict: The dataset details: name, columns, and filename.      Raises:         ConnectionError: When the request did not succeed.         HTTPError: When the request did not succeed. Generates a dataset name from a given filename.      Args:         filename (str): source filename.         path (str): path to check if name exist.         attempt (int, optional): the current attempt of generating a new name.      Returns:         str: new generated dataset name. Creates a dataset from a CSV and writes locally.      Args:         file (bytes): file object content.         filename (str, optional): original file name. Defaults to "file".         name (str, optional): new file name generated by datasets API. Defaults to "file".         path (str, optional): path to be writen. Defaults to "/tmp/data".      Raises:         OSError: When the writing did not succeed. Parses and returns the parameters declared in a notebook.      Parameters     ----------     experiment_notebook : dict      Returns     -------     list:         A list of parameters (name, default, type, label, description). Lists the parameters declared in source code.      Parameters     ----------     source : list         Source code lines.      Returns     -------     list:         A list of parameters (name, default, type, label, description). # Regex to capture a parameter declaration # Inspired by Google Colaboratory Forms # Example of a parameter declaration: # name = "value" #@param ["1st option", "2nd option"] {type:"string", label:"Foo Bar", description:"Foo Bar"} #@param(?:(\s+\[.*\]))?(\s+\{.*\})") # adds quotes to metadata keys