# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/distributed.forecast.ipynb (unless otherwise specified).

__all__ = ['DistributedForecast']

# Cell
from typing import Callable, Dict, Optional

import dask.dataframe as dd
from dask.distributed import Client, default_client

from ..core import predictions_flow, preprocessing_flow
from ..forecast import Forecast
from .core import distributed_preprocess


# Cell
class DistributedForecast(Forecast):
    """Full pipeline encapsulation.

    Takes a model (`LGBMForecast` or `XGBForecast`), a flow configuration and a client."""

    def __init__(self, model, flow_config: Dict, client: Optional[Client] = None):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) -> str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
        )

    def preprocess(
        self, data: dd.DataFrame, prep_fn: Callable = preprocessing_flow
    ) -> dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(
            data, self.flow_config, self.client, prep_fn
        )
        return series_ddf

    def fit(
        self, data: dd.DataFrame, prep_fn: Callable = preprocessing_flow, **fit_kwargs
    ) -> 'DistributedForecast':
        """Perform the preprocessing and fit the model."""
        train_ddf = self.preprocess(data, prep_fn)
        X, y = train_ddf.drop(columns=['ds', 'y']), train_ddf.y
        self.model.fit(X, y, **fit_kwargs)
        return self

    def predict(
        self, horizon: int, predict_fn: Callable = predictions_flow, **predict_fn_kwargs
    ) -> dd.DataFrame:
        """Compute the predictions for the next `horizon` steps using `predict_fn`."""
        model_future = self.client.scatter(self.model.model_, broadcast=True)
        predictions_futures = self.client.map(
            predict_fn,
            self.ts,
            model=model_future,
            horizon=horizon,
            **predict_fn_kwargs,
        )
        meta = self.client.submit(lambda x: x.head(), predictions_futures[0]).result()
        return dd.from_delayed(
            predictions_futures, meta=meta, divisions=self.data_divisions
        )

17bc987ae8034ceee75c061d22b6c5f4cfa17ee6
<mask token>


class DistributedForecast(Forecast):
    <mask token>

    def __init__(self, model, flow_config: Dict, client: Optional[Client]=None
        ):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) ->str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
            )

    def preprocess(self, data: dd.DataFrame, prep_fn: Callable=
        preprocessing_flow) ->dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(data, self.flow_config,
            self.client, prep_fn)
        return series_ddf
    <mask token>
    <mask token>

<mask token>


class DistributedForecast(Forecast):
    <mask token>

    def __init__(self, model, flow_config: Dict, client: Optional[Client]=None
        ):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) ->str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
            )

    def preprocess(self, data: dd.DataFrame, prep_fn: Callable=
        preprocessing_flow) ->dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(data, self.flow_config,
            self.client, prep_fn)
        return series_ddf

    def fit(self, data: dd.DataFrame, prep_fn: Callable=preprocessing_flow,
        **fit_kwargs) ->'DistributedForecast':
        """Perform the preprocessing and fit the model."""
        train_ddf = self.preprocess(data, prep_fn)
        X, y = train_ddf.drop(columns=['ds', 'y']), train_ddf.y
        self.model.fit(X, y, **fit_kwargs)
        return self
    <mask token>

<mask token>


class DistributedForecast(Forecast):
    """Full pipeline encapsulation.

    Takes a model (`LGBMForecast` or `XGBForecast`), a flow configuration and a client."""

    def __init__(self, model, flow_config: Dict, client: Optional[Client]=None
        ):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) ->str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
            )

    def preprocess(self, data: dd.DataFrame, prep_fn: Callable=
        preprocessing_flow) ->dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(data, self.flow_config,
            self.client, prep_fn)
        return series_ddf

    def fit(self, data: dd.DataFrame, prep_fn: Callable=preprocessing_flow,
        **fit_kwargs) ->'DistributedForecast':
        """Perform the preprocessing and fit the model."""
        train_ddf = self.preprocess(data, prep_fn)
        X, y = train_ddf.drop(columns=['ds', 'y']), train_ddf.y
        self.model.fit(X, y, **fit_kwargs)
        return self

    def predict(self, horizon: int, predict_fn: Callable=predictions_flow,
        **predict_fn_kwargs) ->dd.DataFrame:
        """Compute the predictions for the next `horizon` steps using `predict_fn`."""
        model_future = self.client.scatter(self.model.model_, broadcast=True)
        predictions_futures = self.client.map(predict_fn, self.ts, model=
            model_future, horizon=horizon, **predict_fn_kwargs)
        meta = self.client.submit(lambda x: x.head(), predictions_futures[0]
            ).result()
        return dd.from_delayed(predictions_futures, meta=meta, divisions=
            self.data_divisions)

