We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
⚠️ Upstream CI failed ⚠️
[Workflow Run URL](https://github.com/dask/dask/actions/runs/1304259797)

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `dask/dataframe/tseries/resample.py`
Content:
```
1 import numpy as np
2 import pandas as pd
3 from pandas.core.resample import Resampler as pd_Resampler
4 
5 from ...base import tokenize
6 from ...highlevelgraph import HighLevelGraph
7 from ...utils import derived_from
8 from .. import methods
9 from ..core import DataFrame, Series
10 
11 
12 def getnanos(rule):
13     try:
14         return getattr(rule, "nanos", None)
15     except ValueError:
16         return None
17 
18 
19 def _resample_series(
20     series,
21     start,
22     end,
23     reindex_closed,
24     rule,
25     resample_kwargs,
26     how,
27     fill_value,
28     how_args,
29     how_kwargs,
30 ):
31     out = getattr(series.resample(rule, **resample_kwargs), how)(
32         *how_args, **how_kwargs
33     )
34 
35     new_index = pd.date_range(
36         start.tz_localize(None),
37         end.tz_localize(None),
38         freq=rule,
39         closed=reindex_closed,
40         name=out.index.name,
41     ).tz_localize(start.tz, nonexistent="shift_forward")
42 
43     if not out.index.isin(new_index).all():
44         raise ValueError(
45             "Index is not contained within new index. This can often be "
46             "resolved by using larger partitions, or unambiguous "
47             "frequencies: 'Q', 'A'..."
48         )
49 
50     return out.reindex(new_index, fill_value=fill_value)
51 
52 
53 def _resample_bin_and_out_divs(divisions, rule, closed="left", label="left"):
54     rule = pd.tseries.frequencies.to_offset(rule)
55     g = pd.Grouper(freq=rule, how="count", closed=closed, label=label)
56 
57     # Determine bins to apply `how` to. Disregard labeling scheme.
58     divs = pd.Series(range(len(divisions)), index=divisions)
59     temp = divs.resample(rule, closed=closed, label="left").count()
60     tempdivs = temp.loc[temp > 0].index
61 
62     # Cleanup closed == 'right' and label == 'right'
63     res = pd.offsets.Nano() if hasattr(rule, "delta") else pd.offsets.Day()
64     if g.closed == "right":
65         newdivs = tempdivs + res
66     else:
67         newdivs = tempdivs
68     if g.label == "right":
69         outdivs = tempdivs + rule
70     else:
71         outdivs = tempdivs
72 
73     newdivs = methods.tolist(newdivs)
74     outdivs = methods.tolist(outdivs)
75 
76     # Adjust ends
77     if newdivs[0] < divisions[0]:
78         newdivs[0] = divisions[0]
79     if newdivs[-1] < divisions[-1]:
80         if len(newdivs) < len(divs):
81             setter = lambda a, val: a.append(val)
82         else:
83             setter = lambda a, val: a.__setitem__(-1, val)
84         setter(newdivs, divisions[-1] + res)
85         if outdivs[-1] > divisions[-1]:
86             setter(outdivs, outdivs[-1])
87         elif outdivs[-1] < divisions[-1]:
88             setter(outdivs, temp.index[-1])
89 
90     return tuple(map(pd.Timestamp, newdivs)), tuple(map(pd.Timestamp, outdivs))
91 
92 
93 class Resampler:
94     """Class for resampling timeseries data.
95 
96     This class is commonly encountered when using ``obj.resample(...)`` which
97     return ``Resampler`` objects.
98 
99     Parameters
100     ----------
101     obj : Dask DataFrame or Series
102         Data to be resampled.
103     rule : str, tuple, datetime.timedelta, DateOffset or None
104         The offset string or object representing the target conversion.
105     kwargs : optional
106         Keyword arguments passed to underlying pandas resampling function.
107 
108     Returns
109     -------
110     Resampler instance of the appropriate type
111     """
112 
113     def __init__(self, obj, rule, **kwargs):
114         if not obj.known_divisions:
115             msg = (
116                 "Can only resample dataframes with known divisions\n"
117                 "See https://docs.dask.org/en/latest/dataframe-design.html#partitions\n"
118                 "for more information."
119             )
120             raise ValueError(msg)
121         self.obj = obj
122         self._rule = pd.tseries.frequencies.to_offset(rule)
123         self._kwargs = kwargs
124 
125     def _agg(self, how, meta=None, fill_value=np.nan, how_args=(), how_kwargs={}):
126         """Aggregate using one or more operations
127 
128         Parameters
129         ----------
130         how : str
131             Name of aggregation operation
132         fill_value : scalar, optional
133             Value to use for missing values, applied during upsampling.
134             Default is NaN.
135         how_args : optional
136             Positional arguments for aggregation operation.
137         how_kwargs : optional
138             Keyword arguments for aggregation operation.
139 
140         Returns
141         -------
142         Dask DataFrame or Series
143         """
144         rule = self._rule
145         kwargs = self._kwargs
146         name = "resample-" + tokenize(
147             self.obj, rule, kwargs, how, *how_args, **how_kwargs
148         )
149 
150         # Create a grouper to determine closed and label conventions
151         newdivs, outdivs = _resample_bin_and_out_divs(
152             self.obj.divisions, rule, **kwargs
153         )
154 
155         # Repartition divs into bins. These won't match labels after mapping
156         partitioned = self.obj.repartition(newdivs, force=True)
157 
158         keys = partitioned.__dask_keys__()
159         dsk = {}
160 
161         args = zip(keys, outdivs, outdivs[1:], ["left"] * (len(keys) - 1) + [None])
162         for i, (k, s, e, c) in enumerate(args):
163             dsk[(name, i)] = (
164                 _resample_series,
165                 k,
166                 s,
167                 e,
168                 c,
169                 rule,
170                 kwargs,
171                 how,
172                 fill_value,
173                 list(how_args),
174                 how_kwargs,
175             )
176 
177         # Infer output metadata
178         meta_r = self.obj._meta_nonempty.resample(self._rule, **self._kwargs)
179         meta = getattr(meta_r, how)(*how_args, **how_kwargs)
180 
181         graph = HighLevelGraph.from_collections(name, dsk, dependencies=[partitioned])
182         if isinstance(meta, pd.DataFrame):
183             return DataFrame(graph, name, meta, outdivs)
184         return Series(graph, name, meta, outdivs)
185 
186     @derived_from(pd_Resampler)
187     def agg(self, agg_funcs, *args, **kwargs):
188         return self._agg("agg", how_args=(agg_funcs,) + args, how_kwargs=kwargs)
189 
190     @derived_from(pd_Resampler)
191     def count(self):
192         return self._agg("count", fill_value=0)
193 
194     @derived_from(pd_Resampler)
195     def first(self):
196         return self._agg("first")
197 
198     @derived_from(pd_Resampler)
199     def last(self):
200         return self._agg("last")
201 
202     @derived_from(pd_Resampler)
203     def mean(self):
204         return self._agg("mean")
205 
206     @derived_from(pd_Resampler)
207     def min(self):
208         return self._agg("min")
209 
210     @derived_from(pd_Resampler)
211     def median(self):
212         return self._agg("median")
213 
214     @derived_from(pd_Resampler)
215     def max(self):
216         return self._agg("max")
217 
218     @derived_from(pd_Resampler)
219     def nunique(self):
220         return self._agg("nunique", fill_value=0)
221 
222     @derived_from(pd_Resampler)
223     def ohlc(self):
224         return self._agg("ohlc")
225 
226     @derived_from(pd_Resampler)
227     def prod(self):
228         return self._agg("prod")
229 
230     @derived_from(pd_Resampler)
231     def sem(self):
232         return self._agg("sem")
233 
234     @derived_from(pd_Resampler)
235     def std(self):
236         return self._agg("std")
237 
238     @derived_from(pd_Resampler)
239     def size(self):
240         return self._agg("size", fill_value=0)
241 
242     @derived_from(pd_Resampler)
243     def sum(self):
244         return self._agg("sum", fill_value=0)
245 
246     @derived_from(pd_Resampler)
247     def var(self):
248         return self._agg("var")
249 
250     @derived_from(pd_Resampler)
251     def quantile(self):
252         return self._agg("quantile")
253
```

Path: `dask/dataframe/_compat.py`
Content:
```
1 import string
2 
3 import numpy as np
4 import pandas as pd
5 from packaging.version import parse as parse_version
6 
7 PANDAS_VERSION = parse_version(pd.__version__)
8 PANDAS_GT_104 = PANDAS_VERSION >= parse_version("1.0.4")
9 PANDAS_GT_110 = PANDAS_VERSION >= parse_version("1.1.0")
10 PANDAS_GT_120 = PANDAS_VERSION >= parse_version("1.2.0")
11 PANDAS_GT_121 = PANDAS_VERSION >= parse_version("1.2.1")
12 PANDAS_GT_130 = PANDAS_VERSION >= parse_version("1.3.0")
13 PANDAS_GT_131 = PANDAS_VERSION >= parse_version("1.3.1")
14 PANDAS_GT_133 = PANDAS_VERSION >= parse_version("1.3.3")
15 
16 import pandas.testing as tm
17 
18 
19 def assert_categorical_equal(left, right, *args, **kwargs):
20     tm.assert_extension_array_equal(left, right, *args, **kwargs)
21     assert pd.api.types.is_categorical_dtype(
22         left.dtype
23     ), "{} is not categorical dtype".format(left)
24     assert pd.api.types.is_categorical_dtype(
25         right.dtype
26     ), "{} is not categorical dtype".format(right)
27 
28 
29 def assert_numpy_array_equal(left, right):
30     left_na = pd.isna(left)
31     right_na = pd.isna(right)
32     np.testing.assert_array_equal(left_na, right_na)
33 
34     left_valid = left[~left_na]
35     right_valid = right[~right_na]
36     np.testing.assert_array_equal(left_valid, right_valid)
37 
38 
39 def makeDataFrame():
40     data = np.random.randn(30, 4)
41     index = list(string.ascii_letters)[:30]
42     return pd.DataFrame(data, index=index, columns=list("ABCD"))
43 
44 
45 def makeTimeDataFrame():
46     data = makeDataFrame()
47     data.index = makeDateIndex()
48     return data
49 
50 
51 def makeTimeSeries():
52     return makeTimeDataFrame()["A"]
53 
54 
55 def makeDateIndex(k=30, freq="B"):
56     return pd.date_range("2000", periods=k, freq=freq)
57 
58 
59 def makeTimedeltaIndex(k=30, freq="D"):
60     return pd.timedelta_range("1 day", periods=k, freq=freq)
61 
62 
63 def makeMissingDataframe():
64     df = makeDataFrame()
65     data = df.values
66     data = np.where(data > 1, np.nan, data)
67     return pd.DataFrame(data, index=df.index, columns=df.columns)
68 
69 
70 def makeMixedDataFrame():
71     df = pd.DataFrame(
72         {
73             "A": [0.0, 1, 2, 3, 4],
74             "B": [0.0, 1, 0, 1, 0],
75             "C": ["foo{}".format(i) for i in range(5)],
76             "D": pd.date_range("2009-01-01", periods=5),
77         }
78     )
79     return df
80
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/dask/dataframe/_compat.py b/dask/dataframe/_compat.py
--- a/dask/dataframe/_compat.py
+++ b/dask/dataframe/_compat.py
@@ -12,6 +12,7 @@
 PANDAS_GT_130 = PANDAS_VERSION >= parse_version("1.3.0")
 PANDAS_GT_131 = PANDAS_VERSION >= parse_version("1.3.1")
 PANDAS_GT_133 = PANDAS_VERSION >= parse_version("1.3.3")
+PANDAS_GT_140 = PANDAS_VERSION.release == (1, 4, 0)  # include pre-release
 
 import pandas.testing as tm
 
diff --git a/dask/dataframe/tseries/resample.py b/dask/dataframe/tseries/resample.py
--- a/dask/dataframe/tseries/resample.py
+++ b/dask/dataframe/tseries/resample.py
@@ -6,6 +6,7 @@
 from ...highlevelgraph import HighLevelGraph
 from ...utils import derived_from
 from .. import methods
+from .._compat import PANDAS_GT_140
 from ..core import DataFrame, Series
 
 
@@ -32,11 +33,20 @@
         *how_args, **how_kwargs
     )
 
+    if PANDAS_GT_140:
+        if reindex_closed is None:
+            inclusive = "both"
+        else:
+            inclusive = reindex_closed
+        closed_kwargs = {"inclusive": inclusive}
+    else:
+        closed_kwargs = {"closed": reindex_closed}
+
     new_index = pd.date_range(
         start.tz_localize(None),
         end.tz_localize(None),
         freq=rule,
-        closed=reindex_closed,
+        **closed_kwargs,
         name=out.index.name,
     ).tz_localize(start.tz, nonexistent="shift_forward")

{"golden_diff": "diff --git a/dask/dataframe/_compat.py b/dask/dataframe/_compat.py\n--- a/dask/dataframe/_compat.py\n+++ b/dask/dataframe/_compat.py\n@@ -12,6 +12,7 @@\n PANDAS_GT_130 = PANDAS_VERSION >= parse_version(\"1.3.0\")\n PANDAS_GT_131 = PANDAS_VERSION >= parse_version(\"1.3.1\")\n PANDAS_GT_133 = PANDAS_VERSION >= parse_version(\"1.3.3\")\n+PANDAS_GT_140 = PANDAS_VERSION.release == (1, 4, 0)  # include pre-release\n \n import pandas.testing as tm\n \ndiff --git a/dask/dataframe/tseries/resample.py b/dask/dataframe/tseries/resample.py\n--- a/dask/dataframe/tseries/resample.py\n+++ b/dask/dataframe/tseries/resample.py\n@@ -6,6 +6,7 @@\n from ...highlevelgraph import HighLevelGraph\n from ...utils import derived_from\n from .. import methods\n+from .._compat import PANDAS_GT_140\n from ..core import DataFrame, Series\n \n \n@@ -32,11 +33,20 @@\n         *how_args, **how_kwargs\n     )\n \n+    if PANDAS_GT_140:\n+        if reindex_closed is None:\n+            inclusive = \"both\"\n+        else:\n+            inclusive = reindex_closed\n+        closed_kwargs = {\"inclusive\": inclusive}\n+    else:\n+        closed_kwargs = {\"closed\": reindex_closed}\n+\n     new_index = pd.date_range(\n         start.tz_localize(None),\n         end.tz_localize(None),\n         freq=rule,\n-        closed=reindex_closed,\n+        **closed_kwargs,\n         name=out.index.name,\n     ).tz_localize(start.tz, nonexistent=\"shift_forward\")\n", "issue": "\u26a0\ufe0f Upstream CI failed \u26a0\ufe0f\n[Workflow Run URL](https://github.com/dask/dask/actions/runs/1304259797)\n", "before_files": [{"content": "import numpy as np\nimport pandas as pd\nfrom pandas.core.resample import Resampler as pd_Resampler\n\nfrom ...base import tokenize\nfrom ...highlevelgraph import HighLevelGraph\nfrom ...utils import derived_from\nfrom .. import methods\nfrom ..core import DataFrame, Series\n\n\ndef getnanos(rule):\n    try:\n        return getattr(rule, \"nanos\", None)\n    except ValueError:\n        return None\n\n\ndef _resample_series(\n    series,\n    start,\n    end,\n    reindex_closed,\n    rule,\n    resample_kwargs,\n    how,\n    fill_value,\n    how_args,\n    how_kwargs,\n):\n    out = getattr(series.resample(rule, **resample_kwargs), how)(\n        *how_args, **how_kwargs\n    )\n\n    new_index = pd.date_range(\n        start.tz_localize(None),\n        end.tz_localize(None),\n        freq=rule,\n        closed=reindex_closed,\n        name=out.index.name,\n    ).tz_localize(start.tz, nonexistent=\"shift_forward\")\n\n    if not out.index.isin(new_index).all():\n        raise ValueError(\n            \"Index is not contained within new index. This can often be \"\n            \"resolved by using larger partitions, or unambiguous \"\n            \"frequencies: 'Q', 'A'...\"\n        )\n\n    return out.reindex(new_index, fill_value=fill_value)\n\n\ndef _resample_bin_and_out_divs(divisions, rule, closed=\"left\", label=\"left\"):\n    rule = pd.tseries.frequencies.to_offset(rule)\n    g = pd.Grouper(freq=rule, how=\"count\", closed=closed, label=label)\n\n    # Determine bins to apply `how` to. Disregard labeling scheme.\n    divs = pd.Series(range(len(divisions)), index=divisions)\n    temp = divs.resample(rule, closed=closed, label=\"left\").count()\n    tempdivs = temp.loc[temp > 0].index\n\n    # Cleanup closed == 'right' and label == 'right'\n    res = pd.offsets.Nano() if hasattr(rule, \"delta\") else pd.offsets.Day()\n    if g.closed == \"right\":\n        newdivs = tempdivs + res\n    else:\n        newdivs = tempdivs\n    if g.label == \"right\":\n        outdivs = tempdivs + rule\n    else:\n        outdivs = tempdivs\n\n    newdivs = methods.tolist(newdivs)\n    outdivs = methods.tolist(outdivs)\n\n    # Adjust ends\n    if newdivs[0] < divisions[0]:\n        newdivs[0] = divisions[0]\n    if newdivs[-1] < divisions[-1]:\n        if len(newdivs) < len(divs):\n            setter = lambda a, val: a.append(val)\n        else:\n            setter = lambda a, val: a.__setitem__(-1, val)\n        setter(newdivs, divisions[-1] + res)\n        if outdivs[-1] > divisions[-1]:\n            setter(outdivs, outdivs[-1])\n        elif outdivs[-1] < divisions[-1]:\n            setter(outdivs, temp.index[-1])\n\n    return tuple(map(pd.Timestamp, newdivs)), tuple(map(pd.Timestamp, outdivs))\n\n\nclass Resampler:\n    \"\"\"Class for resampling timeseries data.\n\n    This class is commonly encountered when using ``obj.resample(...)`` which\n    return ``Resampler`` objects.\n\n    Parameters\n    ----------\n    obj : Dask DataFrame or Series\n        Data to be resampled.\n    rule : str, tuple, datetime.timedelta, DateOffset or None\n        The offset string or object representing the target conversion.\n    kwargs : optional\n        Keyword arguments passed to underlying pandas resampling function.\n\n    Returns\n    -------\n    Resampler instance of the appropriate type\n    \"\"\"\n\n    def __init__(self, obj, rule, **kwargs):\n        if not obj.known_divisions:\n            msg = (\n                \"Can only resample dataframes with known divisions\\n\"\n                \"See https://docs.dask.org/en/latest/dataframe-design.html#partitions\\n\"\n                \"for more information.\"\n            )\n            raise ValueError(msg)\n        self.obj = obj\n        self._rule = pd.tseries.frequencies.to_offset(rule)\n        self._kwargs = kwargs\n\n    def _agg(self, how, meta=None, fill_value=np.nan, how_args=(), how_kwargs={}):\n        \"\"\"Aggregate using one or more operations\n\n        Parameters\n        ----------\n        how : str\n            Name of aggregation operation\n        fill_value : scalar, optional\n            Value to use for missing values, applied during upsampling.\n            Default is NaN.\n        how_args : optional\n            Positional arguments for aggregation operation.\n        how_kwargs : optional\n            Keyword arguments for aggregation operation.\n\n        Returns\n        -------\n        Dask DataFrame or Series\n        \"\"\"\n        rule = self._rule\n        kwargs = self._kwargs\n        name = \"resample-\" + tokenize(\n            self.obj, rule, kwargs, how, *how_args, **how_kwargs\n        )\n\n        # Create a grouper to determine closed and label conventions\n        newdivs, outdivs = _resample_bin_and_out_divs(\n            self.obj.divisions, rule, **kwargs\n        )\n\n        # Repartition divs into bins. These won't match labels after mapping\n        partitioned = self.obj.repartition(newdivs, force=True)\n\n        keys = partitioned.__dask_keys__()\n        dsk = {}\n\n        args = zip(keys, outdivs, outdivs[1:], [\"left\"] * (len(keys) - 1) + [None])\n        for i, (k, s, e, c) in enumerate(args):\n            dsk[(name, i)] = (\n                _resample_series,\n                k,\n                s,\n                e,\n                c,\n                rule,\n                kwargs,\n                how,\n                fill_value,\n                list(how_args),\n                how_kwargs,\n            )\n\n        # Infer output metadata\n        meta_r = self.obj._meta_nonempty.resample(self._rule, **self._kwargs)\n        meta = getattr(meta_r, how)(*how_args, **how_kwargs)\n\n        graph = HighLevelGraph.from_collections(name, dsk, dependencies=[partitioned])\n        if isinstance(meta, pd.DataFrame):\n            return DataFrame(graph, name, meta, outdivs)\n        return Series(graph, name, meta, outdivs)\n\n    @derived_from(pd_Resampler)\n    def agg(self, agg_funcs, *args, **kwargs):\n        return self._agg(\"agg\", how_args=(agg_funcs,) + args, how_kwargs=kwargs)\n\n    @derived_from(pd_Resampler)\n    def count(self):\n        return self._agg(\"count\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def first(self):\n        return self._agg(\"first\")\n\n    @derived_from(pd_Resampler)\n    def last(self):\n        return self._agg(\"last\")\n\n    @derived_from(pd_Resampler)\n    def mean(self):\n        return self._agg(\"mean\")\n\n    @derived_from(pd_Resampler)\n    def min(self):\n        return self._agg(\"min\")\n\n    @derived_from(pd_Resampler)\n    def median(self):\n        return self._agg(\"median\")\n\n    @derived_from(pd_Resampler)\n    def max(self):\n        return self._agg(\"max\")\n\n    @derived_from(pd_Resampler)\n    def nunique(self):\n        return self._agg(\"nunique\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def ohlc(self):\n        return self._agg(\"ohlc\")\n\n    @derived_from(pd_Resampler)\n    def prod(self):\n        return self._agg(\"prod\")\n\n    @derived_from(pd_Resampler)\n    def sem(self):\n        return self._agg(\"sem\")\n\n    @derived_from(pd_Resampler)\n    def std(self):\n        return self._agg(\"std\")\n\n    @derived_from(pd_Resampler)\n    def size(self):\n        return self._agg(\"size\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def sum(self):\n        return self._agg(\"sum\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def var(self):\n        return self._agg(\"var\")\n\n    @derived_from(pd_Resampler)\n    def quantile(self):\n        return self._agg(\"quantile\")\n", "path": "dask/dataframe/tseries/resample.py"}, {"content": "import string\n\nimport numpy as np\nimport pandas as pd\nfrom packaging.version import parse as parse_version\n\nPANDAS_VERSION = parse_version(pd.__version__)\nPANDAS_GT_104 = PANDAS_VERSION >= parse_version(\"1.0.4\")\nPANDAS_GT_110 = PANDAS_VERSION >= parse_version(\"1.1.0\")\nPANDAS_GT_120 = PANDAS_VERSION >= parse_version(\"1.2.0\")\nPANDAS_GT_121 = PANDAS_VERSION >= parse_version(\"1.2.1\")\nPANDAS_GT_130 = PANDAS_VERSION >= parse_version(\"1.3.0\")\nPANDAS_GT_131 = PANDAS_VERSION >= parse_version(\"1.3.1\")\nPANDAS_GT_133 = PANDAS_VERSION >= parse_version(\"1.3.3\")\n\nimport pandas.testing as tm\n\n\ndef assert_categorical_equal(left, right, *args, **kwargs):\n    tm.assert_extension_array_equal(left, right, *args, **kwargs)\n    assert pd.api.types.is_categorical_dtype(\n        left.dtype\n    ), \"{} is not categorical dtype\".format(left)\n    assert pd.api.types.is_categorical_dtype(\n        right.dtype\n    ), \"{} is not categorical dtype\".format(right)\n\n\ndef assert_numpy_array_equal(left, right):\n    left_na = pd.isna(left)\n    right_na = pd.isna(right)\n    np.testing.assert_array_equal(left_na, right_na)\n\n    left_valid = left[~left_na]\n    right_valid = right[~right_na]\n    np.testing.assert_array_equal(left_valid, right_valid)\n\n\ndef makeDataFrame():\n    data = np.random.randn(30, 4)\n    index = list(string.ascii_letters)[:30]\n    return pd.DataFrame(data, index=index, columns=list(\"ABCD\"))\n\n\ndef makeTimeDataFrame():\n    data = makeDataFrame()\n    data.index = makeDateIndex()\n    return data\n\n\ndef makeTimeSeries():\n    return makeTimeDataFrame()[\"A\"]\n\n\ndef makeDateIndex(k=30, freq=\"B\"):\n    return pd.date_range(\"2000\", periods=k, freq=freq)\n\n\ndef makeTimedeltaIndex(k=30, freq=\"D\"):\n    return pd.timedelta_range(\"1 day\", periods=k, freq=freq)\n\n\ndef makeMissingDataframe():\n    df = makeDataFrame()\n    data = df.values\n    data = np.where(data > 1, np.nan, data)\n    return pd.DataFrame(data, index=df.index, columns=df.columns)\n\n\ndef makeMixedDataFrame():\n    df = pd.DataFrame(\n        {\n            \"A\": [0.0, 1, 2, 3, 4],\n            \"B\": [0.0, 1, 0, 1, 0],\n            \"C\": [\"foo{}\".format(i) for i in range(5)],\n            \"D\": pd.date_range(\"2009-01-01\", periods=5),\n        }\n    )\n    return df\n", "path": "dask/dataframe/_compat.py"}], "after_files": [{"content": "import numpy as np\nimport pandas as pd\nfrom pandas.core.resample import Resampler as pd_Resampler\n\nfrom ...base import tokenize\nfrom ...highlevelgraph import HighLevelGraph\nfrom ...utils import derived_from\nfrom .. import methods\nfrom .._compat import PANDAS_GT_140\nfrom ..core import DataFrame, Series\n\n\ndef getnanos(rule):\n    try:\n        return getattr(rule, \"nanos\", None)\n    except ValueError:\n        return None\n\n\ndef _resample_series(\n    series,\n    start,\n    end,\n    reindex_closed,\n    rule,\n    resample_kwargs,\n    how,\n    fill_value,\n    how_args,\n    how_kwargs,\n):\n    out = getattr(series.resample(rule, **resample_kwargs), how)(\n        *how_args, **how_kwargs\n    )\n\n    if PANDAS_GT_140:\n        if reindex_closed is None:\n            inclusive = \"both\"\n        else:\n            inclusive = reindex_closed\n        closed_kwargs = {\"inclusive\": inclusive}\n    else:\n        closed_kwargs = {\"closed\": reindex_closed}\n\n    new_index = pd.date_range(\n        start.tz_localize(None),\n        end.tz_localize(None),\n        freq=rule,\n        **closed_kwargs,\n        name=out.index.name,\n    ).tz_localize(start.tz, nonexistent=\"shift_forward\")\n\n    if not out.index.isin(new_index).all():\n        raise ValueError(\n            \"Index is not contained within new index. This can often be \"\n            \"resolved by using larger partitions, or unambiguous \"\n            \"frequencies: 'Q', 'A'...\"\n        )\n\n    return out.reindex(new_index, fill_value=fill_value)\n\n\ndef _resample_bin_and_out_divs(divisions, rule, closed=\"left\", label=\"left\"):\n    rule = pd.tseries.frequencies.to_offset(rule)\n    g = pd.Grouper(freq=rule, how=\"count\", closed=closed, label=label)\n\n    # Determine bins to apply `how` to. Disregard labeling scheme.\n    divs = pd.Series(range(len(divisions)), index=divisions)\n    temp = divs.resample(rule, closed=closed, label=\"left\").count()\n    tempdivs = temp.loc[temp > 0].index\n\n    # Cleanup closed == 'right' and label == 'right'\n    res = pd.offsets.Nano() if hasattr(rule, \"delta\") else pd.offsets.Day()\n    if g.closed == \"right\":\n        newdivs = tempdivs + res\n    else:\n        newdivs = tempdivs\n    if g.label == \"right\":\n        outdivs = tempdivs + rule\n    else:\n        outdivs = tempdivs\n\n    newdivs = methods.tolist(newdivs)\n    outdivs = methods.tolist(outdivs)\n\n    # Adjust ends\n    if newdivs[0] < divisions[0]:\n        newdivs[0] = divisions[0]\n    if newdivs[-1] < divisions[-1]:\n        if len(newdivs) < len(divs):\n            setter = lambda a, val: a.append(val)\n        else:\n            setter = lambda a, val: a.__setitem__(-1, val)\n        setter(newdivs, divisions[-1] + res)\n        if outdivs[-1] > divisions[-1]:\n            setter(outdivs, outdivs[-1])\n        elif outdivs[-1] < divisions[-1]:\n            setter(outdivs, temp.index[-1])\n\n    return tuple(map(pd.Timestamp, newdivs)), tuple(map(pd.Timestamp, outdivs))\n\n\nclass Resampler:\n    \"\"\"Class for resampling timeseries data.\n\n    This class is commonly encountered when using ``obj.resample(...)`` which\n    return ``Resampler`` objects.\n\n    Parameters\n    ----------\n    obj : Dask DataFrame or Series\n        Data to be resampled.\n    rule : str, tuple, datetime.timedelta, DateOffset or None\n        The offset string or object representing the target conversion.\n    kwargs : optional\n        Keyword arguments passed to underlying pandas resampling function.\n\n    Returns\n    -------\n    Resampler instance of the appropriate type\n    \"\"\"\n\n    def __init__(self, obj, rule, **kwargs):\n        if not obj.known_divisions:\n            msg = (\n                \"Can only resample dataframes with known divisions\\n\"\n                \"See https://docs.dask.org/en/latest/dataframe-design.html#partitions\\n\"\n                \"for more information.\"\n            )\n            raise ValueError(msg)\n        self.obj = obj\n        self._rule = pd.tseries.frequencies.to_offset(rule)\n        self._kwargs = kwargs\n\n    def _agg(self, how, meta=None, fill_value=np.nan, how_args=(), how_kwargs={}):\n        \"\"\"Aggregate using one or more operations\n\n        Parameters\n        ----------\n        how : str\n            Name of aggregation operation\n        fill_value : scalar, optional\n            Value to use for missing values, applied during upsampling.\n            Default is NaN.\n        how_args : optional\n            Positional arguments for aggregation operation.\n        how_kwargs : optional\n            Keyword arguments for aggregation operation.\n\n        Returns\n        -------\n        Dask DataFrame or Series\n        \"\"\"\n        rule = self._rule\n        kwargs = self._kwargs\n        name = \"resample-\" + tokenize(\n            self.obj, rule, kwargs, how, *how_args, **how_kwargs\n        )\n\n        # Create a grouper to determine closed and label conventions\n        newdivs, outdivs = _resample_bin_and_out_divs(\n            self.obj.divisions, rule, **kwargs\n        )\n\n        # Repartition divs into bins. These won't match labels after mapping\n        partitioned = self.obj.repartition(newdivs, force=True)\n\n        keys = partitioned.__dask_keys__()\n        dsk = {}\n\n        args = zip(keys, outdivs, outdivs[1:], [\"left\"] * (len(keys) - 1) + [None])\n        for i, (k, s, e, c) in enumerate(args):\n            dsk[(name, i)] = (\n                _resample_series,\n                k,\n                s,\n                e,\n                c,\n                rule,\n                kwargs,\n                how,\n                fill_value,\n                list(how_args),\n                how_kwargs,\n            )\n\n        # Infer output metadata\n        meta_r = self.obj._meta_nonempty.resample(self._rule, **self._kwargs)\n        meta = getattr(meta_r, how)(*how_args, **how_kwargs)\n\n        graph = HighLevelGraph.from_collections(name, dsk, dependencies=[partitioned])\n        if isinstance(meta, pd.DataFrame):\n            return DataFrame(graph, name, meta, outdivs)\n        return Series(graph, name, meta, outdivs)\n\n    @derived_from(pd_Resampler)\n    def agg(self, agg_funcs, *args, **kwargs):\n        return self._agg(\"agg\", how_args=(agg_funcs,) + args, how_kwargs=kwargs)\n\n    @derived_from(pd_Resampler)\n    def count(self):\n        return self._agg(\"count\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def first(self):\n        return self._agg(\"first\")\n\n    @derived_from(pd_Resampler)\n    def last(self):\n        return self._agg(\"last\")\n\n    @derived_from(pd_Resampler)\n    def mean(self):\n        return self._agg(\"mean\")\n\n    @derived_from(pd_Resampler)\n    def min(self):\n        return self._agg(\"min\")\n\n    @derived_from(pd_Resampler)\n    def median(self):\n        return self._agg(\"median\")\n\n    @derived_from(pd_Resampler)\n    def max(self):\n        return self._agg(\"max\")\n\n    @derived_from(pd_Resampler)\n    def nunique(self):\n        return self._agg(\"nunique\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def ohlc(self):\n        return self._agg(\"ohlc\")\n\n    @derived_from(pd_Resampler)\n    def prod(self):\n        return self._agg(\"prod\")\n\n    @derived_from(pd_Resampler)\n    def sem(self):\n        return self._agg(\"sem\")\n\n    @derived_from(pd_Resampler)\n    def std(self):\n        return self._agg(\"std\")\n\n    @derived_from(pd_Resampler)\n    def size(self):\n        return self._agg(\"size\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def sum(self):\n        return self._agg(\"sum\", fill_value=0)\n\n    @derived_from(pd_Resampler)\n    def var(self):\n        return self._agg(\"var\")\n\n    @derived_from(pd_Resampler)\n    def quantile(self):\n        return self._agg(\"quantile\")\n", "path": "dask/dataframe/tseries/resample.py"}, {"content": "import string\n\nimport numpy as np\nimport pandas as pd\nfrom packaging.version import parse as parse_version\n\nPANDAS_VERSION = parse_version(pd.__version__)\nPANDAS_GT_104 = PANDAS_VERSION >= parse_version(\"1.0.4\")\nPANDAS_GT_110 = PANDAS_VERSION >= parse_version(\"1.1.0\")\nPANDAS_GT_120 = PANDAS_VERSION >= parse_version(\"1.2.0\")\nPANDAS_GT_121 = PANDAS_VERSION >= parse_version(\"1.2.1\")\nPANDAS_GT_130 = PANDAS_VERSION >= parse_version(\"1.3.0\")\nPANDAS_GT_131 = PANDAS_VERSION >= parse_version(\"1.3.1\")\nPANDAS_GT_133 = PANDAS_VERSION >= parse_version(\"1.3.3\")\nPANDAS_GT_140 = PANDAS_VERSION.release == (1, 4, 0)  # include pre-release\n\nimport pandas.testing as tm\n\n\ndef assert_categorical_equal(left, right, *args, **kwargs):\n    tm.assert_extension_array_equal(left, right, *args, **kwargs)\n    assert pd.api.types.is_categorical_dtype(\n        left.dtype\n    ), \"{} is not categorical dtype\".format(left)\n    assert pd.api.types.is_categorical_dtype(\n        right.dtype\n    ), \"{} is not categorical dtype\".format(right)\n\n\ndef assert_numpy_array_equal(left, right):\n    left_na = pd.isna(left)\n    right_na = pd.isna(right)\n    np.testing.assert_array_equal(left_na, right_na)\n\n    left_valid = left[~left_na]\n    right_valid = right[~right_na]\n    np.testing.assert_array_equal(left_valid, right_valid)\n\n\ndef makeDataFrame():\n    data = np.random.randn(30, 4)\n    index = list(string.ascii_letters)[:30]\n    return pd.DataFrame(data, index=index, columns=list(\"ABCD\"))\n\n\ndef makeTimeDataFrame():\n    data = makeDataFrame()\n    data.index = makeDateIndex()\n    return data\n\n\ndef makeTimeSeries():\n    return makeTimeDataFrame()[\"A\"]\n\n\ndef makeDateIndex(k=30, freq=\"B\"):\n    return pd.date_range(\"2000\", periods=k, freq=freq)\n\n\ndef makeTimedeltaIndex(k=30, freq=\"D\"):\n    return pd.timedelta_range(\"1 day\", periods=k, freq=freq)\n\n\ndef makeMissingDataframe():\n    df = makeDataFrame()\n    data = df.values\n    data = np.where(data > 1, np.nan, data)\n    return pd.DataFrame(data, index=df.index, columns=df.columns)\n\n\ndef makeMixedDataFrame():\n    df = pd.DataFrame(\n        {\n            \"A\": [0.0, 1, 2, 3, 4],\n            \"B\": [0.0, 1, 0, 1, 0],\n            \"C\": [\"foo{}\".format(i) for i in range(5)],\n            \"D\": pd.date_range(\"2009-01-01\", periods=5),\n        }\n    )\n    return df\n", "path": "dask/dataframe/_compat.py"}]}