API: Timedelta constructor pytimedelta, Tick preserve reso (pandas-dev#48918)

* BUG: Timedelta.__new__

* remove assertion

* GH refs

* API: Timedelta(td64_obj) retain resolution

* API: Timedelta constructor pytimedelta, Tick preserve reso

* remove debugging variable

* remove duplicate

Author: jbrockmendel

pandas/_libs/tslibs/timedeltas.pyx

@@ -341,8 +341,9 @@ cdef convert_to_timedelta64(object ts, str unit):
     elif isinstance(ts, _Timedelta):
         # already in the proper format
         if ts._reso != NPY_FR_ns:
-            raise NotImplementedError
-        ts = np.timedelta64(ts.value, "ns")
+            ts = ts._as_unit("ns").asm8
+        else:
+            ts = np.timedelta64(ts.value, "ns")
     elif is_timedelta64_object(ts):
         ts = ensure_td64ns(ts)
     elif is_integer_object(ts):
@@ -1706,7 +1707,13 @@ class Timedelta(_Timedelta):
                 value = parse_timedelta_string(value)
             value = np.timedelta64(value)
         elif PyDelta_Check(value):
-            value = convert_to_timedelta64(value, 'ns')
+            # pytimedelta object -> microsecond resolution
+            new_value = delta_to_nanoseconds(
+                value, reso=NPY_DATETIMEUNIT.NPY_FR_us
+            )
+            return cls._from_value_and_reso(
+                new_value, reso=NPY_DATETIMEUNIT.NPY_FR_us
+            )
         elif is_timedelta64_object(value):
             # Retain the resolution if possible, otherwise cast to the nearest
             #  supported resolution.
@@ -1720,7 +1727,7 @@ class Timedelta(_Timedelta):
             if reso != NPY_DATETIMEUNIT.NPY_FR_GENERIC:
                 try:
                     new_value = convert_reso(
-                        get_timedelta64_value(value),
+                        new_value,
                         reso,
                         new_reso,
                         round_ok=True,
@@ -1730,7 +1737,10 @@ class Timedelta(_Timedelta):
             return cls._from_value_and_reso(new_value, reso=new_reso)
 
         elif is_tick_object(value):
-            value = np.timedelta64(value.nanos, 'ns')
+            new_reso = get_supported_reso(value._reso)
+            new_value = delta_to_nanoseconds(value, reso=new_reso)
+            return cls._from_value_and_reso(new_value, reso=new_reso)
+
         elif is_integer_object(value) or is_float_object(value):
             # unit=None is de-facto 'ns'
             unit = parse_timedelta_unit(unit)

pandas/core/arrays/masked.py

@@ -17,6 +17,10 @@
     lib,
     missing as libmissing,
 )
+from pandas._libs.tslibs import (
+    get_unit_from_dtype,
+    is_supported_unit,
+)
 from pandas._typing import (
     ArrayLike,
     AstypeArg,
@@ -750,12 +754,16 @@ def _maybe_mask_result(self, result, mask):
 
             return BooleanArray(result, mask, copy=False)
 
-        elif result.dtype == "timedelta64[ns]":
+        elif (
+            isinstance(result.dtype, np.dtype)
+            and result.dtype.kind == "m"
+            and is_supported_unit(get_unit_from_dtype(result.dtype))
+        ):
             # e.g. test_numeric_arr_mul_tdscalar_numexpr_path
             from pandas.core.arrays import TimedeltaArray
 
             if not isinstance(result, TimedeltaArray):
-                result = TimedeltaArray._simple_new(result)
+                result = TimedeltaArray._simple_new(result, dtype=result.dtype)
 
             result[mask] = result.dtype.type("NaT")
             return result

pandas/core/arrays/timedeltas.py

@@ -257,10 +257,10 @@ def _generate_range(cls, start, end, periods, freq, closed=None):
             )
 
         if start is not None:
-            start = Timedelta(start)
+            start = Timedelta(start)._as_unit("ns")
 
         if end is not None:
-            end = Timedelta(end)
+            end = Timedelta(end)._as_unit("ns")
 
         left_closed, right_closed = validate_endpoints(closed)
 

pandas/tests/arithmetic/test_numeric.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 from collections import abc
+from datetime import timedelta
 from decimal import Decimal
 import operator
 from typing import Any
@@ -27,6 +28,7 @@
     Int64Index,
     UInt64Index,
 )
