2222748844

Build Type

github

Committed by GitHub

Commit Message

Merge 8adc45c7e into d076d8475

Pull Request Pull Request #33: Support CORDEX and CMIP5/6

Run Details

21 of 103 new or added lines in 9 files covered. (20.39%)

4 existing lines in 3 files now uncovered.

652 of 3206 relevant lines covered (20.34%)

0.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

64.44

/miranda/units.py

import re

import pint
import xarray as xr
from pint import Unit
from xclim.core.calendar import parse_offset

KiB = int(pow(2, 10))
MiB = int(pow(2, 20))
GiB = int(pow(2, 30))

u = pint.UnitRegistry(autoconvert_offset_to_baseunit=True)

# General purpose units
null = pint.Context("none")
u.add_context(null)
u.define("fraction = []")
u.define("percent = 1e-2 fraction = pct")
u.define("degC = kelvin; offset: 273.15 = celsius = C")
u.define("d = day")
u.define("h = hour")  # Not the Planck constant...
u.define("[speed] = [length] / [time]")

u.define("[precipitation] = [mass] / [length] ** 2 / [time]")
u.define("mmday = 1 kg / meter ** 2 / day")
u.define("[discharge] = [length] ** 3 / [time]")
u.define("cms = meter ** 3 / second")

# Allows transformation from mm day to kg / m2 / s
hq = pint.Context("hq")

hq.add_transformation(
    "[mass] / [length]**2",
    "[length]",
    lambda ureg, x: x / (1000 * ureg.kg / ureg.m**3),
)
hq.add_transformation(
    "[mass] / [length]**2 / [time]",
    "[length] / [time]",
    lambda ureg, x: x / (1000 * ureg.kg / ureg.m**3),
)
hq.add_transformation(
    "[length] / [time]",
    "[mass] / [length]**2 / [time]",
    lambda ureg, x: x * (1000 * ureg.kg / ureg.m**3),
)
u.add_context(hq)
u.enable_contexts(hq)


def units2pint(value: str) -> Unit:
    """Return the pint Unit for the DataArray units.

    Parameters
    ----------
    value : str
      Unit expression.

    Returns
    -------
    pint.Unit
      Pint compatible units.

    """

    def _transform(s):
        """Convert a CF-unit string to a pint expression."""
        return re.subn(r"\^?(-?\d)", r"**\g<1>", s)[0]

    value = value.replace("%", "pct")
    try:  # Pint compatible
        return u.parse_expression(value).units
    except (
        pint.UndefinedUnitError,
        pint.DimensionalityError,
    ):  # Convert from CF-units to pint-compatible
        return u.parse_expression(_transform(value)).units


def get_time_frequency(d: xr.Dataset):
    freq = xr.infer_freq(d.time)
    if freq is None:
        raise TypeError()
    offset = [int(parse_offset(freq)[0]), parse_offset(freq)[1]]

    time_units = {
        "s": "second",
        "m": "minute",
        "h": "hour",
        "D": "day",
        "W": "week",
        "Y": "year",
    }
    if offset[1] in ["S", "M", "H"]:
        offset[1] = offset[1].lower()
    offset_meaning = time_units[offset[1]]
    return offset, offset_meaning

1	import re	3✔
2
3	import pint	3✔
4	import xarray as xr	3✔
5	from pint import Unit	3✔
6	from xclim.core.calendar import parse_offset	3✔
7
8	KiB = int(pow(2, 10))	3✔
9	MiB = int(pow(2, 20))	3✔
10	GiB = int(pow(2, 30))	3✔
11
12	u = pint.UnitRegistry(autoconvert_offset_to_baseunit=True)	3✔
13
14	# General purpose units
15	null = pint.Context("none")	3✔
16	u.add_context(null)	3✔
17	u.define("fraction = []")	3✔
18	u.define("percent = 1e-2 fraction = pct")	3✔
19	u.define("degC = kelvin; offset: 273.15 = celsius = C")	3✔
20	u.define("d = day")	3✔
21	u.define("h = hour") # Not the Planck constant...	3✔
22	u.define("[speed] = [length] / [time]")	3✔
23
24	u.define("[precipitation] = [mass] / [length] ** 2 / [time]")	3✔
25	u.define("mmday = 1 kg / meter ** 2 / day")	3✔
26	u.define("[discharge] = [length] ** 3 / [time]")	3✔
27	u.define("cms = meter ** 3 / second")	3✔
28
29	# Allows transformation from mm day to kg / m2 / s
30	hq = pint.Context("hq")	3✔
31
32	hq.add_transformation(	3✔
33	"[mass] / [length]**2",
34	"[length]",
35	lambda ureg, x: x / (1000 * ureg.kg / ureg.m**3),
36	)
37	hq.add_transformation(	3✔
38	"[mass] / [length]**2 / [time]",
39	"[length] / [time]",
40	lambda ureg, x: x / (1000 * ureg.kg / ureg.m**3),
41	)
42	hq.add_transformation(	3✔
43	"[length] / [time]",
44	"[mass] / [length]**2 / [time]",
45	lambda ureg, x: x * (1000 * ureg.kg / ureg.m**3),
46	)
47	u.add_context(hq)	3✔
48	u.enable_contexts(hq)	3✔
49
50
51	def units2pint(value: str) -> Unit:	3✔
52	"""Return the pint Unit for the DataArray units.
53
54	Parameters
55	----------
56	value : str
57	Unit expression.
58
59	Returns
60	-------
61	pint.Unit
62	Pint compatible units.
63
64	"""
65
66	def _transform(s):	×
67	"""Convert a CF-unit string to a pint expression."""
68	return re.subn(r"\^?(-?\d)", r"**\g<1>", s)[0]	×
69
70	value = value.replace("%", "pct")	×
71	try: # Pint compatible	×
72	return u.parse_expression(value).units	×
73	except (	×
74	pint.UndefinedUnitError,
75	pint.DimensionalityError,
76	): # Convert from CF-units to pint-compatible
77	return u.parse_expression(_transform(value)).units	×
78
79
80	def get_time_frequency(d: xr.Dataset):	3✔
NEW 81	freq = xr.infer_freq(d.time)	×
NEW 82	if freq is None:	×
NEW 83	raise TypeError()	×
NEW 84	offset = [int(parse_offset(freq)[0]), parse_offset(freq)[1]]	×
85
NEW 86	time_units = {	×
87	"s": "second",
88	"m": "minute",
89	"h": "hour",
90	"D": "day",
91	"W": "week",
92	"Y": "year",
93	}
NEW 94	if offset[1] in ["S", "M", "H"]:	×
NEW 95	offset[1] = offset[1].lower()	×
NEW 96	offset_meaning = time_units[offset[1]]	×
NEW 97	return offset, offset_meaning	×

Ouranosinc / miranda / 2222748844

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous