#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#    http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#

"""
A wrapper for ResampledData to behave like pandas Resampler.
"""
from abc import ABCMeta, abstractmethod
from functools import partial
from typing import (
    Any,
    Generic,
    List,
    Optional,
)

import numpy as np
import pandas as pd
from pandas.tseries.frequencies import to_offset

from pyspark.sql import Column, functions as F
from pyspark.sql.internal import InternalFunction as SF
from pyspark.sql.types import (
    NumericType,
    StructField,
    TimestampNTZType,
    DataType,
)
from pyspark import pandas as ps  # For running doctests and reference resolution in PyCharm.
from pyspark.pandas._typing import FrameLike
from pyspark.pandas.frame import DataFrame
from pyspark.pandas.internal import (
    InternalField,
    InternalFrame,
    SPARK_DEFAULT_INDEX_NAME,
)
from pyspark.pandas.missing.resample import (
    MissingPandasLikeDataFrameResampler,
    MissingPandasLikeSeriesResampler,
)
from pyspark.pandas.series import Series, first_series
from pyspark.pandas.utils import (
    scol_for,
    verify_temp_column_name,
)


class Resampler(Generic[FrameLike], metaclass=ABCMeta):
    """
    Class for resampling datetimelike data, a groupby-like operation.

    It's easiest to use obj.resample(...) to use Resampler.

    Parameters
    ----------
    psdf : DataFrame

    Returns
    -------
    a Resampler of the appropriate type

    Notes
    -----
    After resampling, see aggregate, apply, and transform functions.
    """

    def __init__(
        self,
        psdf: DataFrame,
        resamplekey: Optional[Series],
        rule: str,
        closed: Optional[str] = None,
        label: Optional[str] = None,
        agg_columns: List[Series] = [],
    ):
        self._psdf = psdf
        self._resamplekey = resamplekey

        self._offset = to_offset(rule)

        if self._offset.rule_code not in ["A-DEC", "M", "ME", "D", "H", "h", "T", "min", "S", "s"]:
            raise ValueError("rule code {} is not supported".format(self._offset.rule_code))
        if not getattr(self._offset, "n") > 0:
            raise ValueError("rule offset must be positive")

        if closed is None:
            self._closed = "right" if self._offset.rule_code in ["A-DEC", "M", "ME"] else "left"
        elif closed in ["left", "right"]:
            self._closed = closed
        else:
            raise ValueError("invalid closed: '{}'".format(closed))

        if label is None:
            self._label = "right" if self._offset.rule_code in ["A-DEC", "M", "ME"] else "left"
        elif label in ["left", "right"]:
            self._label = label
        else:
            raise ValueError("invalid label: '{}'".format(label))

        self._agg_columns = agg_columns

    @property
    def _resamplekey_scol(self) -> Column:
        if self._resamplekey is None:
            return self._psdf.index.spark.column
        else:
            return self._resamplekey.spark.column

    @property
    def _resamplekey_type(self) -> DataType:
        if self._resamplekey is None:
            return self._psdf.index.spark.data_type
        else:
            return self._resamplekey.spark.data_type

    @property
    def _agg_columns_scols(self) -> List[Column]:
        return [s.spark.column for s in self._agg_columns]

    def _bin_timestamp(self, origin: pd.Timestamp, ts_scol: Column) -> Column:
        key_type = self._resamplekey_type
        origin_scol = F.lit(origin)
        (rule_code, n) = (self._offset.rule_code, getattr(self._offset, "n"))
        left_closed, right_closed = (self._closed == "left", self._closed == "right")
        left_labeled, right_labeled = (self._label == "left", self._label == "right")

        if rule_code == "A-DEC":
            assert (
                origin.month == 12
                and origin.day == 31
                and origin.hour == 0
                and origin.minute == 0
                and origin.second == 0
            )

            diff = F.year(ts_scol) - F.year(origin_scol)
            mod = F.lit(0) if n == 1 else (diff % n)
            edge_cond = (mod == 0) & (F.month(ts_scol) == 12) & (F.dayofmonth(ts_scol) == 31)

            edge_label = F.year(ts_scol)
            if left_closed and right_labeled:
                edge_label += n
            elif right_closed and left_labeled:
                edge_label -= n

            if left_labeled:
                non_edge_label = F.when(mod == 0, F.year(ts_scol) - n).otherwise(
                    F.year(ts_scol) - mod
                )
            else:
                non_edge_label = F.when(mod == 0, F.year(ts_scol)).otherwise(
                    F.year(ts_scol) - (mod - n)
                )

            ret = F.to_timestamp(
                F.make_date(
                    F.when(edge_cond, edge_label).otherwise(non_edge_label), F.lit(12), F.lit(31)
                )
            )

        elif rule_code in ["ME", "M"]:
            assert (
                origin.is_month_end
                and origin.hour == 0
                and origin.minute == 0
                and origin.second == 0
            )

            diff = (
                (F.year(ts_scol) - F.year(origin_scol)) * 12
                + F.month(ts_scol)
                - F.month(origin_scol)
            )
            mod = F.lit(0) if n == 1 else (diff % n)
            edge_cond = (mod == 0) & (F.dayofmonth(ts_scol) == F.dayofmonth(F.last_day(ts_scol)))

            truncated_ts_scol = F.date_trunc("MONTH", ts_scol)
            edge_label = truncated_ts_scol
            if left_closed and right_labeled:
                edge_label += SF.make_interval("MONTH", n)
            elif right_closed and left_labeled:
                edge_label -= SF.make_interval("MONTH", n)

            if left_labeled:
                non_edge_label = F.when(
                    mod == 0,
                    truncated_ts_scol - SF.make_interval("MONTH", n),
                ).otherwise(truncated_ts_scol - SF.make_interval("MONTH", mod))
            else:
                non_edge_label = F.when(mod == 0, truncated_ts_scol).otherwise(
                    truncated_ts_scol - SF.make_interval("MONTH", mod - n)
                )

            ret = F.to_timestamp(
                F.last_day(F.when(edge_cond, edge_label).otherwise(non_edge_label))
            )

        elif rule_code == "D":
            assert origin.hour == 0 and origin.minute == 0 and origin.second == 0

            if n == 1:
                # NOTE: the logic to process '1D' is different from the cases with n>1,
                # since hour/minute/second parts are taken into account to determine edges!
                edge_cond = (
                    (F.hour(ts_scol) == 0) & (F.minute(ts_scol) == 0) & (F.second(ts_scol) == 0)
                )

                if left_closed and left_labeled:
                    ret = F.date_trunc("DAY", ts_scol)
                elif left_closed and right_labeled:
                    ret = F.date_trunc("DAY", F.date_add(ts_scol, 1))
                elif right_closed and left_labeled:
                    ret = F.when(edge_cond, F.date_trunc("DAY", F.date_sub(ts_scol, 1))).otherwise(
                        F.date_trunc("DAY", ts_scol)
                    )
                else:
                    ret = F.when(edge_cond, F.date_trunc("DAY", ts_scol)).otherwise(
                        F.date_trunc("DAY", F.date_add(ts_scol, 1))
                    )

            else:
                diff = F.datediff(end=ts_scol, start=origin_scol)
                mod = diff % n

                edge_cond = mod == 0

                truncated_ts_scol = F.date_trunc("DAY", ts_scol)
                edge_label = truncated_ts_scol
                if left_closed and right_labeled:
                    edge_label = F.date_add(truncated_ts_scol, n)
                elif right_closed and left_labeled:
                    edge_label = F.date_sub(truncated_ts_scol, n)

                if left_labeled:
                    non_edge_label = F.date_sub(truncated_ts_scol, mod)
                else:
                    non_edge_label = F.date_sub(truncated_ts_scol, mod - n)

                ret = F.when(edge_cond, edge_label).otherwise(non_edge_label)

        elif rule_code in ["h", "min", "s", "H", "T", "S"]:
            unit_mapping = {
                "h": "HOUR",
                "min": "MINUTE",
                "s": "SECOND",
                "H": "HOUR",
                "T": "MINUTE",
                "S": "SECOND",
            }
            unit_str = unit_mapping[rule_code]

            truncated_ts_scol = F.date_trunc(unit_str, ts_scol)
            if isinstance(key_type, TimestampNTZType):
                truncated_ts_scol = F.to_timestamp_ntz(truncated_ts_scol)
            diff = F.timestamp_diff(unit_str, origin_scol, truncated_ts_scol)
            mod = F.lit(0) if n == 1 else (diff % F.lit(n))

            if rule_code in ["h", "H"]:
                assert origin.minute == 0 and origin.second == 0
                edge_cond = (mod == 0) & (F.minute(ts_scol) == 0) & (F.second(ts_scol) == 0)
            elif rule_code in ["min", "T"]:
                assert origin.second == 0
                edge_cond = (mod == 0) & (F.second(ts_scol) == 0)
            else:
                edge_cond = mod == 0

            edge_label = truncated_ts_scol
            if left_closed and right_labeled:
                edge_label += SF.make_interval(unit_str, n)
            elif right_closed and left_labeled:
                edge_label -= SF.make_interval(unit_str, n)

            if left_labeled:
                non_edge_label = F.when(mod == 0, truncated_ts_scol).otherwise(
                    truncated_ts_scol - SF.make_interval(unit_str, mod)
                )
            else:
                non_edge_label = F.when(
                    mod == 0,
                    truncated_ts_scol + SF.make_interval(unit_str, n),
                ).otherwise(truncated_ts_scol - SF.make_interval(unit_str, mod - n))

            ret = F.when(edge_cond, edge_label).otherwise(non_edge_label)

        else:
            raise ValueError("Got the unexpected unit {}".format(rule_code))

        if isinstance(key_type, TimestampNTZType):
            return F.to_timestamp_ntz(ret)
        else:
            return ret

    def _downsample(self, f: str) -> DataFrame:
        """
        Downsample the defined function.

        Parameters
        ----------
        how : string / mapped function
        **kwargs : kw args passed to how function
        """

        # a simple example to illustrate the computation:
        #   dates = [
        #         datetime(2012, 1, 2),
        #         datetime(2012, 5, 3),
        #         datetime(2022, 5, 3),
        #   ]
        #   index = pd.DatetimeIndex(dates)
        #   pdf = pd.DataFrame(np.array([1,2,3]), index=index, columns=['A'])
        #   pdf.resample('3Y').max()
        #                 A
        #   2012-12-31  2.0
        #   2015-12-31  NaN
        #   2018-12-31  NaN
        #   2021-12-31  NaN
        #   2024-12-31  3.0
        #
        # in this case:
        # 1, obtain one origin point to bin all timestamps, we can get one (2009-12-31)
        # from the minimum timestamp (2012-01-02);
        # 2, the default intervals for 'Y' are right-closed, so intervals are:
        # (2009-12-31, 2012-12-31], (2012-12-31, 2015-12-31], (2015-12-31, 2018-12-31], ...
        # 3, bin all timestamps, for example, 2022-05-03 belongs to interval
        # (2021-12-31, 2024-12-31], since the default label is 'right', label it with the right
        # edge 2024-12-31;
        # 4, some intervals maybe too large for this down sampling, so we need to pad the dataframe
        # to avoid missing some results, like: 2015-12-31, 2018-12-31 and 2021-12-31;
        # 5, union the binned dataframe and padded dataframe, and apply aggregation 'max' to get
        # the final results;

        # one action to obtain the range, in the future we may cache it in the index.
        ts_min, ts_max = (
            self._psdf._internal.spark_frame.select(
                F.min(self._resamplekey_scol), F.max(self._resamplekey_scol)
            )
            .toPandas()
            .iloc[0]
        )

        # the logic to obtain an origin point to bin the timestamps is too complex to follow,
        # here just use Pandas' resample on a 1-length series to get it.
        ts_origin = (
            pd.Series([0], index=[ts_min])
            .resample(rule=self._offset.freqstr, closed=self._closed, label="left")
            .sum()
            .index[0]
        )
        assert ts_origin <= ts_min

        bin_col_name = "__tmp_resample_bin_col__"
        bin_col_label = verify_temp_column_name(self._psdf, bin_col_name)
        bin_col_field = InternalField(
            dtype=np.dtype("datetime64[ns]"),
            struct_field=StructField(bin_col_name, self._resamplekey_type, True),
        )
        bin_scol = self._bin_timestamp(ts_origin, self._resamplekey_scol)

        agg_columns = [
            psser for psser in self._agg_columns if (isinstance(psser.spark.data_type, NumericType))
        ]
        assert len(agg_columns) > 0

        # in the binning side, label the timestamps according to the origin and the freq(rule)
        bin_sdf = self._psdf._internal.spark_frame.select(
            F.col(SPARK_DEFAULT_INDEX_NAME),
            bin_scol.alias(bin_col_name),
            *[psser.spark.column for psser in agg_columns],
        )

        # in the padding side, insert necessary points
        # again, directly apply Pandas' resample on a 2-length series to obtain the indices
        pad_sdf = (
            ps.from_pandas(
                pd.Series([0, 0], index=[ts_min, ts_max])
                .resample(rule=self._offset.freqstr, closed=self._closed, label=self._label)
                .sum()
                .index
            )
            ._internal.spark_frame.select(F.col(SPARK_DEFAULT_INDEX_NAME).alias(bin_col_name))
            .where((ts_min <= F.col(bin_col_name)) & (F.col(bin_col_name) <= ts_max))
        )

        # union the above two spark dataframes.
        sdf = bin_sdf.unionByName(pad_sdf, allowMissingColumns=True).where(
            ~F.isnull(F.col(bin_col_name))
        )

        internal = InternalFrame(
            spark_frame=sdf,
            index_spark_columns=[scol_for(sdf, SPARK_DEFAULT_INDEX_NAME)],
            data_spark_columns=[F.col(bin_col_name)]
            + [scol_for(sdf, psser._internal.data_spark_column_names[0]) for psser in agg_columns],
            column_labels=[bin_col_label] + [psser._column_label for psser in agg_columns],
            data_fields=[bin_col_field]
            + [psser._internal.data_fields[0].copy(nullable=True) for psser in agg_columns],
            column_label_names=self._psdf._internal.column_label_names,
        )
        psdf: DataFrame = DataFrame(internal)

        groupby = psdf.groupby(psdf._psser_for(bin_col_label), dropna=False)
        downsampled = getattr(groupby, f)()
        downsampled.index.name = None

        return downsampled

    @abstractmethod
    def _handle_output(self, psdf: DataFrame) -> FrameLike:
        pass

    def min(self) -> FrameLike:
        """
        Compute min of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").min().sort_index()
                           A         B
        2022-05-01  0.171162  0.338864
        2022-05-04  0.010527  0.561204
        2022-05-07       NaN       NaN
        2022-05-10  0.813726  0.745100
        """
        return self._handle_output(self._downsample("min"))

    def max(self) -> FrameLike:
        """
        Compute max of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").max().sort_index()
                           A         B
        2022-05-01  0.420538  0.859182
        2022-05-04  0.270533  0.691041
        2022-05-07       NaN       NaN
        2022-05-10  0.813726  0.745100
        """
        return self._handle_output(self._downsample("max"))

    def sum(self) -> FrameLike:
        """
        Compute sum of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").sum().sort_index()
                           A         B
        2022-05-01  0.800160  1.679727
        2022-05-04  0.281060  1.252245
        2022-05-07  0.000000  0.000000
        2022-05-10  0.813726  0.745100
        """
        return self._handle_output(self._downsample("sum").fillna(0.0))

    def mean(self) -> FrameLike:
        """
        Compute mean of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").mean().sort_index()
                           A         B
        2022-05-01  0.266720  0.559909
        2022-05-04  0.140530  0.626123
        2022-05-07       NaN       NaN
        2022-05-10  0.813726  0.745100
        """
        return self._handle_output(self._downsample("mean"))

    def std(self) -> FrameLike:
        """
        Compute std of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").std().sort_index()
                           A         B
        2022-05-01  0.134509  0.268835
        2022-05-04  0.183852  0.091809
        2022-05-07       NaN       NaN
        2022-05-10       NaN       NaN
        """
        return self._handle_output(self._downsample("std"))

    def var(self) -> FrameLike:
        """
        Compute var of resampled values.

        .. versionadded:: 3.4.0

        See Also
        --------
        pyspark.pandas.Series.groupby
        pyspark.pandas.DataFrame.groupby

        Examples
        --------
        >>> import numpy as np
        >>> from datetime import datetime
        >>> np.random.seed(22)
        >>> dates = [
        ...    datetime(2022, 5, 1, 4, 5, 6),
        ...    datetime(2022, 5, 3),
        ...    datetime(2022, 5, 3, 23, 59, 59),
        ...    datetime(2022, 5, 4),
        ...    pd.NaT,
        ...    datetime(2022, 5, 4, 0, 0, 1),
        ...    datetime(2022, 5, 11),
        ... ]
        >>> df = ps.DataFrame(
        ...    np.random.rand(len(dates), 2), index=pd.DatetimeIndex(dates), columns=["A", "B"]
        ... )
        >>> df
                                    A         B
        2022-05-01 04:05:06  0.208461  0.481681
        2022-05-03 00:00:00  0.420538  0.859182
        2022-05-03 23:59:59  0.171162  0.338864
        2022-05-04 00:00:00  0.270533  0.691041
        NaT                  0.220405  0.811951
        2022-05-04 00:00:01  0.010527  0.561204
        2022-05-11 00:00:00  0.813726  0.745100
        >>> df.resample("3D").var().sort_index()
                           A         B
        2022-05-01  0.018093  0.072272
        2022-05-04  0.033802  0.008429
        2022-05-07       NaN       NaN
        2022-05-10       NaN       NaN
        """
        return self._handle_output(self._downsample("var"))


