getml.pipeline.Pipeline

def check(
    self,
    population_table: Union[DataFrame, View, data.Subset],
    peripheral_tables: Optional[
        Union[
            Dict[str, Union[DataFrame, View]],
            Sequence[Union[DataFrame, View]],
        ]
    ] = None,
) -> Optional[Issues]:
    """
    Checks the validity of the data model.

    Args:
        population_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable.

        peripheral_tables:
            Additional tables corresponding to the ``peripheral``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If passed as a list, the order needs to
            match the order of the corresponding placeholders passed
            to ``peripheral``.

            If you pass a [`Subset`][getml.data.Subset] to `population_table`,
            the peripheral tables from that subset will be used. If you use
            a [`Container`][getml.data.Container], [`StarSchema`][getml.data.StarSchema]
            or [`TimeSeries`][getml.data.TimeSeries], that means you are passing
            a [`Subset`][getml.data.Subset].

    """

    if isinstance(population_table, data.Subset):
        peripheral_tables = population_table.peripheral
        population_table = population_table.population

    peripheral_tables = _transform_peripheral(peripheral_tables, self.peripheral)

    _check_df_types(population_table, peripheral_tables)

    temp = copy.deepcopy(self)

    temp._send()

    cmd: Dict[str, Any] = {}

    cmd["type_"] = temp.type + ".check"
    cmd["name_"] = temp.id

    cmd["peripheral_dfs_"] = [df._getml_deserialize() for df in peripheral_tables]
    cmd["population_df_"] = population_table._getml_deserialize()

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Found!":
            comm.handle_engine_exception(msg)
        print("Checking data model...")
        msg = comm.log(sock)
        if msg != "Success!":
            comm.handle_engine_exception(msg)
        issues = Issues(comm.recv_issues(sock))
        if len(issues) == 0:
            print("OK.")
        else:
            print(
                f"The pipeline check generated {len(issues.info)} "
                + f"issues labeled INFO and {len(issues.warnings)} "
                + "issues labeled WARNING."
            )

    temp.delete()

    return None if len(issues) == 0 else issues

def delete(self) -> None:
    """
    Deletes the pipeline from the Engine.

    Warning:
        You can not undo this action!
    """
    self._check_whether_fitted()

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".delete"
    cmd["name_"] = self.id
    cmd["mem_only_"] = False

    comm.send(cmd)

    self._id = NOT_FITTED

def deploy(self, deploy: bool) -> None:
    """Allows a fitted pipeline to be addressable via an HTTP request.
    See [deployment][deployment] for details.

    Args:
        deploy: If `True`, the deployment of the pipeline
            will be triggered.
    """
    self._check_whether_fitted()

    if not isinstance(deploy, bool):
        raise TypeError("'deploy' must be of type bool")

    self._validate()

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".deploy"
    cmd["name_"] = self.id
    cmd["deploy_"] = deploy

    comm.send(cmd)

    self._save()

def fit(
    self,
    population_table: Union[DataFrame, View, data.Subset],
    peripheral_tables: Optional[
        Union[
            Sequence[Union[DataFrame, View]],
            Dict[str, Union[DataFrame, View]],
        ]
    ] = None,
    validation_table: Optional[Union[DataFrame, View, data.Subset]] = None,
    check: bool = True,
) -> Pipeline:
    """Trains the feature learning algorithms, feature selectors
    and predictors.

    Args:
        population_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable.

        peripheral_tables:
            Additional tables corresponding to the ``peripheral``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If passed as a list, the order needs to
            match the order of the corresponding placeholders passed
            to ``peripheral``.

            If you pass a [`Subset`][getml.data.Subset] to `population_table`,
            the peripheral tables from that subset will be used. If you use
            a [`Container`][getml.data.Container], [`StarSchema`][getml.data.StarSchema]
            or [`TimeSeries`][getml.data.TimeSeries], that means you are passing
            a [`Subset`][getml.data.Subset].

        validation_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If you are passing a subset, that subset
            must be derived from the same container as *population_table*.

            Only used for early stopping in [`XGBoostClassifier`][getml.predictors.XGBoostClassifier]
            and [`XGBoostRegressor`][getml.predictors.XGBoostRegressor].

        check:
            Whether you want to check the data model before fitting. The checks are
            equivalent to the checks run by [`check`][getml.Pipeline.check].

    Returns:
        The fitted pipeline.
    """

    additional_tags = (
        ["container-" + population_table.container_id]
        if isinstance(population_table, data.Subset)
        else []
    )

    if (
        isinstance(population_table, data.Subset)
        and isinstance(validation_table, data.Subset)
        and validation_table.container_id != population_table.container_id
    ):
        raise ValueError(
            "The subset used for validation must be from the same container "
            + "as the subset used for training."
        )

    if isinstance(population_table, data.Subset):
        peripheral_tables = population_table.peripheral
        population_table = population_table.population

    if isinstance(validation_table, data.Subset):
        validation_table = validation_table.population

    peripheral_tables = _transform_peripheral(peripheral_tables, self.peripheral)

    _check_df_types(population_table, peripheral_tables)

    if check:
        warnings = self.check(population_table, peripheral_tables)
        if warnings:
            print("To see the issues in full, run .check() on the pipeline.")

    self._send(additional_tags)

    cmd: Dict[str, Any] = {}

    cmd["type_"] = self.type + ".fit"
    cmd["name_"] = self.id

    cmd["peripheral_dfs_"] = [df._getml_deserialize() for df in peripheral_tables]
    cmd["population_df_"] = population_table._getml_deserialize()

    if validation_table is not None:
        cmd["validation_df_"] = validation_table._getml_deserialize()

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)

        if msg != "Found!":
            comm.handle_engine_exception(msg)

        begin = time.monotonic()

        msg = comm.log(sock)

        end = time.monotonic()

        if "Trained" in msg:
            print(msg)
            _print_time_taken(begin, end, "Time taken: ")
        else:
            comm.handle_engine_exception(msg)

    self._save()

    return self.refresh()

