from __future__ import annotations import warnings from datetime import datetime, timedelta from itertools import product, starmap from typing import Literal import numpy as np import pandas as pd import pytest from pandas.errors import OutOfBoundsDatetime, OutOfBoundsTimedelta from xarray import ( DataArray, Dataset, Variable, conventions, date_range, decode_cf, ) from xarray.coders import CFDatetimeCoder, CFTimedeltaCoder from xarray.coding.times import ( _encode_datetime_with_cftime, _netcdf_to_numpy_timeunit, _numpy_to_netcdf_timeunit, _should_cftime_be_used, cftime_to_nptime, decode_cf_datetime, decode_cf_timedelta, encode_cf_datetime, encode_cf_timedelta, format_cftime_datetime, infer_datetime_units, infer_timedelta_units, ) from xarray.coding.variables import SerializationWarning from xarray.conventions import _update_bounds_attributes, cf_encoder from xarray.core.common import contains_cftime_datetimes from xarray.core.types import PDDatetimeUnitOptions from xarray.core.utils import is_duck_dask_array from xarray.testing import assert_equal, assert_identical from xarray.tests import ( _ALL_CALENDARS, _NON_STANDARD_CALENDARS, _STANDARD_CALENDAR_NAMES, _STANDARD_CALENDARS, DuckArrayWrapper, FirstElementAccessibleArray, _all_cftime_date_types, arm_xfail, assert_array_equal, assert_duckarray_allclose, assert_duckarray_equal, assert_no_warnings, has_cftime, requires_cftime, requires_dask, ) _CF_DATETIME_NUM_DATES_UNITS = [ (np.arange(10), "days since 2000-01-01", "s"), (np.arange(10).astype("float64"), "days since 2000-01-01", "s"), (np.arange(10).astype("float32"), "days since 2000-01-01", "s"), (np.arange(10).reshape(2, 5), "days since 2000-01-01", "s"), (12300 + np.arange(5), "hours since 1680-01-01 00:00:00", "s"), # here we add a couple minor formatting errors to test # the robustness of the parsing algorithm. (12300 + np.arange(5), "hour since 1680-01-01 00:00:00", "s"), (12300 + np.arange(5), "Hour since 1680-01-01 00:00:00", "s"), (12300 + np.arange(5), " Hour since 1680-01-01 00:00:00 ", "s"), (10, "days since 2000-01-01", "s"), ([10], "daYs since 2000-01-01", "s"), ([[10]], "days since 2000-01-01", "s"), ([10, 10], "days since 2000-01-01", "s"), (np.array(10), "days since 2000-01-01", "s"), (0, "days since 1000-01-01", "s"), ([0], "days since 1000-01-01", "s"), ([[0]], "days since 1000-01-01", "s"), (np.arange(2), "days since 1000-01-01", "s"), (np.arange(0, 100000, 20000), "days since 1900-01-01", "s"), (np.arange(0, 100000, 20000), "days since 1-01-01", "s"), (17093352.0, "hours since 1-1-1 00:00:0.0", "s"), ([0.5, 1.5], "hours since 1900-01-01T00:00:00", "s"), (0, "milliseconds since 2000-01-01T00:00:00", "s"), (0, "microseconds since 2000-01-01T00:00:00", "s"), (np.int32(788961600), "seconds since 1981-01-01", "s"), # GH2002 (12300 + np.arange(5), "hour since 1680-01-01 00:00:00.500000", "us"), (164375, "days since 1850-01-01 00:00:00", "s"), (164374.5, "days since 1850-01-01 00:00:00", "s"), ([164374.5, 168360.5], "days since 1850-01-01 00:00:00", "s"), ] _CF_DATETIME_TESTS = [ num_dates_units + (calendar,) for num_dates_units, calendar in product( _CF_DATETIME_NUM_DATES_UNITS, _STANDARD_CALENDAR_NAMES ) ] @requires_cftime @pytest.mark.filterwarnings("ignore:Ambiguous reference date string") @pytest.mark.filterwarnings("ignore:Times can't be serialized faithfully") @pytest.mark.parametrize( ["num_dates", "units", "minimum_resolution", "calendar"], _CF_DATETIME_TESTS ) def test_cf_datetime( num_dates, units: str, minimum_resolution: PDDatetimeUnitOptions, calendar: str, time_unit: PDDatetimeUnitOptions, ) -> None: import cftime expected = cftime.num2date( num_dates, units, calendar, only_use_cftime_datetimes=True ) with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime(num_dates, units, calendar, time_unit=time_unit) if actual.dtype.kind != "O": if np.timedelta64(1, time_unit) > np.timedelta64(1, minimum_resolution): expected_unit = minimum_resolution else: expected_unit = time_unit expected = cftime_to_nptime(expected, time_unit=expected_unit) assert_array_equal(actual, expected) encoded1, _, _ = encode_cf_datetime(actual, units, calendar) assert_array_equal(num_dates, encoded1) if hasattr(num_dates, "ndim") and num_dates.ndim == 1 and "1000" not in units: # verify that wrapping with a pandas.Index works # note that it *does not* currently work to put # non-datetime64 compatible dates into a pandas.Index encoded2, _, _ = encode_cf_datetime(pd.Index(actual), units, calendar) assert_array_equal(num_dates, encoded2) @requires_cftime def test_decode_cf_datetime_overflow(time_unit: PDDatetimeUnitOptions) -> None: # checks for # https://github.com/pydata/pandas/issues/14068 # https://github.com/pydata/xarray/issues/975 from cftime import DatetimeGregorian datetime = DatetimeGregorian units = "days since 2000-01-01 00:00:00" # date after 2262 and before 1678 days = (-117710, 95795) expected = (datetime(1677, 9, 20), datetime(2262, 4, 12)) for i, day in enumerate(days): with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") result = decode_cf_datetime( day, units, calendar="standard", time_unit=time_unit ) assert result == expected[i] # additional check to see if type/dtypes are correct if time_unit == "ns": assert isinstance(result.item(), datetime) else: assert result.dtype == np.dtype(f"=M8[{time_unit}]") def test_decode_cf_datetime_non_standard_units() -> None: expected = pd.date_range(periods=100, start="1970-01-01", freq="h") # netCDFs from madis.noaa.gov use this format for their time units # they cannot be parsed by cftime, but pd.Timestamp works units = "hours since 1-1-1970" actual = decode_cf_datetime(np.arange(100), units) assert_array_equal(actual, expected) @requires_cftime def test_decode_cf_datetime_non_iso_strings() -> None: # datetime strings that are _almost_ ISO compliant but not quite, # but which cftime.num2date can still parse correctly expected = pd.date_range(periods=100, start="2000-01-01", freq="h") cases = [ (np.arange(100), "hours since 2000-01-01 0"), (np.arange(100), "hours since 2000-1-1 0"), (np.arange(100), "hours since 2000-01-01 0:00"), ] for num_dates, units in cases: actual = decode_cf_datetime(num_dates, units) assert_array_equal(actual, expected) @requires_cftime @pytest.mark.parametrize("calendar", _STANDARD_CALENDARS) def test_decode_standard_calendar_inside_timestamp_range( calendar, time_unit: PDDatetimeUnitOptions ) -> None: import cftime units = "hours since 0001-01-01" times = pd.date_range( "2001-04-01-00", end="2001-04-30-23", unit=time_unit, freq="h" ) # to_pydatetime() will return microsecond time = cftime.date2num(times.to_pydatetime(), units, calendar=calendar) expected = times.values # for cftime we get "us" resolution # ns resolution is handled by cftime due to the reference date # being out of bounds, but the times themselves are # representable with nanosecond resolution. actual = decode_cf_datetime(time, units, calendar=calendar, time_unit=time_unit) assert actual.dtype == np.dtype(f"=M8[{time_unit}]") assert_array_equal(actual, expected) @requires_cftime @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) def test_decode_non_standard_calendar_inside_timestamp_range(calendar) -> None: import cftime units = "days since 0001-01-01" times = pd.date_range("2001-04-01-00", end="2001-04-30-23", freq="h") non_standard_time = cftime.date2num(times.to_pydatetime(), units, calendar=calendar) expected = cftime.num2date( non_standard_time, units, calendar=calendar, only_use_cftime_datetimes=True ) expected_dtype = np.dtype("O") actual = decode_cf_datetime(non_standard_time, units, calendar=calendar) assert actual.dtype == expected_dtype assert_array_equal(actual, expected) @requires_cftime @pytest.mark.parametrize("calendar", _ALL_CALENDARS) def test_decode_dates_outside_timestamp_range( calendar, time_unit: PDDatetimeUnitOptions ) -> None: import cftime units = "days since 0001-01-01" times = [datetime(1, 4, 1, h) for h in range(1, 5)] time = cftime.date2num(times, units, calendar=calendar) expected = cftime.num2date( time, units, calendar=calendar, only_use_cftime_datetimes=True ) if calendar == "proleptic_gregorian" and time_unit != "ns": expected = cftime_to_nptime(expected, time_unit=time_unit) expected_date_type = type(expected[0]) with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime(time, units, calendar=calendar, time_unit=time_unit) assert all(isinstance(value, expected_date_type) for value in actual) assert_array_equal(actual, expected) @requires_cftime @pytest.mark.parametrize("calendar", _STANDARD_CALENDARS) @pytest.mark.parametrize("num_time", [735368, [735368], [[735368]]]) def test_decode_standard_calendar_single_element_inside_timestamp_range( calendar, time_unit: PDDatetimeUnitOptions, num_time, ) -> None: units = "days since 0001-01-01" with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime( num_time, units, calendar=calendar, time_unit=time_unit ) assert actual.dtype == np.dtype(f"=M8[{time_unit}]") @requires_cftime @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) def test_decode_non_standard_calendar_single_element_inside_timestamp_range( calendar, ) -> None: units = "days since 0001-01-01" for num_time in [735368, [735368], [[735368]]]: with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime(num_time, units, calendar=calendar) assert actual.dtype == np.dtype("O") @requires_cftime @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) def test_decode_single_element_outside_timestamp_range(calendar) -> None: import cftime units = "days since 0001-01-01" for days in [1, 1470376]: for num_time in [days, [days], [[days]]]: with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime(num_time, units, calendar=calendar) expected = cftime.num2date( days, units, calendar, only_use_cftime_datetimes=True ) assert isinstance(actual.item(), type(expected)) @requires_cftime @pytest.mark.parametrize("calendar", _STANDARD_CALENDARS) def test_decode_standard_calendar_multidim_time_inside_timestamp_range( calendar, time_unit: PDDatetimeUnitOptions, ) -> None: import cftime units = "days since 0001-01-01" times1 = pd.date_range("2001-04-01", end="2001-04-05", freq="D") times2 = pd.date_range("2001-05-01", end="2001-05-05", freq="D") time1 = cftime.date2num(times1.to_pydatetime(), units, calendar=calendar) time2 = cftime.date2num(times2.to_pydatetime(), units, calendar=calendar) mdim_time = np.empty((len(time1), 2)) mdim_time[:, 0] = time1 mdim_time[:, 1] = time2 expected1 = times1.values expected2 = times2.values actual = decode_cf_datetime( mdim_time, units, calendar=calendar, time_unit=time_unit ) assert actual.dtype == np.dtype(f"=M8[{time_unit}]") assert_array_equal(actual[:, 0], expected1) assert_array_equal(actual[:, 1], expected2) @requires_cftime @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) def test_decode_nonstandard_calendar_multidim_time_inside_timestamp_range( calendar, ) -> None: import cftime units = "days since 0001-01-01" times1 = pd.date_range("2001-04-01", end="2001-04-05", freq="D") times2 = pd.date_range("2001-05-01", end="2001-05-05", freq="D") time1 = cftime.date2num(times1.to_pydatetime(), units, calendar=calendar) time2 = cftime.date2num(times2.to_pydatetime(), units, calendar=calendar) mdim_time = np.empty((len(time1), 2)) mdim_time[:, 0] = time1 mdim_time[:, 1] = time2 if cftime.__name__ == "cftime": expected1 = cftime.num2date( time1, units, calendar, only_use_cftime_datetimes=True ) expected2 = cftime.num2date( time2, units, calendar, only_use_cftime_datetimes=True ) else: expected1 = cftime.num2date(time1, units, calendar) expected2 = cftime.num2date(time2, units, calendar) expected_dtype = np.dtype("O") actual = decode_cf_datetime(mdim_time, units, calendar=calendar) assert actual.dtype == expected_dtype assert_array_equal(actual[:, 0], expected1) assert_array_equal(actual[:, 1], expected2) @requires_cftime @pytest.mark.parametrize("calendar", _ALL_CALENDARS) def test_decode_multidim_time_outside_timestamp_range( calendar, time_unit: PDDatetimeUnitOptions ) -> None: import cftime units = "days since 0001-01-01" times1 = [datetime(1, 4, day) for day in range(1, 6)] times2 = [datetime(1, 5, day) for day in range(1, 6)] time1 = cftime.date2num(times1, units, calendar=calendar) time2 = cftime.date2num(times2, units, calendar=calendar) mdim_time = np.empty((len(time1), 2)) mdim_time[:, 0] = time1 mdim_time[:, 1] = time2 expected1 = cftime.num2date(time1, units, calendar, only_use_cftime_datetimes=True) expected2 = cftime.num2date(time2, units, calendar, only_use_cftime_datetimes=True) if calendar == "proleptic_gregorian" and time_unit != "ns": expected1 = cftime_to_nptime(expected1, time_unit=time_unit) expected2 = cftime_to_nptime(expected2, time_unit=time_unit) with warnings.catch_warnings(): warnings.filterwarnings("ignore", "Unable to decode time axis") actual = decode_cf_datetime( mdim_time, units, calendar=calendar, time_unit=time_unit ) dtype: np.dtype dtype = np.dtype("O") if calendar == "proleptic_gregorian" and time_unit != "ns": dtype = np.dtype(f"=M8[{time_unit}]") assert actual.dtype == dtype assert_array_equal(actual[:, 0], expected1) assert_array_equal(actual[:, 1], expected2) @requires_cftime @pytest.mark.parametrize( ("calendar", "num_time"), [("360_day", 720058.0), ("all_leap", 732059.0), ("366_day", 732059.0)], ) def test_decode_non_standard_calendar_single_element(calendar, num_time) -> None: import cftime units = "days since 0001-01-01" actual = decode_cf_datetime(num_time, units, calendar=calendar) expected = np.asarray( cftime.num2date(num_time, units, calendar, only_use_cftime_datetimes=True) ) assert actual.dtype == np.dtype("O") assert expected == actual @requires_cftime def test_decode_360_day_calendar() -> None: import cftime calendar = "360_day" # ensure leap year doesn't matter for year in [2010, 2011, 2012, 2013, 2014]: units = f"days since {year}-01-01" num_times = np.arange(100) expected = cftime.num2date( num_times, units, calendar, only_use_cftime_datetimes=True ) with warnings.catch_warnings(record=True) as w: warnings.simplefilter("always") actual = decode_cf_datetime(num_times, units, calendar=calendar) assert len(w) == 0 assert actual.dtype == np.dtype("O") assert_array_equal(actual, expected) @requires_cftime def test_decode_abbreviation() -> None: """Test making sure we properly fall back to cftime on abbreviated units.""" import cftime val = np.array([1586628000000.0]) units = "msecs since 1970-01-01T00:00:00Z" actual = decode_cf_datetime(val, units) expected = cftime_to_nptime(cftime.num2date(val, units)) assert_array_equal(actual, expected) @arm_xfail @requires_cftime @pytest.mark.parametrize( ["num_dates", "units", "expected_list"], [ ([np.nan], "days since 2000-01-01", ["NaT"]), ([np.nan, 0], "days since 2000-01-01", ["NaT", "2000-01-01T00:00:00Z"]), ( [np.nan, 0, 1], "days since 2000-01-01", ["NaT", "2000-01-01T00:00:00Z", "2000-01-02T00:00:00Z"], ), ], ) def test_cf_datetime_nan(num_dates, units, expected_list) -> None: with warnings.catch_warnings(): warnings.filterwarnings("ignore", "All-NaN") actual = decode_cf_datetime(num_dates, units) # use pandas because numpy will deprecate timezone-aware conversions expected = pd.to_datetime(expected_list).to_numpy(dtype="datetime64[ns]") assert_array_equal(expected, actual) @requires_cftime def test_decoded_cf_datetime_array_2d(time_unit: PDDatetimeUnitOptions) -> None: # regression test for GH1229 variable = Variable( ("x", "y"), np.array([[0, 1], [2, 3]]), {"units": "days since 2000-01-01"} ) result = CFDatetimeCoder(time_unit=time_unit).decode(variable) assert result.dtype == f"datetime64[{time_unit}]" expected = pd.date_range("2000-01-01", periods=4).values.reshape(2, 2) assert_array_equal(np.asarray(result), expected) @pytest.mark.parametrize("decode_times", [True, False]) @pytest.mark.parametrize("mask_and_scale", [True, False]) def test_decode_datetime_mask_and_scale( decode_times: bool, mask_and_scale: bool ) -> None: attrs = { "units": "nanoseconds since 1970-01-01", "calendar": "proleptic_gregorian", "_FillValue": np.int16(-1), "add_offset": 100000.0, } encoded = Variable(["time"], np.array([0, -1, 1], "int16"), attrs=attrs) decoded = conventions.decode_cf_variable( "foo", encoded, mask_and_scale=mask_and_scale, decode_times=decode_times ) result = conventions.encode_cf_variable(decoded, name="foo") assert_identical(encoded, result) assert encoded.dtype == result.dtype FREQUENCIES_TO_ENCODING_UNITS = { "ns": "nanoseconds", "us": "microseconds", "ms": "milliseconds", "s": "seconds", "min": "minutes", "h": "hours", "D": "days", } @pytest.mark.parametrize(("freq", "units"), FREQUENCIES_TO_ENCODING_UNITS.items()) def test_infer_datetime_units(freq, units) -> None: dates = pd.date_range("2000", periods=2, freq=freq) expected = f"{units} since 2000-01-01 00:00:00" assert expected == infer_datetime_units(dates) @pytest.mark.parametrize( ["dates", "expected"], [ ( pd.to_datetime(["1900-01-01", "1900-01-02", "NaT"], unit="ns"), "days since 1900-01-01 00:00:00", ), ( pd.to_datetime(["NaT", "1900-01-01"], unit="ns"), "days since 1900-01-01 00:00:00", ), (pd.to_datetime(["NaT"], unit="ns"), "days since 1970-01-01 00:00:00"), ], ) def test_infer_datetime_units_with_NaT(dates, expected) -> None: assert expected == infer_datetime_units(dates) _CFTIME_DATETIME_UNITS_TESTS = [ ([(1900, 1, 1), (1900, 1, 1)], "days since 1900-01-01 00:00:00.000000"), ( [(1900, 1, 1), (1900, 1, 2), (1900, 1, 2, 0, 0, 1)], "seconds since 1900-01-01 00:00:00.000000", ), ( [(1900, 1, 1), (1900, 1, 8), (1900, 1, 16)], "days since 1900-01-01 00:00:00.000000", ), ] @requires_cftime @pytest.mark.parametrize( "calendar", _NON_STANDARD_CALENDARS + ["gregorian", "proleptic_gregorian"] ) @pytest.mark.parametrize(("date_args", "expected"), _CFTIME_DATETIME_UNITS_TESTS) def test_infer_cftime_datetime_units(calendar, date_args, expected) -> None: date_type = _all_cftime_date_types()[calendar] dates = list(starmap(date_type, date_args)) assert expected == infer_datetime_units(dates) @pytest.mark.filterwarnings("ignore:Timedeltas can't be serialized faithfully") @pytest.mark.parametrize( ["timedeltas", "units", "numbers"], [ ("1D", "days", np.int64(1)), (["1D", "2D", "3D"], "days", np.array([1, 2, 3], "int64")), ("1h", "hours", np.int64(1)), ("1ms", "milliseconds", np.int64(1)), ("1us", "microseconds", np.int64(1)), ("1ns", "nanoseconds", np.int64(1)), (["NaT", "0s", "1s"], None, [np.iinfo(np.int64).min, 0, 1]), (["30m", "60m"], "hours", [0.5, 1.0]), ("NaT", "days", np.iinfo(np.int64).min), (["NaT", "NaT"], "days", [np.iinfo(np.int64).min, np.iinfo(np.int64).min]), ], ) def test_cf_timedelta(timedeltas, units, numbers) -> None: if timedeltas == "NaT": timedeltas = np.timedelta64("NaT", "ns") else: timedeltas = pd.to_timedelta(timedeltas).to_numpy() numbers = np.array(numbers) expected = numbers actual, _ = encode_cf_timedelta(timedeltas, units) assert_array_equal(expected, actual) assert expected.dtype == actual.dtype if units is not None: expected = timedeltas actual = decode_cf_timedelta(numbers, units) assert_array_equal(expected, actual) assert expected.dtype == actual.dtype expected = np.timedelta64("NaT", "ns") actual = decode_cf_timedelta(np.array(np.nan), "days") assert_array_equal(expected, actual) assert expected.dtype == actual.dtype def test_cf_timedelta_2d() -> None: units = "days" numbers = np.atleast_2d([1, 2, 3]) timedeltas = np.atleast_2d(pd.to_timedelta(["1D", "2D", "3D"]).to_numpy()) expected = timedeltas actual = decode_cf_timedelta(numbers, units) assert_array_equal(expected, actual) assert expected.dtype == actual.dtype @pytest.mark.parametrize("encoding_unit", FREQUENCIES_TO_ENCODING_UNITS.values()) def test_decode_cf_timedelta_time_unit( time_unit: PDDatetimeUnitOptions, encoding_unit ) -> None: encoded = 1 encoding_unit_as_numpy = _netcdf_to_numpy_timeunit(encoding_unit) if np.timedelta64(1, time_unit) > np.timedelta64(1, encoding_unit_as_numpy): expected = np.timedelta64(encoded, encoding_unit_as_numpy) else: expected = np.timedelta64(encoded, encoding_unit_as_numpy).astype( f"timedelta64[{time_unit}]" ) result = decode_cf_timedelta(encoded, encoding_unit, time_unit) assert result == expected assert result.dtype == expected.dtype def test_decode_cf_timedelta_time_unit_out_of_bounds( time_unit: PDDatetimeUnitOptions, ) -> None: # Define a scale factor that will guarantee overflow with the given # time_unit. scale_factor = np.timedelta64(1, time_unit) // np.timedelta64(1, "ns") encoded = scale_factor * 300 * 365 with pytest.raises(OutOfBoundsTimedelta): decode_cf_timedelta(encoded, "days", time_unit) def test_cf_timedelta_roundtrip_large_value(time_unit: PDDatetimeUnitOptions) -> None: value = np.timedelta64(np.iinfo(np.int64).max, time_unit) encoded, units = encode_cf_timedelta(value) decoded = decode_cf_timedelta(encoded, units, time_unit=time_unit) assert value == decoded assert value.dtype == decoded.dtype @pytest.mark.parametrize( ["deltas", "expected"], [ (pd.to_timedelta(["1 day", "2 days"]), "days"), (pd.to_timedelta(["1h", "1 day 1 hour"]), "hours"), (pd.to_timedelta(["1m", "2m", np.nan]), "minutes"), (pd.to_timedelta(["1m3s", "1m4s"]), "seconds"), ], ) def test_infer_timedelta_units(deltas, expected) -> None: assert expected == infer_timedelta_units(deltas) @requires_cftime @pytest.mark.parametrize( ["date_args", "expected"], [ ((1, 2, 3, 4, 5, 6), "0001-02-03 04:05:06.000000"), ((10, 2, 3, 4, 5, 6), "0010-02-03 04:05:06.000000"), ((100, 2, 3, 4, 5, 6), "0100-02-03 04:05:06.000000"), ((1000, 2, 3, 4, 5, 6), "1000-02-03 04:05:06.000000"), ], ) def test_format_cftime_datetime(date_args, expected) -> None: date_types = _all_cftime_date_types() for date_type in date_types.values(): result = format_cftime_datetime(date_type(*date_args)) assert result == expected @pytest.mark.parametrize("calendar", _ALL_CALENDARS) def test_decode_cf(calendar, time_unit: PDDatetimeUnitOptions) -> None: days = [1.0, 2.0, 3.0] # TODO: GH5690 — do we want to allow this type for `coords`? da = DataArray(days, coords=[days], dims=["time"], name="test") ds = da.to_dataset() for v in ["test", "time"]: ds[v].attrs["units"] = "days since 2001-01-01" ds[v].attrs["calendar"] = calendar if not has_cftime and calendar not in _STANDARD_CALENDAR_NAMES: with pytest.raises(ValueError): ds = decode_cf(ds) else: ds = decode_cf(ds, decode_times=CFDatetimeCoder(time_unit=time_unit)) if calendar not in _STANDARD_CALENDAR_NAMES: assert ds.test.dtype == np.dtype("O") else: assert ds.test.dtype == np.dtype(f"=M8[{time_unit}]") def test_decode_cf_time_bounds(time_unit: PDDatetimeUnitOptions) -> None: da = DataArray( np.arange(6, dtype="int64").reshape((3, 2)), coords={"time": [1, 2, 3]}, dims=("time", "nbnd"), name="time_bnds", ) attrs = { "units": "days since 2001-01", "calendar": "standard", "bounds": "time_bnds", } ds = da.to_dataset() ds["time"].attrs.update(attrs) _update_bounds_attributes(ds.variables) assert ds.variables["time_bnds"].attrs == { "units": "days since 2001-01", "calendar": "standard", } dsc = decode_cf(ds, decode_times=CFDatetimeCoder(time_unit=time_unit)) assert dsc.time_bnds.dtype == np.dtype(f"=M8[{time_unit}]") dsc = decode_cf(ds, decode_times=False) assert dsc.time_bnds.dtype == np.dtype("int64") # Do not overwrite existing attrs ds = da.to_dataset() ds["time"].attrs.update(attrs) bnd_attr = {"units": "hours since 2001-01", "calendar": "noleap"} ds["time_bnds"].attrs.update(bnd_attr) _update_bounds_attributes(ds.variables) assert ds.variables["time_bnds"].attrs == bnd_attr # If bounds variable not available do not complain ds = da.to_dataset() ds["time"].attrs.update(attrs) ds["time"].attrs["bounds"] = "fake_var" _update_bounds_attributes(ds.variables) @requires_cftime def test_encode_time_bounds() -> None: time = pd.date_range("2000-01-16", periods=1) time_bounds = pd.date_range("2000-01-01", periods=2, freq="MS") ds = Dataset(dict(time=time, time_bounds=time_bounds)) ds.time.attrs = {"bounds": "time_bounds"} ds.time.encoding = {"calendar": "noleap", "units": "days since 2000-01-01"} expected = {} # expected['time'] = Variable(data=np.array([15]), dims=['time']) expected["time_bounds"] = Variable(data=np.array([0, 31]), dims=["time_bounds"]) encoded, _ = cf_encoder(ds.variables, ds.attrs) assert_equal(encoded["time_bounds"], expected["time_bounds"]) assert "calendar" not in encoded["time_bounds"].attrs assert "units" not in encoded["time_bounds"].attrs # if time_bounds attrs are same as time attrs, it doesn't matter ds.time_bounds.encoding = {"calendar": "noleap", "units": "days since 2000-01-01"} encoded, _ = cf_encoder(dict(ds.variables.items()), ds.attrs) assert_equal(encoded["time_bounds"], expected["time_bounds"]) assert "calendar" not in encoded["time_bounds"].attrs assert "units" not in encoded["time_bounds"].attrs # for CF-noncompliant case of time_bounds attrs being different from # time attrs; preserve them for faithful roundtrip ds.time_bounds.encoding = {"calendar": "noleap", "units": "days since 1849-01-01"} encoded, _ = cf_encoder(dict(ds.variables.items()), ds.attrs) with pytest.raises(AssertionError): assert_equal(encoded["time_bounds"], expected["time_bounds"]) assert "calendar" not in encoded["time_bounds"].attrs assert encoded["time_bounds"].attrs["units"] == ds.time_bounds.encoding["units"] ds.time.encoding = {} with pytest.warns(UserWarning): cf_encoder(ds.variables, ds.attrs) @pytest.fixture(params=_ALL_CALENDARS) def calendar(request): return request.param @pytest.fixture def times(calendar): import cftime return cftime.num2date( np.arange(4), units="hours since 2000-01-01", calendar=calendar, only_use_cftime_datetimes=True, ) @pytest.fixture def data(times): data = np.random.rand(2, 2, 4) lons = np.linspace(0, 11, 2) lats = np.linspace(0, 20, 2) return DataArray( data, coords=[lons, lats, times], dims=["lon", "lat", "time"], name="data" ) @pytest.fixture def times_3d(times): lons = np.linspace(0, 11, 2) lats = np.linspace(0, 20, 2) times_arr = np.random.choice(times, size=(2, 2, 4)) return DataArray( times_arr, coords=[lons, lats, times], dims=["lon", "lat", "time"], name="data" ) @requires_cftime def test_contains_cftime_datetimes_1d(data) -> None: assert contains_cftime_datetimes(data.time.variable) @requires_cftime @requires_dask def test_contains_cftime_datetimes_dask_1d(data) -> None: assert contains_cftime_datetimes(data.time.variable.chunk()) @requires_cftime def test_contains_cftime_datetimes_3d(times_3d) -> None: assert contains_cftime_datetimes(times_3d.variable) @requires_cftime @requires_dask def test_contains_cftime_datetimes_dask_3d(times_3d) -> None: assert contains_cftime_datetimes(times_3d.variable.chunk()) @pytest.mark.parametrize("non_cftime_data", [DataArray([]), DataArray([1, 2])]) def test_contains_cftime_datetimes_non_cftimes(non_cftime_data) -> None: assert not contains_cftime_datetimes(non_cftime_data.variable) @requires_dask @pytest.mark.parametrize("non_cftime_data", [DataArray([]), DataArray([1, 2])]) def test_contains_cftime_datetimes_non_cftimes_dask(non_cftime_data) -> None: assert not contains_cftime_datetimes(non_cftime_data.variable.chunk()) @requires_cftime @pytest.mark.parametrize("shape", [(24,), (8, 3), (2, 4, 3)]) def test_encode_cf_datetime_overflow(shape) -> None: # Test for fix to GH 2272 dates = pd.date_range("2100", periods=24).values.reshape(shape) units = "days since 1800-01-01" calendar = "standard" num, _, _ = encode_cf_datetime(dates, units, calendar) roundtrip = decode_cf_datetime(num, units, calendar) np.testing.assert_array_equal(dates, roundtrip) def test_encode_expected_failures() -> None: dates = pd.date_range("2000", periods=3) with pytest.raises(ValueError, match="invalid time units"): encode_cf_datetime(dates, units="days after 2000-01-01") with pytest.raises(ValueError, match="invalid reference date"): encode_cf_datetime(dates, units="days since NO_YEAR") def test_encode_cf_datetime_pandas_min() -> None: # GH 2623 dates = pd.date_range("2000", periods=3) num, units, calendar = encode_cf_datetime(dates) expected_num = np.array([0.0, 1.0, 2.0]) expected_units = "days since 2000-01-01 00:00:00" expected_calendar = "proleptic_gregorian" np.testing.assert_array_equal(num, expected_num) assert units == expected_units assert calendar == expected_calendar @requires_cftime def test_encode_cf_datetime_invalid_pandas_valid_cftime() -> None: num, units, calendar = encode_cf_datetime( pd.date_range("2000", periods=3), # Pandas fails to parse this unit, but cftime is quite happy with it "days since 1970-01-01 00:00:00 00", "standard", ) expected_num = [10957, 10958, 10959] expected_units = "days since 1970-01-01 00:00:00 00" expected_calendar = "standard" assert_array_equal(num, expected_num) assert units == expected_units assert calendar == expected_calendar @requires_cftime def test_time_units_with_timezone_roundtrip(calendar) -> None: # Regression test for GH 2649 expected_units = "days since 2000-01-01T00:00:00-05:00" expected_num_dates = np.array([1, 2, 3]) dates = decode_cf_datetime(expected_num_dates, expected_units, calendar) # Check that dates were decoded to UTC; here the hours should all # equal 5. result_hours = DataArray(dates).dt.hour expected_hours = DataArray([5, 5, 5]) assert_equal(result_hours, expected_hours) # Check that the encoded values are accurately roundtripped. result_num_dates, result_units, result_calendar = encode_cf_datetime( dates, expected_units, calendar ) if calendar in _STANDARD_CALENDARS: assert_duckarray_equal(result_num_dates, expected_num_dates) else: # cftime datetime arithmetic is not quite exact. assert_duckarray_allclose(result_num_dates, expected_num_dates) assert result_units == expected_units assert result_calendar == calendar @pytest.mark.parametrize("calendar", _STANDARD_CALENDARS) def test_use_cftime_default_standard_calendar_in_range(calendar) -> None: numerical_dates = [0, 1] units = "days since 2000-01-01" expected = pd.date_range("2000", periods=2) with assert_no_warnings(): result = decode_cf_datetime(numerical_dates, units, calendar) np.testing.assert_array_equal(result, expected) @requires_cftime @pytest.mark.parametrize("calendar", ["standard", "gregorian"]) @pytest.mark.parametrize("units_year", [1500, 1580]) def test_use_cftime_default_standard_calendar_out_of_range( calendar, units_year ) -> None: from cftime import num2date numerical_dates = [0, 1] units = f"days since {units_year}-01-01" expected = num2date( numerical_dates, units, calendar, only_use_cftime_datetimes=True ) with pytest.warns(SerializationWarning): result = decode_cf_datetime(numerical_dates, units, calendar) np.testing.assert_array_equal(result, expected) @requires_cftime @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) @pytest.mark.parametrize("units_year", [1500, 2000, 2500]) def test_use_cftime_default_non_standard_calendar( calendar, units_year, time_unit: PDDatetimeUnitOptions ) -> None: from cftime import num2date numerical_dates = [0, 1] units = f"days since {units_year}-01-01" expected = num2date( numerical_dates, units, calendar, only_use_cftime_datetimes=True ) if time_unit == "ns" and units_year == 2500: with pytest.warns(SerializationWarning, match="Unable to decode time axis"): result = decode_cf_datetime( numerical_dates, units, calendar, time_unit=time_unit ) else: with assert_no_warnings(): result = decode_cf_datetime( numerical_dates, units, calendar, time_unit=time_unit ) np.testing.assert_array_equal(result, expected) @requires_cftime @pytest.mark.parametrize("calendar", _ALL_CALENDARS) @pytest.mark.parametrize("units_year", [1500, 2000, 2500]) def test_use_cftime_true(calendar, units_year) -> None: from cftime import num2date numerical_dates = [0, 1] units = f"days since {units_year}-01-01" expected = num2date( numerical_dates, units, calendar, only_use_cftime_datetimes=True ) with assert_no_warnings(): result = decode_cf_datetime(numerical_dates, units, calendar, use_cftime=True) np.testing.assert_array_equal(result, expected) @pytest.mark.parametrize("calendar", _STANDARD_CALENDARS) def test_use_cftime_false_standard_calendar_in_range(calendar) -> None: numerical_dates = [0, 1] units = "days since 2000-01-01" expected = pd.date_range("2000", periods=2) with assert_no_warnings(): result = decode_cf_datetime(numerical_dates, units, calendar, use_cftime=False) np.testing.assert_array_equal(result, expected) @pytest.mark.parametrize("calendar", ["standard", "gregorian"]) @pytest.mark.parametrize("units_year", [1500, 1582]) def test_use_cftime_false_standard_calendar_out_of_range(calendar, units_year) -> None: numerical_dates = [0, 1] units = f"days since {units_year}-01-01" with pytest.raises(OutOfBoundsDatetime): decode_cf_datetime(numerical_dates, units, calendar, use_cftime=False) @pytest.mark.parametrize("calendar", _NON_STANDARD_CALENDARS) @pytest.mark.parametrize("units_year", [1500, 2000, 2500]) def test_use_cftime_false_non_standard_calendar(calendar, units_year) -> None: numerical_dates = [0, 1] units = f"days since {units_year}-01-01" with pytest.raises(OutOfBoundsDatetime): decode_cf_datetime(numerical_dates, units, calendar, use_cftime=False) @requires_cftime @pytest.mark.parametrize("calendar", _ALL_CALENDARS) def test_decode_ambiguous_time_warns(calendar) -> None: # GH 4422, 4506 from cftime import num2date # we don't decode non-standard calendards with # pandas so expect no warning will be emitted is_standard_calendar = calendar in _STANDARD_CALENDAR_NAMES dates = [1, 2, 3] units = "days since 1-1-1" expected = num2date(dates, units, calendar=calendar, only_use_cftime_datetimes=True) if is_standard_calendar: with pytest.warns(SerializationWarning) as record: result = decode_cf_datetime(dates, units, calendar=calendar) relevant_warnings = [ r for r in record.list if str(r.message).startswith("Ambiguous reference date string: 1-1-1") ] assert len(relevant_warnings) == 1 else: with assert_no_warnings(): result = decode_cf_datetime(dates, units, calendar=calendar) np.testing.assert_array_equal(result, expected) @pytest.mark.filterwarnings("ignore:Times can't be serialized faithfully") @pytest.mark.parametrize("encoding_units", FREQUENCIES_TO_ENCODING_UNITS.values()) @pytest.mark.parametrize("freq", FREQUENCIES_TO_ENCODING_UNITS.keys()) @pytest.mark.parametrize("use_cftime", [True, False]) def test_encode_cf_datetime_defaults_to_correct_dtype( encoding_units, freq, use_cftime ) -> None: if not has_cftime and use_cftime: pytest.skip("Test requires cftime") if (freq == "ns" or encoding_units == "nanoseconds") and use_cftime: pytest.skip("Nanosecond frequency is not valid for cftime dates.") times = date_range("2000", periods=3, freq=freq, use_cftime=use_cftime) units = f"{encoding_units} since 2000-01-01" encoded, _units, _ = encode_cf_datetime(times, units) numpy_timeunit = _netcdf_to_numpy_timeunit(encoding_units) encoding_units_as_timedelta = np.timedelta64(1, numpy_timeunit) if pd.to_timedelta(1, freq) >= encoding_units_as_timedelta: assert encoded.dtype == np.int64 else: assert encoded.dtype == np.float64 @pytest.mark.parametrize("freq", FREQUENCIES_TO_ENCODING_UNITS.keys()) def test_encode_decode_roundtrip_datetime64( freq, time_unit: PDDatetimeUnitOptions ) -> None: # See GH 4045. Prior to GH 4684 this test would fail for frequencies of # "s", "ms", "us", and "ns". initial_time = pd.date_range("1678-01-01", periods=1) times = initial_time.append(pd.date_range("1968", periods=2, freq=freq)) variable = Variable(["time"], times) encoded = conventions.encode_cf_variable(variable) decoded = conventions.decode_cf_variable( "time", encoded, decode_times=CFDatetimeCoder(time_unit=time_unit) ) assert_equal(variable, decoded) @requires_cftime @pytest.mark.parametrize("freq", ["us", "ms", "s", "min", "h", "D"]) def test_encode_decode_roundtrip_cftime(freq) -> None: initial_time = date_range("0001", periods=1, use_cftime=True) times = initial_time.append( date_range("0001", periods=2, freq=freq, use_cftime=True) + timedelta(days=291000 * 365) ) variable = Variable(["time"], times) encoded = conventions.encode_cf_variable(variable) decoder = CFDatetimeCoder(use_cftime=True) decoded = conventions.decode_cf_variable("time", encoded, decode_times=decoder) assert_equal(variable, decoded) @requires_cftime def test__encode_datetime_with_cftime() -> None: # See GH 4870. cftime versions > 1.4.0 required us to adapt the # way _encode_datetime_with_cftime was written. import cftime calendar = "gregorian" times = cftime.num2date([0, 1], "hours since 2000-01-01", calendar) encoding_units = "days since 2000-01-01" # Since netCDF files do not support storing float128 values, we ensure that # float64 values are used by setting longdouble=False in num2date. This try # except logic can be removed when xarray's minimum version of cftime is at # least 1.6.2. try: expected = cftime.date2num(times, encoding_units, calendar, longdouble=False) except TypeError: expected = cftime.date2num(times, encoding_units, calendar) result = _encode_datetime_with_cftime(times, encoding_units, calendar) np.testing.assert_equal(result, expected) @requires_cftime def test_round_trip_standard_calendar_cftime_datetimes_pre_reform() -> None: from cftime import DatetimeGregorian dates = np.array([DatetimeGregorian(1, 1, 1), DatetimeGregorian(2000, 1, 1)]) encoded = encode_cf_datetime(dates, "seconds since 2000-01-01", "standard") with pytest.warns(SerializationWarning, match="Unable to decode time axis"): decoded = decode_cf_datetime(*encoded) np.testing.assert_equal(decoded, dates) @pytest.mark.parametrize("calendar", ["standard", "gregorian"]) def test_encode_cf_datetime_gregorian_proleptic_gregorian_mismatch_error( calendar: str, time_unit: PDDatetimeUnitOptions, ) -> None: if time_unit == "ns": pytest.skip("datetime64[ns] values can only be defined post reform") dates = np.array(["0001-01-01", "2001-01-01"], dtype=f"datetime64[{time_unit}]") with pytest.raises(ValueError, match="proleptic_gregorian"): encode_cf_datetime(dates, "seconds since 2000-01-01", calendar) @pytest.mark.parametrize("calendar", ["gregorian", "Gregorian", "GREGORIAN"]) def test_decode_encode_roundtrip_with_non_lowercase_letters( calendar, time_unit: PDDatetimeUnitOptions ) -> None: # See GH 5093. times = [0, 1] units = "days since 2000-01-01" attrs = {"calendar": calendar, "units": units} variable = Variable(["time"], times, attrs) decoded = conventions.decode_cf_variable( "time", variable, decode_times=CFDatetimeCoder(time_unit=time_unit) ) encoded = conventions.encode_cf_variable(decoded) # Previously this would erroneously be an array of cftime.datetime # objects. We check here that it is decoded properly to np.datetime64. assert np.issubdtype(decoded.dtype, np.datetime64) # Use assert_identical to ensure that the calendar attribute maintained its # original form throughout the roundtripping process, uppercase letters and # all. assert_identical(variable, encoded) @requires_cftime def test_should_cftime_be_used_source_outside_range(): src = date_range( "1000-01-01", periods=100, freq="MS", calendar="noleap", use_cftime=True ) with pytest.raises( ValueError, match="Source time range is not valid for numpy datetimes." ): _should_cftime_be_used(src, "standard", False) @requires_cftime def test_should_cftime_be_used_target_not_npable(): src = date_range( "2000-01-01", periods=100, freq="MS", calendar="noleap", use_cftime=True ) with pytest.raises( ValueError, match="Calendar 'noleap' is only valid with cftime." ): _should_cftime_be_used(src, "noleap", False) @pytest.mark.parametrize( "dtype", [np.int8, np.int16, np.int32, np.int64, np.uint8, np.uint16, np.uint32, np.uint64], ) def test_decode_cf_datetime_varied_integer_dtypes(dtype): units = "seconds since 2018-08-22T03:23:03Z" num_dates = dtype(50) # Set use_cftime=False to ensure we cannot mask a failure by falling back # to cftime. result = decode_cf_datetime(num_dates, units, use_cftime=False) expected = np.asarray(np.datetime64("2018-08-22T03:23:53", "ns")) np.testing.assert_equal(result, expected) @requires_cftime def test_decode_cf_datetime_uint64_with_cftime(): units = "days since 1700-01-01" num_dates = np.uint64(182621) result = decode_cf_datetime(num_dates, units) expected = np.asarray(np.datetime64("2200-01-01", "ns")) np.testing.assert_equal(result, expected) def test_decode_cf_datetime_uint64_with_pandas_overflow_error(): units = "nanoseconds since 1970-01-01" calendar = "standard" num_dates = np.uint64(1_000_000 * 86_400 * 360 * 500_000) with pytest.raises(OutOfBoundsTimedelta): decode_cf_datetime(num_dates, units, calendar, use_cftime=False) @requires_cftime def test_decode_cf_datetime_uint64_with_cftime_overflow_error(): units = "microseconds since 1700-01-01" calendar = "360_day" num_dates = np.uint64(1_000_000 * 86_400 * 360 * 500_000) with pytest.raises(OverflowError): decode_cf_datetime(num_dates, units, calendar) @pytest.mark.parametrize("use_cftime", [True, False]) def test_decode_0size_datetime(use_cftime): # GH1329 if use_cftime and not has_cftime: pytest.skip() dtype = object if use_cftime else "=M8[ns]" expected = np.array([], dtype=dtype) actual = decode_cf_datetime( np.zeros(shape=0, dtype=np.int64), units="days since 1970-01-01 00:00:00", calendar="proleptic_gregorian", use_cftime=use_cftime, ) np.testing.assert_equal(expected, actual) def test_decode_float_datetime(): num_dates = np.array([1867128, 1867134, 1867140], dtype="float32") units = "hours since 1800-01-01" calendar = "standard" expected = np.array( ["2013-01-01T00:00:00", "2013-01-01T06:00:00", "2013-01-01T12:00:00"], dtype="datetime64[ns]", ) actual = decode_cf_datetime( num_dates, units=units, calendar=calendar, use_cftime=False ) np.testing.assert_equal(actual, expected) @pytest.mark.parametrize("time_unit", ["ms", "us", "ns"]) def test_decode_float_datetime_with_decimals( time_unit: PDDatetimeUnitOptions, ) -> None: # test resolution enhancement for floats values = np.array([0, 0.125, 0.25, 0.375, 0.75, 1.0], dtype="float32") expected = np.array( [ "2000-01-01T00:00:00.000", "2000-01-01T00:00:00.125", "2000-01-01T00:00:00.250", "2000-01-01T00:00:00.375", "2000-01-01T00:00:00.750", "2000-01-01T00:00:01.000", ], dtype=f"=M8[{time_unit}]", ) units = "seconds since 2000-01-01" calendar = "standard" actual = decode_cf_datetime(values, units, calendar, time_unit=time_unit) assert actual.dtype == expected.dtype np.testing.assert_equal(actual, expected) @pytest.mark.parametrize( "time_unit, num", [("s", 0.123), ("ms", 0.1234), ("us", 0.1234567)] ) def test_coding_float_datetime_warning( time_unit: PDDatetimeUnitOptions, num: float ) -> None: units = "seconds since 2000-01-01" calendar = "standard" values = np.array([num], dtype="float32") with pytest.warns( SerializationWarning, match=f"Can't decode floating point datetimes to {time_unit!r}", ): decode_cf_datetime(values, units, calendar, time_unit=time_unit) @requires_cftime def test_scalar_unit() -> None: # test that a scalar units (often NaN when using to_netcdf) does not raise an error variable = Variable(("x", "y"), np.array([[0, 1], [2, 3]]), {"units": np.nan}) result = CFDatetimeCoder().decode(variable) assert np.isnan(result.attrs["units"]) @requires_cftime def test_contains_cftime_lazy() -> None: import cftime from xarray.core.common import _contains_cftime_datetimes times = np.array( [cftime.DatetimeGregorian(1, 1, 2, 0), cftime.DatetimeGregorian(1, 1, 2, 0)], dtype=object, ) array = FirstElementAccessibleArray(times) assert _contains_cftime_datetimes(array) @pytest.mark.parametrize( "timestr, format, dtype, fill_value, use_encoding", [ ("1677-09-21T00:12:43.145224193", "ns", np.int64, 20, True), ("1970-09-21T00:12:44.145224808", "ns", np.float64, 1e30, True), ( "1677-09-21T00:12:43.145225216", "ns", np.float64, -9.223372036854776e18, True, ), ("1677-09-21T00:12:43.145224193", "ns", np.int64, None, False), ("1677-09-21T00:12:43.145225", "us", np.int64, None, False), ("1970-01-01T00:00:01.000001", "us", np.int64, None, False), ("1677-09-21T00:21:52.901038080", "ns", np.float32, 20.0, True), ], ) def test_roundtrip_datetime64_nanosecond_precision( timestr: str, format: Literal["ns", "us"], dtype: np.typing.DTypeLike, fill_value: int | float | None, use_encoding: bool, time_unit: PDDatetimeUnitOptions, ) -> None: # test for GH7817 time = np.datetime64(timestr, format) times = [np.datetime64("1970-01-01T00:00:00", format), np.datetime64("NaT"), time] if use_encoding: encoding = dict(dtype=dtype, _FillValue=fill_value) else: encoding = {} var = Variable(["time"], times, encoding=encoding) assert var.dtype == np.dtype(f"=M8[{format}]") encoded_var = conventions.encode_cf_variable(var) assert ( encoded_var.attrs["units"] == f"{_numpy_to_netcdf_timeunit(format)} since 1970-01-01 00:00:00" ) assert encoded_var.attrs["calendar"] == "proleptic_gregorian" assert encoded_var.data.dtype == dtype decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_times=CFDatetimeCoder(time_unit=time_unit) ) result_unit = ( format if np.timedelta64(1, format) <= np.timedelta64(1, time_unit) else time_unit ) assert decoded_var.dtype == np.dtype(f"=M8[{result_unit}]") assert ( decoded_var.encoding["units"] == f"{_numpy_to_netcdf_timeunit(format)} since 1970-01-01 00:00:00" ) assert decoded_var.encoding["dtype"] == dtype assert decoded_var.encoding["calendar"] == "proleptic_gregorian" assert_identical(var, decoded_var) def test_roundtrip_datetime64_nanosecond_precision_warning( time_unit: PDDatetimeUnitOptions, ) -> None: # test warning if times can't be serialized faithfully times = [ np.datetime64("1970-01-01T00:01:00", time_unit), np.datetime64("NaT", time_unit), np.datetime64("1970-01-02T00:01:00", time_unit), ] units = "days since 1970-01-10T01:01:00" needed_units = "hours" new_units = f"{needed_units} since 1970-01-10T01:01:00" encoding = dict(dtype=None, _FillValue=20, units=units) var = Variable(["time"], times, encoding=encoding) with pytest.warns(UserWarning, match=f"Resolution of {needed_units!r} needed."): encoded_var = conventions.encode_cf_variable(var) assert encoded_var.dtype == np.float64 assert encoded_var.attrs["units"] == units assert encoded_var.attrs["_FillValue"] == 20.0 decoded_var = conventions.decode_cf_variable("foo", encoded_var) assert_identical(var, decoded_var) encoding = dict(dtype="int64", _FillValue=20, units=units) var = Variable(["time"], times, encoding=encoding) with pytest.warns( UserWarning, match=f"Serializing with units {new_units!r} instead." ): encoded_var = conventions.encode_cf_variable(var) assert encoded_var.dtype == np.int64 assert encoded_var.attrs["units"] == new_units assert encoded_var.attrs["_FillValue"] == 20 decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_times=CFDatetimeCoder(time_unit=time_unit) ) assert_identical(var, decoded_var) encoding = dict(dtype="float64", _FillValue=20, units=units) var = Variable(["time"], times, encoding=encoding) with warnings.catch_warnings(): warnings.simplefilter("error") encoded_var = conventions.encode_cf_variable(var) assert encoded_var.dtype == np.float64 assert encoded_var.attrs["units"] == units assert encoded_var.attrs["_FillValue"] == 20.0 decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_times=CFDatetimeCoder(time_unit=time_unit) ) assert_identical(var, decoded_var) encoding = dict(dtype="int64", _FillValue=20, units=new_units) var = Variable(["time"], times, encoding=encoding) with warnings.catch_warnings(): warnings.simplefilter("error") encoded_var = conventions.encode_cf_variable(var) assert encoded_var.dtype == np.int64 assert encoded_var.attrs["units"] == new_units assert encoded_var.attrs["_FillValue"] == 20 decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_times=CFDatetimeCoder(time_unit=time_unit) ) assert_identical(var, decoded_var) @pytest.mark.parametrize( "dtype, fill_value", [(np.int64, 20), (np.int64, np.iinfo(np.int64).min), (np.float64, 1e30)], ) def test_roundtrip_timedelta64_nanosecond_precision( dtype: np.typing.DTypeLike, fill_value: int | float, time_unit: PDDatetimeUnitOptions, ) -> None: # test for GH7942 one_day = np.timedelta64(1, "ns") nat = np.timedelta64("nat", "ns") timedelta_values = (np.arange(5) * one_day).astype("timedelta64[ns]") timedelta_values[2] = nat timedelta_values[4] = nat encoding = dict(dtype=dtype, _FillValue=fill_value, units="nanoseconds") var = Variable(["time"], timedelta_values, encoding=encoding) encoded_var = conventions.encode_cf_variable(var) decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_times=CFDatetimeCoder(time_unit=time_unit), decode_timedelta=CFTimedeltaCoder(time_unit=time_unit), ) assert_identical(var, decoded_var) def test_roundtrip_timedelta64_nanosecond_precision_warning() -> None: # test warning if timedeltas can't be serialized faithfully one_day = np.timedelta64(1, "D") nat = np.timedelta64("nat", "ns") timedelta_values = (np.arange(5) * one_day).astype("timedelta64[ns]") timedelta_values[2] = nat timedelta_values[4] = np.timedelta64(12, "h").astype("timedelta64[ns]") units = "days" needed_units = "hours" wmsg = ( f"Timedeltas can't be serialized faithfully with requested units {units!r}. " f"Serializing with units {needed_units!r} instead." ) encoding = dict(dtype=np.int64, _FillValue=20, units=units) var = Variable(["time"], timedelta_values, encoding=encoding) with pytest.warns(UserWarning, match=wmsg): encoded_var = conventions.encode_cf_variable(var) assert encoded_var.dtype == np.int64 assert encoded_var.attrs["units"] == needed_units assert encoded_var.attrs["_FillValue"] == 20 decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_timedelta=CFTimedeltaCoder(time_unit="ns") ) assert_identical(var, decoded_var) assert decoded_var.encoding["dtype"] == np.int64 _TEST_ROUNDTRIP_FLOAT_TIMES_TESTS = { "GH-8271": ( 20.0, np.array( ["1970-01-01 00:00:00", "1970-01-01 06:00:00", "NaT"], dtype="datetime64[ns]", ), "days since 1960-01-01", np.array([3653, 3653.25, 20.0]), ), "GH-9488-datetime64[ns]": ( 1.0e20, np.array(["2010-01-01 12:00:00", "NaT"], dtype="datetime64[ns]"), "seconds since 2010-01-01", np.array([43200, 1.0e20]), ), "GH-9488-timedelta64[ns]": ( 1.0e20, np.array([1_000_000_000, "NaT"], dtype="timedelta64[ns]"), "seconds", np.array([1.0, 1.0e20]), ), } @pytest.mark.parametrize( ("fill_value", "times", "units", "encoded_values"), _TEST_ROUNDTRIP_FLOAT_TIMES_TESTS.values(), ids=_TEST_ROUNDTRIP_FLOAT_TIMES_TESTS.keys(), ) def test_roundtrip_float_times(fill_value, times, units, encoded_values) -> None: # Regression test for GitHub issues #8271 and #9488 var = Variable( ["time"], times, encoding=dict(dtype=np.float64, _FillValue=fill_value, units=units), ) encoded_var = conventions.encode_cf_variable(var) np.testing.assert_array_equal(encoded_var, encoded_values) assert encoded_var.attrs["units"] == units assert encoded_var.attrs["_FillValue"] == fill_value decoded_var = conventions.decode_cf_variable( "foo", encoded_var, decode_timedelta=CFTimedeltaCoder(time_unit="ns") ) assert_identical(var, decoded_var) assert decoded_var.encoding["units"] == units assert decoded_var.encoding["_FillValue"] == fill_value _ENCODE_DATETIME64_VIA_DASK_TESTS = { "pandas-encoding-with-prescribed-units-and-dtype": ( "D", "days since 1700-01-01", np.dtype("int32"), ), "mixed-cftime-pandas-encoding-with-prescribed-units-and-dtype": pytest.param( "250YS", "days since 1700-01-01", np.dtype("int32"), marks=requires_cftime ), "pandas-encoding-with-default-units-and-dtype": ("250YS", None, None), } @requires_dask @pytest.mark.parametrize( ("freq", "units", "dtype"), _ENCODE_DATETIME64_VIA_DASK_TESTS.values(), ids=_ENCODE_DATETIME64_VIA_DASK_TESTS.keys(), ) def test_encode_cf_datetime_datetime64_via_dask( freq, units, dtype, time_unit: PDDatetimeUnitOptions ) -> None: import dask.array times_pd = pd.date_range(start="1700", freq=freq, periods=3, unit=time_unit) times = dask.array.from_array(times_pd, chunks=1) encoded_times, encoding_units, encoding_calendar = encode_cf_datetime( times, units, None, dtype ) assert is_duck_dask_array(encoded_times) assert encoded_times.chunks == times.chunks if units is not None and dtype is not None: assert encoding_units == units assert encoded_times.dtype == dtype else: expected_netcdf_time_unit = _numpy_to_netcdf_timeunit(time_unit) assert encoding_units == f"{expected_netcdf_time_unit} since 1970-01-01" assert encoded_times.dtype == np.dtype("int64") assert encoding_calendar == "proleptic_gregorian" decoded_times = decode_cf_datetime( encoded_times, encoding_units, encoding_calendar, time_unit=time_unit ) np.testing.assert_equal(decoded_times, times) assert decoded_times.dtype == times.dtype @requires_dask @pytest.mark.parametrize( ("range_function", "start", "units", "dtype"), [ (pd.date_range, "2000", None, np.dtype("int32")), (pd.date_range, "2000", "days since 2000-01-01", None), (pd.timedelta_range, "0D", None, np.dtype("int32")), (pd.timedelta_range, "0D", "days", None), ], ) def test_encode_via_dask_cannot_infer_error( range_function, start, units, dtype ) -> None: values = range_function(start=start, freq="D", periods=3) encoding = dict(units=units, dtype=dtype) variable = Variable(["time"], values, encoding=encoding).chunk({"time": 1}) with pytest.raises(ValueError, match="When encoding chunked arrays"): conventions.encode_cf_variable(variable) @requires_cftime @requires_dask @pytest.mark.parametrize( ("units", "dtype"), [("days since 1700-01-01", np.dtype("int32")), (None, None)] ) def test_encode_cf_datetime_cftime_datetime_via_dask(units, dtype) -> None: import dask.array calendar = "standard" times_idx = date_range( start="1700", freq="D", periods=3, calendar=calendar, use_cftime=True ) times = dask.array.from_array(times_idx, chunks=1) encoded_times, encoding_units, encoding_calendar = encode_cf_datetime( times, units, None, dtype ) assert is_duck_dask_array(encoded_times) assert encoded_times.chunks == times.chunks if units is not None and dtype is not None: assert encoding_units == units assert encoded_times.dtype == dtype else: assert encoding_units == "microseconds since 1970-01-01" assert encoded_times.dtype == np.int64 assert encoding_calendar == calendar decoded_times = decode_cf_datetime( encoded_times, encoding_units, encoding_calendar, use_cftime=True ) np.testing.assert_equal(decoded_times, times) @pytest.mark.parametrize( "use_cftime", [False, pytest.param(True, marks=requires_cftime)] ) @pytest.mark.parametrize("use_dask", [False, pytest.param(True, marks=requires_dask)]) def test_encode_cf_datetime_units_change(use_cftime, use_dask) -> None: times = date_range(start="2000", freq="12h", periods=3, use_cftime=use_cftime) encoding = dict(units="days since 2000-01-01", dtype=np.dtype("int64")) variable = Variable(["time"], times, encoding=encoding) if use_dask: variable = variable.chunk({"time": 1}) with pytest.raises(ValueError, match="Times can't be serialized"): conventions.encode_cf_variable(variable).compute() else: with pytest.warns(UserWarning, match="Times can't be serialized"): encoded = conventions.encode_cf_variable(variable) if use_cftime: expected_units = "hours since 2000-01-01 00:00:00.000000" else: expected_units = "hours since 2000-01-01" assert encoded.attrs["units"] == expected_units decoded = conventions.decode_cf_variable( "name", encoded, decode_times=CFDatetimeCoder(use_cftime=use_cftime) ) assert_equal(variable, decoded) @pytest.mark.parametrize("use_dask", [False, pytest.param(True, marks=requires_dask)]) def test_encode_cf_datetime_precision_loss_regression_test(use_dask) -> None: # Regression test for # https://github.com/pydata/xarray/issues/9134#issuecomment-2191446463 times = date_range("2000", periods=5, freq="ns") encoding = dict(units="seconds since 1970-01-01", dtype=np.dtype("int64")) variable = Variable(["time"], times, encoding=encoding) if use_dask: variable = variable.chunk({"time": 1}) with pytest.raises(ValueError, match="Times can't be serialized"): conventions.encode_cf_variable(variable).compute() else: with pytest.warns(UserWarning, match="Times can't be serialized"): encoded = conventions.encode_cf_variable(variable) decoded = conventions.decode_cf_variable("name", encoded) assert_equal(variable, decoded) @requires_dask @pytest.mark.parametrize( ("units", "dtype"), [("days", np.dtype("int32")), (None, None)] ) def test_encode_cf_timedelta_via_dask( units: str | None, dtype: np.dtype | None, time_unit: PDDatetimeUnitOptions ) -> None: import dask.array times_pd = pd.timedelta_range(start="0D", freq="D", periods=3, unit=time_unit) # type: ignore[call-arg] times = dask.array.from_array(times_pd, chunks=1) encoded_times, encoding_units = encode_cf_timedelta(times, units, dtype) assert is_duck_dask_array(encoded_times) assert encoded_times.chunks == times.chunks if units is not None and dtype is not None: assert encoding_units == units assert encoded_times.dtype == dtype else: assert encoding_units == _numpy_to_netcdf_timeunit(time_unit) assert encoded_times.dtype == np.dtype("int64") decoded_times = decode_cf_timedelta( encoded_times, encoding_units, time_unit=time_unit ) np.testing.assert_equal(decoded_times, times) assert decoded_times.dtype == times.dtype @pytest.mark.parametrize("use_dask", [False, pytest.param(True, marks=requires_dask)]) def test_encode_cf_timedelta_units_change(use_dask) -> None: timedeltas = pd.timedelta_range(start="0h", freq="12h", periods=3) encoding = dict(units="days", dtype=np.dtype("int64")) variable = Variable(["time"], timedeltas, encoding=encoding) if use_dask: variable = variable.chunk({"time": 1}) with pytest.raises(ValueError, match="Timedeltas can't be serialized"): conventions.encode_cf_variable(variable).compute() else: # In this case we automatically modify the encoding units to continue # encoding with integer values. with pytest.warns(UserWarning, match="Timedeltas can't be serialized"): encoded = conventions.encode_cf_variable(variable) assert encoded.attrs["units"] == "hours" decoded = conventions.decode_cf_variable( "name", encoded, decode_timedelta=CFTimedeltaCoder(time_unit="ns") ) assert_equal(variable, decoded) @pytest.mark.parametrize("use_dask", [False, pytest.param(True, marks=requires_dask)]) def test_encode_cf_timedelta_small_dtype_missing_value(use_dask) -> None: # Regression test for GitHub issue #9134 timedeltas = np.array([1, 2, "NaT", 4], dtype="timedelta64[D]").astype( "timedelta64[ns]" ) encoding = dict(units="days", dtype=np.dtype("int16"), _FillValue=np.int16(-1)) variable = Variable(["time"], timedeltas, encoding=encoding) if use_dask: variable = variable.chunk({"time": 1}) encoded = conventions.encode_cf_variable(variable) decoded = conventions.decode_cf_variable("name", encoded, decode_timedelta=True) assert_equal(variable, decoded) _DECODE_TIMEDELTA_VIA_UNITS_TESTS = { "default": (True, None, np.dtype("timedelta64[ns]"), True), "decode_timedelta=True": (True, True, np.dtype("timedelta64[ns]"), False), "decode_timedelta=False": (True, False, np.dtype("int64"), False), "inherit-time_unit-from-decode_times": ( CFDatetimeCoder(time_unit="s"), None, np.dtype("timedelta64[s]"), True, ), "set-time_unit-via-CFTimedeltaCoder-decode_times=True": ( True, CFTimedeltaCoder(time_unit="s"), np.dtype("timedelta64[s]"), False, ), "set-time_unit-via-CFTimedeltaCoder-decode_times=False": ( False, CFTimedeltaCoder(time_unit="s"), np.dtype("timedelta64[s]"), False, ), "override-time_unit-from-decode_times": ( CFDatetimeCoder(time_unit="ns"), CFTimedeltaCoder(time_unit="s"), np.dtype("timedelta64[s]"), False, ), } @pytest.mark.parametrize( ("decode_times", "decode_timedelta", "expected_dtype", "warns"), list(_DECODE_TIMEDELTA_VIA_UNITS_TESTS.values()), ids=list(_DECODE_TIMEDELTA_VIA_UNITS_TESTS.keys()), ) def test_decode_timedelta_via_units( decode_times, decode_timedelta, expected_dtype, warns ) -> None: timedeltas = pd.timedelta_range(0, freq="D", periods=3) attrs = {"units": "days"} var = Variable(["time"], timedeltas, encoding=attrs) encoded = Variable(["time"], np.array([0, 1, 2]), attrs=attrs) if warns: with pytest.warns( FutureWarning, match="xarray will not decode the variable 'foo' into a timedelta64 dtype", ): decoded = conventions.decode_cf_variable( "foo", encoded, decode_times=decode_times, decode_timedelta=decode_timedelta, ) else: decoded = conventions.decode_cf_variable( "foo", encoded, decode_times=decode_times, decode_timedelta=decode_timedelta ) if decode_timedelta is False: assert_equal(encoded, decoded) else: assert_equal(var, decoded) assert decoded.dtype == expected_dtype _DECODE_TIMEDELTA_VIA_DTYPE_TESTS = { "default": (True, None, "ns", np.dtype("timedelta64[ns]")), "decode_timedelta=False": (True, False, "ns", np.dtype("int64")), "decode_timedelta=True": (True, True, "ns", np.dtype("timedelta64[ns]")), "use-original-units": (True, True, "s", np.dtype("timedelta64[s]")), "inherit-time_unit-from-decode_times": ( CFDatetimeCoder(time_unit="s"), None, "ns", np.dtype("timedelta64[s]"), ), "set-time_unit-via-CFTimedeltaCoder-decode_times=True": ( True, CFTimedeltaCoder(time_unit="s"), "ns", np.dtype("timedelta64[s]"), ), "set-time_unit-via-CFTimedeltaCoder-decode_times=False": ( False, CFTimedeltaCoder(time_unit="s"), "ns", np.dtype("timedelta64[s]"), ), "override-time_unit-from-decode_times": ( CFDatetimeCoder(time_unit="ns"), CFTimedeltaCoder(time_unit="s"), "ns", np.dtype("timedelta64[s]"), ), "decode-different-units": ( True, CFTimedeltaCoder(time_unit="us"), "s", np.dtype("timedelta64[us]"), ), } @pytest.mark.parametrize( ("decode_times", "decode_timedelta", "original_unit", "expected_dtype"), list(_DECODE_TIMEDELTA_VIA_DTYPE_TESTS.values()), ids=list(_DECODE_TIMEDELTA_VIA_DTYPE_TESTS.keys()), ) def test_decode_timedelta_via_dtype( decode_times, decode_timedelta, original_unit, expected_dtype ) -> None: timedeltas = pd.timedelta_range(0, freq="D", periods=3, unit=original_unit) # type: ignore[call-arg] encoding = {"units": "days"} var = Variable(["time"], timedeltas, encoding=encoding) encoded = conventions.encode_cf_variable(var) assert encoded.attrs["dtype"] == f"timedelta64[{original_unit}]" assert encoded.attrs["units"] == encoding["units"] decoded = conventions.decode_cf_variable( "foo", encoded, decode_times=decode_times, decode_timedelta=decode_timedelta ) if decode_timedelta is False: assert_equal(encoded, decoded) else: assert_equal(var, decoded) assert decoded.dtype == expected_dtype def test_lazy_decode_timedelta_unexpected_dtype() -> None: attrs = {"units": "seconds"} encoded = Variable(["time"], [0, 0.5, 1], attrs=attrs) decoded = conventions.decode_cf_variable( "foo", encoded, decode_timedelta=CFTimedeltaCoder(time_unit="s") ) expected_dtype_upon_lazy_decoding = np.dtype("timedelta64[s]") assert decoded.dtype == expected_dtype_upon_lazy_decoding expected_dtype_upon_loading = np.dtype("timedelta64[ms]") with pytest.warns(SerializationWarning, match="Can't decode floating"): assert decoded.load().dtype == expected_dtype_upon_loading def test_lazy_decode_timedelta_error() -> None: attrs = {"units": "seconds"} encoded = Variable(["time"], [0, np.iinfo(np.int64).max, 1], attrs=attrs) decoded = conventions.decode_cf_variable( "foo", encoded, decode_timedelta=CFTimedeltaCoder(time_unit="ms") ) with pytest.raises(OutOfBoundsTimedelta, match="overflow"): decoded.load() @pytest.mark.parametrize( "calendar", [ "standard", pytest.param( "360_day", marks=pytest.mark.skipif(not has_cftime, reason="no cftime") ), ], ) def test_duck_array_decode_times(calendar) -> None: from xarray.core.indexing import LazilyIndexedArray days = LazilyIndexedArray(DuckArrayWrapper(np.array([1.0, 2.0, 3.0]))) var = Variable( ["time"], days, {"units": "days since 2001-01-01", "calendar": calendar} ) decoded = conventions.decode_cf_variable( "foo", var, decode_times=CFDatetimeCoder(use_cftime=None) ) if calendar not in _STANDARD_CALENDAR_NAMES: assert decoded.dtype == np.dtype("O") else: assert decoded.dtype == np.dtype("=M8[ns]") @pytest.mark.parametrize("decode_timedelta", [True, False]) @pytest.mark.parametrize("mask_and_scale", [True, False]) def test_decode_timedelta_mask_and_scale( decode_timedelta: bool, mask_and_scale: bool ) -> None: attrs = { "dtype": "timedelta64[ns]", "units": "nanoseconds", "_FillValue": np.int16(-1), "add_offset": 100000.0, } encoded = Variable(["time"], np.array([0, -1, 1], "int16"), attrs=attrs) decoded = conventions.decode_cf_variable( "foo", encoded, mask_and_scale=mask_and_scale, decode_timedelta=decode_timedelta ) result = conventions.encode_cf_variable(decoded, name="foo") assert_identical(encoded, result) assert encoded.dtype == result.dtype def test_decode_floating_point_timedelta_no_serialization_warning() -> None: attrs = {"units": "seconds"} encoded = Variable(["time"], [0, 0.1, 0.2], attrs=attrs) decoded = conventions.decode_cf_variable("foo", encoded, decode_timedelta=True) with assert_no_warnings(): decoded.load() def test_timedelta64_coding_via_dtype(time_unit: PDDatetimeUnitOptions) -> None: timedeltas = np.array([0, 1, "NaT"], dtype=f"timedelta64[{time_unit}]") variable = Variable(["time"], timedeltas) expected_units = _numpy_to_netcdf_timeunit(time_unit) encoded = conventions.encode_cf_variable(variable) assert encoded.attrs["dtype"] == f"timedelta64[{time_unit}]" assert encoded.attrs["units"] == expected_units decoded = conventions.decode_cf_variable("timedeltas", encoded) assert decoded.encoding["dtype"] == np.dtype("int64") assert decoded.encoding["units"] == expected_units assert_identical(decoded, variable) assert decoded.dtype == variable.dtype reencoded = conventions.encode_cf_variable(decoded) assert_identical(reencoded, encoded) assert reencoded.dtype == encoded.dtype def test_timedelta_coding_via_dtype_non_pandas_coarse_resolution_warning() -> None: attrs = {"dtype": "timedelta64[D]", "units": "days"} encoded = Variable(["time"], [0, 1, 2], attrs=attrs) with pytest.warns(UserWarning, match="xarray only supports"): decoded = conventions.decode_cf_variable("timedeltas", encoded) expected_array = np.array([0, 1, 2], dtype="timedelta64[D]") expected_array = expected_array.astype("timedelta64[s]") expected = Variable(["time"], expected_array) assert_identical(decoded, expected) assert decoded.dtype == np.dtype("timedelta64[s]") @pytest.mark.xfail(reason="xarray does not recognize picoseconds as time-like") def test_timedelta_coding_via_dtype_non_pandas_fine_resolution_warning() -> None: attrs = {"dtype": "timedelta64[ps]", "units": "picoseconds"} encoded = Variable(["time"], [0, 1000, 2000], attrs=attrs) with pytest.warns(UserWarning, match="xarray only supports"): decoded = conventions.decode_cf_variable("timedeltas", encoded) expected_array = np.array([0, 1000, 2000], dtype="timedelta64[ps]") expected_array = expected_array.astype("timedelta64[ns]") expected = Variable(["time"], expected_array) assert_identical(decoded, expected) assert decoded.dtype == np.dtype("timedelta64[ns]") def test_timedelta_decode_via_dtype_invalid_encoding() -> None: attrs = {"dtype": "timedelta64[s]", "units": "seconds"} encoding = {"units": "foo"} encoded = Variable(["time"], [0, 1, 2], attrs=attrs, encoding=encoding) with pytest.raises(ValueError, match="failed to prevent"): conventions.decode_cf_variable("timedeltas", encoded) @pytest.mark.parametrize("attribute", ["dtype", "units"]) def test_timedelta_encode_via_dtype_invalid_attribute(attribute) -> None: timedeltas = pd.timedelta_range(0, freq="D", periods=3) attrs = {attribute: "foo"} variable = Variable(["time"], timedeltas, attrs=attrs) with pytest.raises(ValueError, match="failed to prevent"): conventions.encode_cf_variable(variable) @pytest.mark.parametrize( ("decode_via_units", "decode_via_dtype", "attrs", "expect_timedelta64"), [ (True, True, {"units": "seconds"}, True), (True, False, {"units": "seconds"}, True), (False, True, {"units": "seconds"}, False), (False, False, {"units": "seconds"}, False), (True, True, {"dtype": "timedelta64[s]", "units": "seconds"}, True), (True, False, {"dtype": "timedelta64[s]", "units": "seconds"}, True), (False, True, {"dtype": "timedelta64[s]", "units": "seconds"}, True), (False, False, {"dtype": "timedelta64[s]", "units": "seconds"}, False), ], ids=lambda x: f"{x!r}", ) def test_timedelta_decoding_options( decode_via_units, decode_via_dtype, attrs, expect_timedelta64 ) -> None: array = np.array([0, 1, 2], dtype=np.dtype("int64")) encoded = Variable(["time"], array, attrs=attrs) # Confirm we decode to the expected dtype. decode_timedelta = CFTimedeltaCoder( time_unit="s", decode_via_units=decode_via_units, decode_via_dtype=decode_via_dtype, ) decoded = conventions.decode_cf_variable( "foo", encoded, decode_timedelta=decode_timedelta ) if expect_timedelta64: assert decoded.dtype == np.dtype("timedelta64[s]") else: assert decoded.dtype == np.dtype("int64") # Confirm we exactly roundtrip. reencoded = conventions.encode_cf_variable(decoded) expected = encoded.copy() if "dtype" not in attrs and decode_via_units: expected.attrs["dtype"] = "timedelta64[s]" assert_identical(reencoded, expected) def test_timedelta_encoding_explicit_non_timedelta64_dtype() -> None: encoding = {"dtype": np.dtype("int32")} timedeltas = pd.timedelta_range(0, freq="D", periods=3) variable = Variable(["time"], timedeltas, encoding=encoding) encoded = conventions.encode_cf_variable(variable) assert encoded.attrs["units"] == "days" assert encoded.attrs["dtype"] == "timedelta64[ns]" assert encoded.dtype == np.dtype("int32") decoded = conventions.decode_cf_variable("foo", encoded) assert_identical(decoded, variable) reencoded = conventions.encode_cf_variable(decoded) assert_identical(reencoded, encoded) assert encoded.attrs["units"] == "days" assert encoded.attrs["dtype"] == "timedelta64[ns]" assert encoded.dtype == np.dtype("int32") @pytest.mark.parametrize("mask_attribute", ["_FillValue", "missing_value"]) def test_timedelta64_coding_via_dtype_with_mask( time_unit: PDDatetimeUnitOptions, mask_attribute: str ) -> None: timedeltas = np.array([0, 1, "NaT"], dtype=f"timedelta64[{time_unit}]") mask = 10 variable = Variable(["time"], timedeltas, encoding={mask_attribute: mask}) expected_dtype = f"timedelta64[{time_unit}]" expected_units = _numpy_to_netcdf_timeunit(time_unit) encoded = conventions.encode_cf_variable(variable) assert encoded.attrs["dtype"] == expected_dtype assert encoded.attrs["units"] == expected_units assert encoded.attrs[mask_attribute] == mask assert encoded[-1] == mask decoded = conventions.decode_cf_variable("timedeltas", encoded) assert decoded.encoding["dtype"] == np.dtype("int64") assert decoded.encoding["units"] == expected_units assert decoded.encoding[mask_attribute] == mask assert np.isnat(decoded[-1]) assert_identical(decoded, variable) assert decoded.dtype == variable.dtype reencoded = conventions.encode_cf_variable(decoded) assert_identical(reencoded, encoded) assert reencoded.dtype == encoded.dtype def test_roundtrip_0size_timedelta(time_unit: PDDatetimeUnitOptions) -> None: # regression test for GitHub issue #10310 encoding = {"units": "days", "dtype": np.dtype("int64")} data = np.array([], dtype=f"=m8[{time_unit}]") decoded = Variable(["time"], data, encoding=encoding) encoded = conventions.encode_cf_variable(decoded, name="foo") assert encoded.dtype == encoding["dtype"] assert encoded.attrs["units"] == encoding["units"] decoded = conventions.decode_cf_variable("foo", encoded, decode_timedelta=True) assert decoded.dtype == np.dtype(f"=m8[{time_unit}]") with assert_no_warnings(): decoded.load() assert decoded.dtype == np.dtype("=m8[s]") assert decoded.encoding == encoding