class DataFrameResampler(Resampler[DataFrame]):
    def __init__(
        self,
        psdf: DataFrame,
        resamplekey: Optional[Series],
        rule: str,
        closed: Optional[str] = None,
        label: Optional[str] = None,
        agg_columns: List[Series] = [],
    ):
        super().__init__(
            psdf=psdf,
            resamplekey=resamplekey,
            rule=rule,
            closed=closed,
            label=label,
            agg_columns=agg_columns,
        )

    def __getattr__(self, item: str) -> Any:
        if hasattr(MissingPandasLikeDataFrameResampler, item):
            property_or_func = getattr(MissingPandasLikeDataFrameResampler, item)
            if isinstance(property_or_func, property):
                return property_or_func.fget(self)
            else:
                return partial(property_or_func, self)

    def _handle_output(self, psdf: DataFrame) -> DataFrame:
        return psdf


class SeriesResampler(Resampler[Series]):
    def __init__(
        self,
        psser: Series,
        resamplekey: Optional[Series],
        rule: str,
        closed: Optional[str] = None,
        label: Optional[str] = None,
        agg_columns: List[Series] = [],
    ):
        super().__init__(
            psdf=psser._psdf,
            resamplekey=resamplekey,
            rule=rule,
            closed=closed,
            label=label,
            agg_columns=agg_columns,
        )
        self._psser = psser

    def __getattr__(self, item: str) -> Any:
        if hasattr(MissingPandasLikeSeriesResampler, item):
            property_or_func = getattr(MissingPandasLikeSeriesResampler, item)
            if isinstance(property_or_func, property):
                return property_or_func.fget(self)
            else:
                return partial(property_or_func, self)

    def _handle_output(self, psdf: DataFrame) -> Series:
        return first_series(psdf).rename(self._psser.name)


def _test() -> None:
    import os
    import doctest
    import sys
    from pyspark.sql import SparkSession
    import pyspark.pandas.resample

    os.chdir(os.environ["SPARK_HOME"])

    globs = pyspark.pandas.resample.__dict__.copy()
    globs["ps"] = pyspark.pandas
    spark = (
        SparkSession.builder.master("local[4]")
        .appName("pyspark.pandas.resample tests")
        .getOrCreate()
    )
    (failure_count, test_count) = doctest.testmod(
        pyspark.pandas.resample,
        globs=globs,
        optionflags=doctest.ELLIPSIS | doctest.NORMALIZE_WHITESPACE,
    )
    spark.stop()
    if failure_count:
        sys.exit(-1)


if __name__ == "__main__":
    _test()