def predict(
    self,
    population_table: Union[DataFrame, View, data.Subset],
    peripheral_tables: Optional[
        Union[
            Sequence[Union[DataFrame, View]],
            Dict[str, Union[DataFrame, View]],
        ]
    ] = None,
    table_name: str = "",
) -> Union[NDArray[np.float_], None]:
    """Forecasts on new, unseen data using the trained ``predictor``.

    Returns the predictions generated by the pipeline based on
    `population_table` and `peripheral_tables` or writes them into
    a database named `table_name`.

    Args:
        population_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable.

        peripheral_tables:
            Additional tables corresponding to the ``peripheral``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If passed as a list, the order needs to
            match the order of the corresponding placeholders passed
            to ``peripheral``.

            If you pass a [`Subset`][getml.data.Subset] to `population_table`,
            the peripheral tables from that subset will be used. If you use
            a [`Container`][getml.data.Container], [`StarSchema`][getml.data.StarSchema]
            or [`TimeSeries`][getml.data.TimeSeries], that means you are passing
            a [`Subset`][getml.data.Subset].

        table_name:
            If not an empty string, the resulting predictions will
            be written into a table in a [`database`][getml.database].
            Refer to [Unified import interface][importing-data-unified-interface] for further information.

    Returns:
        Resulting predictions provided in an array of the (number of rows in `population_table`, number of targets in `population_table`).

    Note:
        Only fitted pipelines
        ([`fit`][getml.Pipeline.fit]) can be used for
        prediction.


    """

    self._check_whether_fitted()

    if isinstance(population_table, data.Subset):
        peripheral_tables = population_table.peripheral
        population_table = population_table.population

    peripheral_tables = _transform_peripheral(peripheral_tables, self.peripheral)

    _check_df_types(population_table, peripheral_tables)

    if not isinstance(table_name, str):
        raise TypeError("'table_name' must be of type str")

    self._validate()

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".transform"
    cmd["name_"] = self.id
    cmd["http_request_"] = False

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Found!":
            comm.handle_engine_exception(msg)
        y_hat = self._transform(
            peripheral_tables,
            population_table,
            sock,
            predict=True,
            table_name=table_name,
        )

    return y_hat

def refresh(self) -> Pipeline:
    """Reloads the pipeline from the Engine.

    This discards all local changes you have made since the
    last time you called [`fit`][getml.Pipeline.fit].

    Returns:
            Current instance
    """

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".refresh"
    cmd["name_"] = self.id

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)

    if msg[0] != "{":
        comm.handle_engine_exception(msg)

    json_obj = json.loads(msg)

    self._parse_json_obj(json_obj)

    return self

