Function _numbers_to_timedelta

xarray/coding/times.py:616–675

Transform numbers to np.timedelta64.

(
    flat_num: np.ndarray,
    time_unit: NPDatetimeUnitOptions,
    ref_unit: PDDatetimeUnitOptions,
    datatype: str,
    target_unit: PDDatetimeUnitOptions | None = None,
)

def _numbers_to_timedelta(
    flat_num: np.ndarray,
    time_unit: NPDatetimeUnitOptions,
    ref_unit: PDDatetimeUnitOptions,
    datatype: str,
    target_unit: PDDatetimeUnitOptions | None = None,
) -> np.ndarray:
    """Transform numbers to np.timedelta64."""
    # keep NaT/nan mask
    if flat_num.dtype.kind == "f":
        nan = np.asarray(np.isnan(flat_num))
    elif flat_num.dtype.kind == "i":
        nan = np.asarray(flat_num == np.iinfo(np.int64).min)
    elif flat_num.dtype.kind == "u":
        nan = np.broadcast_to(np.asarray(False), flat_num.shape)

    # in case we need to change the unit, we fix the numbers here
    # this should be safe, as errors would have been raised above
    ns_time_unit = _NS_PER_TIME_DELTA[time_unit]
    ns_ref_date_unit = _NS_PER_TIME_DELTA[ref_unit]
    if ns_time_unit > ns_ref_date_unit:
        flat_num = np.asarray(flat_num * np.int64(ns_time_unit / ns_ref_date_unit))
        time_unit = ref_unit

    # estimate fitting resolution for floating point values
    # this iterates until all floats are fractionless or time_unit == "ns"
    if flat_num.dtype.kind == "f" and time_unit != "ns":
        flat_num, new_time_unit = _check_higher_resolution(
            flat_num, cast(PDDatetimeUnitOptions, time_unit)
        )
        if time_unit != new_time_unit:
            if target_unit is None or np.timedelta64(1, target_unit) > np.timedelta64(
                1, new_time_unit
            ):
                if datatype == "datetimes":
                    kwarg = "decode_times"
                    coder = "CFDatetimeCoder"
                else:
                    kwarg = "decode_timedelta"
                    coder = "CFTimedeltaCoder"
                formatted_kwarg = f"{kwarg}={coder}(time_unit={new_time_unit!r})"
                message = (
                    f"Can't decode floating point {datatype} to {time_unit!r} "
                    f"without precision loss; decoding to {new_time_unit!r} "
                    f"instead. To silence this warning pass {formatted_kwarg} "
                    f"to your opening function."
                )
                emit_user_level_warning(message, SerializationWarning)
            time_unit = new_time_unit

    # Cast input ordinals to integers and properly handle NaN/NaT
    # to prevent casting NaN to int
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", RuntimeWarning)
        flat_num = flat_num.astype(np.int64)
    if nan.any():
        flat_num[nan] = np.iinfo(np.int64).min