Module pipelines.rj_escritorio.dummy_predict.tasks

Tasks for the predict flow example usage.

Functions

def get_current_timestamp() ‑> str

Expand source code

@task(checkpoint=False)
def get_current_timestamp() -> str:
    """
    Get current timestamp.
    """
    return pendulum.now(tz=constants.DEFAULT_TIMEZONE.value).to_datetime_string()

Get current timestamp.

def get_dummy_input_data() ‑> pandas.core.frame.DataFrame

Expand source code

@task(checkpoint=False)
def get_dummy_input_data() -> pd.DataFrame:
    """
    Return some dummy data
    """
    columns = [
        "alcohol",
        "chlorides",
        "citric acid",
        "density",
        "fixed acidity",
        "free sulfur dioxide",
        "pH",
        "residual sugar",
        "sulphates",
        "total sulfur dioxide",
        "volatile acidity",
    ]
    data = [
        [
            random() * 15,
            random(),
            random(),
            random(),
            random() * 7,
            random() * 35,
            random() * 5,
            random() * 2,
            random(),
            random() * 100,
            random(),
        ]
    ]
    return pd.DataFrame(data=data, columns=columns)

Return some dummy data

def random()

random() -> x in the interval [0, 1).