def score(
    self,
    population_table: Union[DataFrame, View, data.Subset],
    peripheral_tables: Optional[
        Union[
            Sequence[Union[DataFrame, View]],
            Dict[str, Union[DataFrame, View]],
        ]
    ] = None,
) -> Scores:
    """Calculates the performance of the ``predictor``.

    Returns different scores calculated on `population_table` and
    `peripheral_tables`.

    Args:
        population_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable.

        peripheral_tables:
            Additional tables corresponding to the ``peripheral``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If passed as a list, the order needs to
            match the order of the corresponding placeholders passed
            to ``peripheral``.

            If you pass a [`Subset`][getml.data.Subset] to `population_table`,
            the peripheral tables from that subset will be used. If you use
            a [`Container`][getml.data.Container], [`StarSchema`][getml.data.StarSchema]
            or [`TimeSeries`][getml.data.TimeSeries], that means you are passing
            a [`Subset`][getml.data.Subset].

    Returns:
        The scores of the pipeline.

    Note:
        Only fitted pipelines
        ([`fit`][getml.Pipeline.fit]) can be
        scored.

    """

    self._check_whether_fitted()

    if isinstance(population_table, data.Subset):
        peripheral_tables = population_table.peripheral
        population_table = population_table.population

    peripheral_tables = _transform_peripheral(peripheral_tables, self.peripheral)

    _check_df_types(population_table, peripheral_tables)

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".transform"
    cmd["name_"] = self.id
    cmd["http_request_"] = False

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)

        if msg != "Found!":
            comm.handle_engine_exception(msg)

        self._transform(
            peripheral_tables, population_table, sock, predict=True, score=True
        )

        msg = comm.recv_string(sock)

        if msg != "Success!":
            comm.handle_engine_exception(msg)

        scores = comm.recv_string(sock)

        scores = json.loads(scores)

    self.refresh()

    self._save()

    return self.scores

def transform(
    self,
    population_table: Union[DataFrame, View, data.Subset],
    peripheral_tables: Optional[
        Union[
            Sequence[Union[DataFrame, View]],
            Dict[str, Union[DataFrame, View]],
        ]
    ] = None,
    df_name: str = "",
    table_name: str = "",
) -> Union[DataFrame, NDArray[np.float_], None]:
    """Translates new data into the trained features.

    Transforms the data passed in `population_table` and
    `peripheral_tables` into features, which can be inserted into
    machine learning models.


    Args:
        population_table:
            Main table containing the target variable(s) and
            corresponding to the ``population``
            [`Placeholder`][getml.data.Placeholder] instance
            variable.

        peripheral_tables:
            Additional tables corresponding to the ``peripheral``
            [`Placeholder`][getml.data.Placeholder] instance
            variable. If passed as a list, the order needs to
            match the order of the corresponding placeholders passed
            to ``peripheral``.

            If you pass a [`Subset`][getml.data.Subset] to `population_table`,
            the peripheral tables from that subset will be used. If you use
            a [`Container`][getml.data.Container], [`StarSchema`][getml.data.StarSchema]
            or [`TimeSeries`][getml.data.TimeSeries], that means you are passing
            a [`Subset`][getml.data.Subset].

        df_name:
            If not an empty string, the resulting features will be
            written into a newly created DataFrame.

        table_name:
            If not an empty string, the resulting features will
            be written into a table in a [`database`][getml.database].
            Refer to [Unified import interface][importing-data-unified-interface] for further information.

    Returns:
        The features generated by the pipeline.

    ??? example
        By default, `transform` returns a [`ndarray`](https://numpy.org/doc/stable/reference/generated/numpy.ndarray.html):
        ```python
        my_features_array = pipe.transform()
        ```
        You can also export your features as a [`DataFrame`][getml.DataFrame]
        by providing the `df_name` argument:
        ```python
        my_features_df = pipe.transform(df_name="my_features")
        ```
        Or you can write the results directly into a database:
        ```python
        getml.database.connect_odbc(...)
        pipe.transform(table_name="MY_FEATURES")
        ```

    Note:
        Only fitted pipelines
        ([`fit`][getml.Pipeline.fit]) can transform
        data into features.

    """

    self._check_whether_fitted()

    if isinstance(population_table, data.Subset):
        peripheral_tables = population_table.peripheral
        population_table = population_table.population

    peripheral_tables = _transform_peripheral(peripheral_tables, self.peripheral)

    _check_df_types(population_table, peripheral_tables)

    self._validate()

    cmd: Dict[str, Any] = {}
    cmd["type_"] = self.type + ".transform"
    cmd["name_"] = self.id
    cmd["http_request_"] = False

    with comm.send_and_get_socket(cmd) as sock:
        msg = comm.recv_string(sock)
        if msg != "Found!":
            comm.handle_engine_exception(msg)
        y_hat = self._transform(
            peripheral_tables,
            population_table,
            sock,
            df_name=df_name,
            table_name=table_name,
        )

    if df_name != "":
        return data.DataFrame(name=df_name).refresh()

    return y_hat