+from pandas.core.arrays import TimedeltaArray
 from pandas.core.computation import expressions as expr
 from pandas.tests.arithmetic.common import (
     assert_invalid_addsub_type,
@@ -209,6 +211,11 @@ def test_numeric_arr_mul_tdscalar(self, scalar_td, numeric_idx, box_with_array):
             tda = expected._data
             dtype = scalar_td.dtype
             expected = type(tda)._simple_new(tda._ndarray.astype(dtype), dtype=dtype)
+        elif type(scalar_td) is timedelta and box not in [Index, Series]:
+            # TODO(2.0): once TDA.astype converts to m8, just do expected.astype
+            tda = expected._data
+            dtype = np.dtype("m8[us]")
+            expected = type(tda)._simple_new(tda._ndarray.astype(dtype), dtype=dtype)
 
         index = tm.box_expected(index, box)
         expected = tm.box_expected(expected, box)
@@ -240,6 +247,13 @@ def test_numeric_arr_mul_tdscalar_numexpr_path(
         obj = tm.box_expected(arr, box, transpose=False)
 
         expected = arr_i8.view("timedelta64[D]").astype("timedelta64[ns]")
+        if type(scalar_td) is timedelta and box is array:
+            # TODO(2.0): this shouldn't depend on 'box'
+            expected = expected.astype("timedelta64[us]")
+            # TODO(2.0): won't be necessary to construct TimedeltaArray
+            #  explicitly.
+            expected = TimedeltaArray._simple_new(expected, dtype=expected.dtype)
+
         expected = tm.box_expected(expected, box, transpose=False)
 
         result = obj * scalar_td
@@ -262,6 +276,11 @@ def test_numeric_arr_rdiv_tdscalar(self, three_days, numeric_idx, box_with_array
                 # i.e. resolution is lower -> use lowest supported resolution
                 dtype = np.dtype("m8[s]")
             expected = type(tda)._simple_new(tda._ndarray.astype(dtype), dtype=dtype)
+        elif type(three_days) is timedelta and box not in [Index, Series]:
+            # TODO(2.0): just use expected.astype
+            tda = expected._data
+            dtype = np.dtype("m8[us]")
+            expected = type(tda)._simple_new(tda._ndarray.astype(dtype), dtype=dtype)
 
         index = tm.box_expected(index, box)
         expected = tm.box_expected(expected, box)

pandas/tests/dtypes/cast/test_promote.py

@@ -480,6 +480,12 @@ def test_maybe_promote_any_with_timedelta64(
                 "Timedelta scalar"
             )
             request.node.add_marker(mark)
+        elif type(fill_value) is datetime.timedelta:
+            mark = pytest.mark.xfail(
+                reason="maybe_promote not yet updated to handle non-nano "
+                "Timedelta scalar"
+            )
+            request.node.add_marker(mark)
     else:
         expected_dtype = np.dtype(object)
         exp_val_for_scalar = fill_value

pandas/tests/io/json/test_pandas.py

@@ -1039,11 +1039,11 @@ def test_timedelta(self):
         tm.assert_frame_equal(frame, result)
 
     def test_mixed_timedelta_datetime(self):
-        frame = DataFrame({"a": [timedelta(23), Timestamp("20130101")]}, dtype=object)
+        td = timedelta(23)
+        ts = Timestamp("20130101")
+        frame = DataFrame({"a": [td, ts]}, dtype=object)
 
-        expected = DataFrame(
-            {"a": [pd.Timedelta(frame.a[0]).value, Timestamp(frame.a[1]).value]}
-        )
+        expected = DataFrame({"a": [pd.Timedelta(td)._as_unit("ns").value, ts.value]})
         result = read_json(frame.to_json(date_unit="ns"), dtype={"a": "int64"})
         tm.assert_frame_equal(result, expected, check_index_type=False)
 

pandas/tests/scalar/timedelta/test_constructors.py

@@ -61,6 +61,38 @@ def test_from_td64_retain_resolution():
     assert td3._reso == NpyDatetimeUnit.NPY_FR_us.value
 
 
+def test_from_pytimedelta_us_reso():
+    # pytimedelta has microsecond resolution, so Timedelta(pytd) inherits that
+    td = timedelta(days=4, minutes=3)
+    result = Timedelta(td)
+    assert result.to_pytimedelta() == td
+    assert result._reso == NpyDatetimeUnit.NPY_FR_us.value
+
+
+def test_from_tick_reso():
+    tick = offsets.Nano()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_ns.value
+
+    tick = offsets.Micro()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_us.value
+
+    tick = offsets.Milli()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_ms.value
+
+    tick = offsets.Second()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_s.value
+
+    # everything above Second gets cast to the closest supported reso: second
+    tick = offsets.Minute()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_s.value
+
+    tick = offsets.Hour()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_s.value
+
+    tick = offsets.Day()
+    assert Timedelta(tick)._reso == NpyDatetimeUnit.NPY_FR_s.value
+
+
 def test_construction():
     expected = np.timedelta64(10, "D").astype("m8[ns]").view("i8")
     assert Timedelta(10, unit="d").value == expected
@@ -248,9 +280,10 @@ def test_overflow_on_construction():
     with pytest.raises(OutOfBoundsTimedelta, match=msg):
         Timedelta(7 * 19999, unit="D")
 
-    msg = "Cannot cast 259987 days, 0:00:00 to unit=ns without overflow"
-    with pytest.raises(OutOfBoundsTimedelta, match=msg):
-        Timedelta(timedelta(days=13 * 19999))
+    # used to overflow before non-ns support
+    td = Timedelta(timedelta(days=13 * 19999))
+    assert td._reso == NpyDatetimeUnit.NPY_FR_us.value
+    assert td.days == 13 * 19999
 
 
 @pytest.mark.parametrize(

pandas/tests/scalar/timedelta/test_timedelta.py

@@ -183,13 +183,16 @@ def test_truediv_timedeltalike(self, td):
         assert (2.5 * td) / td == 2.5
 
         other = Timedelta(td.value)
-        msg = "Cannot cast 106752 days 00:00:00 to unit='ns' without overflow"
+        msg = "Cannot cast 106752 days 00:00:00 to unit='ns' without overflow."
         with pytest.raises(OutOfBoundsTimedelta, match=msg):
             td / other
 
-        with pytest.raises(OutOfBoundsTimedelta, match=msg):
-            # __rtruediv__
-            other.to_pytimedelta() / td
+        # Timedelta(other.to_pytimedelta()) has microsecond resolution,
+        #  so the division doesn't require casting all the way to nanos,
+        #  so succeeds
+        res = other.to_pytimedelta() / td
+        expected = other.to_pytimedelta() / td.to_pytimedelta()
+        assert res == expected
 
         # if there's no overflow, we cast to the higher reso
         left = Timedelta._from_value_and_reso(50, NpyDatetimeUnit.NPY_FR_us.value)
@@ -220,9 +223,11 @@ def test_floordiv_timedeltalike(self, td):
         with pytest.raises(OutOfBoundsTimedelta, match=msg):
             td // other
 
-        with pytest.raises(ValueError, match=msg):
-            # __rfloordiv__
-            other.to_pytimedelta() // td
+        # Timedelta(other.to_pytimedelta()) has microsecond resolution,
+        #  so the floordiv doesn't require casting all the way to nanos,
+        #  so succeeds
+        res = other.to_pytimedelta() // td
+        assert res == 0
 
         # if there's no overflow, we cast to the higher reso
         left = Timedelta._from_value_and_reso(50050, NpyDatetimeUnit.NPY_FR_us.value)