__all__ = ['DistributedForecast']
<mask token>


class DistributedForecast(Forecast):
    """Full pipeline encapsulation.

    Takes a model (`LGBMForecast` or `XGBForecast`), a flow configuration and a client."""

    def __init__(self, model, flow_config: Dict, client: Optional[Client]=None
        ):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) ->str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
            )

    def preprocess(self, data: dd.DataFrame, prep_fn: Callable=
        preprocessing_flow) ->dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(data, self.flow_config,
            self.client, prep_fn)
        return series_ddf

    def fit(self, data: dd.DataFrame, prep_fn: Callable=preprocessing_flow,
        **fit_kwargs) ->'DistributedForecast':
        """Perform the preprocessing and fit the model."""
        train_ddf = self.preprocess(data, prep_fn)
        X, y = train_ddf.drop(columns=['ds', 'y']), train_ddf.y
        self.model.fit(X, y, **fit_kwargs)
        return self

    def predict(self, horizon: int, predict_fn: Callable=predictions_flow,
        **predict_fn_kwargs) ->dd.DataFrame:
        """Compute the predictions for the next `horizon` steps using `predict_fn`."""
        model_future = self.client.scatter(self.model.model_, broadcast=True)
        predictions_futures = self.client.map(predict_fn, self.ts, model=
            model_future, horizon=horizon, **predict_fn_kwargs)
        meta = self.client.submit(lambda x: x.head(), predictions_futures[0]
            ).result()
        return dd.from_delayed(predictions_futures, meta=meta, divisions=
            self.data_divisions)

# AUTOGENERATED! DO NOT EDIT! File to edit: nbs/distributed.forecast.ipynb (unless otherwise specified).

__all__ = ['DistributedForecast']

# Cell
from typing import Callable, Dict, Optional

import dask.dataframe as dd
from dask.distributed import Client, default_client

from ..core import predictions_flow, preprocessing_flow
from ..forecast import Forecast
from .core import distributed_preprocess


# Cell
class DistributedForecast(Forecast):
    """Full pipeline encapsulation.

    Takes a model (`LGBMForecast` or `XGBForecast`), a flow configuration and a client."""

    def __init__(self, model, flow_config: Dict, client: Optional[Client] = None):
        self.model = model
        self.flow_config = flow_config
        self.client = client or default_client()
        self.model.client = self.client

    def __repr__(self) -> str:
        return (
            f'DistributedForecast(model={self.model}, flow_config={self.flow_config})'
        )

    def preprocess(
        self, data: dd.DataFrame, prep_fn: Callable = preprocessing_flow
    ) -> dd.DataFrame:
        """Applies `prep_fn(partition, **self.flow_config)` on each partition of `data`.

        Saves the resulting `TimeSeries` objects as well as the divisions in `data` for the forecasting step.
        Returns a dask dataframe with the computed features."""
        self.data_divisions = data.divisions
        self.ts, series_ddf = distributed_preprocess(
            data, self.flow_config, self.client, prep_fn
        )
        return series_ddf

    def fit(
        self, data: dd.DataFrame, prep_fn: Callable = preprocessing_flow, **fit_kwargs
    ) -> 'DistributedForecast':
        """Perform the preprocessing and fit the model."""
        train_ddf = self.preprocess(data, prep_fn)
        X, y = train_ddf.drop(columns=['ds', 'y']), train_ddf.y
        self.model.fit(X, y, **fit_kwargs)
        return self

    def predict(
        self, horizon: int, predict_fn: Callable = predictions_flow, **predict_fn_kwargs
    ) -> dd.DataFrame:
        """Compute the predictions for the next `horizon` steps using `predict_fn`."""
        model_future = self.client.scatter(self.model.model_, broadcast=True)
        predictions_futures = self.client.map(
            predict_fn,
            self.ts,
            model=model_future,
            horizon=horizon,
            **predict_fn_kwargs,
        )
        meta = self.client.submit(lambda x: x.head(), predictions_futures[0]).result()
        return dd.from_delayed(
            predictions_futures, meta=meta, divisions=self.data_divisions
        )
