2117681512

Build Type

github

Committed by GitHub

Commit Message

Merge 9bbb68671 into bf78f91b7

Pull Request Pull Request #24: Add CMIP file structure, use pyessv controlled vocabularies, and major refactoring

Run Details

241 of 1086 new or added lines in 35 files covered. (22.19%)

13 existing lines in 4 files now uncovered.

735 of 3230 relevant lines covered (22.76%)

0.68 hits per line

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

31.25

/miranda/convert/_data_definitions.py

import json
import logging.config
import os
from pathlib import Path
from typing import Dict, List, Union

from miranda.scripting import LOGGING_CONFIG
from miranda.storage import report_file_size

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = [
    "gather_agcfsr",
    "gather_agmerra",
    "gather_era5_land",
    "gather_era5_single_levels",
    "gather_nrcan_gridded_obs",
    "gather_sc_earth",
    "gather_wfdei_gem_capa",
    "era5_variables",
    "nrcan_variables",
    "nasa_ag_variables",
    "sc_earth_variables",
    "wfdei_gem_capa_variables",
    "reanalysis_project_institutes",
    "xarray_frequencies_to_cmip6like",
]

data_folder = Path(__file__).parent / "data"
era5_variables = json.load(open(data_folder / "ecmwf_cf_attrs.json"))[
    "variable_entry"
].keys()
nrcan_variables = ["tasmin", "tasmax", "pr"]
nasa_ag_variables = json.load(open(data_folder / "nasa_cf_attrs.json"))[
    "variable_entry"
].keys()
sc_earth_variables = ["prcp", "tdew", "tmean", "trange", "wind"]
wfdei_gem_capa_variables = json.load(open(data_folder / "usask_cf_attrs.json"))[
    "variable_entry"
].keys()

reanalysis_project_institutes = {
    "cfsr": "ncar",
    "era5": "ecmwf",
    "era5-single-levels-preliminary-back-extension": "ecmwf",
    "era5-single-levels": "ecmwf",
    "era5-land": "ecmwf",
    "merra2": "nasa",
    "nrcan-gridded-10km": "nrcan",
    "wfdei-gem-capa": "usask",
}


# Manually map xarray frequencies to CMIP6/CMIP5 controlled vocabulary.
# see: https://github.com/ES-DOC/pyessv-archive
xarray_frequencies_to_cmip6like = {
    "H": "hr",
    "D": "day",
    "W": "sem",
    "M": "mon",
    "Q": "qtr",  # TODO does this make sense? does not exist in cmip6 CV
    "A": "yr",
    "Y": "yr",
}


def gather_era5_single_levels(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # ERA5 source data
    source_era5 = Path(path)
    logging.info("Gathering ERA5 from %s" % source_era5.as_posix())
    infiles_era5 = list()
    for v in era5_variables:
        infiles_era5.extend(list(sorted(source_era5.rglob(f"{v}_*.nc"))))
    logging.info(
        f"Found {len(infiles_era5)} files, totalling {report_file_size(infiles_era5)}."
    )
    return {"era5-single-levels": infiles_era5}


def gather_era5_land_sea_mask(path: Union[str, os.PathLike]) -> Dict:
    try:
        land_sea_mask = dict(lsm=next(Path(path).glob("sftlf*.nc")))
    except StopIteration:
        logging.error("No land_sea_mask found for ERA5.")
        raise FileNotFoundError()
    return land_sea_mask


def gather_era5_land(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # ERA5-Land source data
    source_era5l = Path(path)
    logging.info("Gathering ERA5-Land from %s" % source_era5l.as_posix())
    infiles_era5l = list()
    for v in era5_variables:
        infiles_era5l.extend(list(sorted(source_era5l.rglob(f"{v}_*.nc"))))
    logging.info(
        f"Found {len(infiles_era5l)} files, totalling {report_file_size(infiles_era5l)}."
    )
    return {"era5-land": infiles_era5l}


def gather_agmerra(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # agMERRA source data
    source_agmerra = Path(path)
    logging.info("Gathering agMERRA from %s" % source_agmerra.as_posix())
    infiles_agmerra = list()
    for v in nasa_ag_variables:
        infiles_agmerra.extend(list(sorted(source_agmerra.rglob(f"AgMERRA_*_{v}.nc4"))))
    logging.info(
        f"Found {len(infiles_agmerra)} files, totalling {report_file_size(infiles_agmerra)}."
    )
    return dict(cfsr=infiles_agmerra)


def gather_agcfsr(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # agCFSR source data
    source_agcfsr = Path(path)
    logging.info("Gathering CFSR from %s" % source_agcfsr.as_posix())
    infiles_agcfsr = list()
    for v in nasa_ag_variables:
        infiles_agcfsr.extend(list(sorted(source_agcfsr.rglob(f"AgCFSR_*_{v}.nc4"))))
    logging.info(
        f"Found {len(infiles_agcfsr)} files, totalling {report_file_size(infiles_agcfsr)}."
    )
    return dict(cfsr=infiles_agcfsr)


def gather_nrcan_gridded_obs(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # NRCan Gridded Obs source data
    source_nrcan = Path(path)
    logging.info("Gathering NRCAN Gridded Obs from %s" % source_nrcan.as_posix())
    infiles_nrcan = list()
    for v in nrcan_variables:
        infiles_nrcan.extend(list(sorted(source_nrcan.joinpath(v).glob(f"*{v}_*.nc"))))
    logging.info(
        f"Found {len(infiles_nrcan)} files, totalling {report_file_size(infiles_nrcan)}."
    )
    return dict(nrcan=infiles_nrcan)


def gather_wfdei_gem_capa(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # WFDEI-GEM-CaPa source data
    source_wfdei = Path(path)
    logging.info("Gathering WFDEI-GEM_CaPa from %s" % source_wfdei.as_posix())
    infiles_wfdei = list()
    for v in wfdei_gem_capa_variables:
        infiles_wfdei.extend(list(sorted(source_wfdei.rglob(f"{v}_*.nc"))))
    logging.info(
        f"Found {len(infiles_wfdei)} files, totalling {report_file_size(infiles_wfdei)}."
    )
    return {"wfdei-gem-capa": infiles_wfdei}


def gather_sc_earth(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:
    # SC_Earth source data
    source_sc_earth = Path(path)
    logging.info("Gathering SC-Earth from %s" % source_sc_earth.as_posix())
    infiles_sc_earth = list()
    for v in sc_earth_variables:
        infiles_sc_earth.extend(
            list(sorted(source_sc_earth.rglob(f"SC-Earth_{v}_*.nc")))
        )
    logging.info(
        f"Found {len(infiles_sc_earth)} files, totalling {report_file_size(infiles_sc_earth)}."
    )
    return {"wfdei-gem-capa": infiles_sc_earth}

1	import json	3✔
2	import logging.config	3✔
3	import os	3✔
4	from pathlib import Path	3✔
5	from typing import Dict, List, Union	3✔
6
7	from miranda.scripting import LOGGING_CONFIG	3✔
8	from miranda.storage import report_file_size	3✔
9
10	logging.config.dictConfig(LOGGING_CONFIG)	3✔
11
12	__all__ = [	3✔
13	"gather_agcfsr",
14	"gather_agmerra",
15	"gather_era5_land",
16	"gather_era5_single_levels",
17	"gather_nrcan_gridded_obs",
18	"gather_sc_earth",
19	"gather_wfdei_gem_capa",
20	"era5_variables",
21	"nrcan_variables",
22	"nasa_ag_variables",
23	"sc_earth_variables",
24	"wfdei_gem_capa_variables",
25	"reanalysis_project_institutes",
26	"xarray_frequencies_to_cmip6like",
27	]
28
29	data_folder = Path(__file__).parent / "data"	3✔
30	era5_variables = json.load(open(data_folder / "ecmwf_cf_attrs.json"))[	3✔
31	"variable_entry"
32	].keys()
33	nrcan_variables = ["tasmin", "tasmax", "pr"]	3✔
34	nasa_ag_variables = json.load(open(data_folder / "nasa_cf_attrs.json"))[	3✔
35	"variable_entry"
36	].keys()
37	sc_earth_variables = ["prcp", "tdew", "tmean", "trange", "wind"]	3✔
38	wfdei_gem_capa_variables = json.load(open(data_folder / "usask_cf_attrs.json"))[	3✔
39	"variable_entry"
40	].keys()
41
42	reanalysis_project_institutes = {	3✔
43	"cfsr": "ncar",
44	"era5": "ecmwf",
45	"era5-single-levels-preliminary-back-extension": "ecmwf",
46	"era5-single-levels": "ecmwf",
47	"era5-land": "ecmwf",
48	"merra2": "nasa",
49	"nrcan-gridded-10km": "nrcan",
50	"wfdei-gem-capa": "usask",
51	}
52
53
54	# Manually map xarray frequencies to CMIP6/CMIP5 controlled vocabulary.
55	# see: https://github.com/ES-DOC/pyessv-archive
56	xarray_frequencies_to_cmip6like = {	3✔
57	"H": "hr",
58	"D": "day",
59	"W": "sem",
60	"M": "mon",
61	"Q": "qtr", # TODO does this make sense? does not exist in cmip6 CV
62	"A": "yr",
63	"Y": "yr",
64	}
65
66
67	def gather_era5_single_levels(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
68	# ERA5 source data
69	source_era5 = Path(path)	×
70	logging.info("Gathering ERA5 from %s" % source_era5.as_posix())	×
71	infiles_era5 = list()	×
NEW 72	for v in era5_variables:	×
73	infiles_era5.extend(list(sorted(source_era5.rglob(f"{v}_*.nc"))))	×
NEW 74	logging.info(	×
75	f"Found {len(infiles_era5)} files, totalling {report_file_size(infiles_era5)}."
76	)
UNCOV 77	return {"era5-single-levels": infiles_era5}	×
78
79
80	def gather_era5_land_sea_mask(path: Union[str, os.PathLike]) -> Dict:	3✔
81	try:	×
82	land_sea_mask = dict(lsm=next(Path(path).glob("sftlf*.nc")))	×
83	except StopIteration:	×
84	logging.error("No land_sea_mask found for ERA5.")	×
85	raise FileNotFoundError()	×
UNCOV 86	return land_sea_mask	×
87
88
89	def gather_era5_land(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
90	# ERA5-Land source data
91	source_era5l = Path(path)	×
92	logging.info("Gathering ERA5-Land from %s" % source_era5l.as_posix())	×
93	infiles_era5l = list()	×
NEW 94	for v in era5_variables:	×
95	infiles_era5l.extend(list(sorted(source_era5l.rglob(f"{v}_*.nc"))))	×
NEW 96	logging.info(	×
97	f"Found {len(infiles_era5l)} files, totalling {report_file_size(infiles_era5l)}."
98	)
UNCOV 99	return {"era5-land": infiles_era5l}	×
100
101
102	def gather_agmerra(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
103	# agMERRA source data
104	source_agmerra = Path(path)	×
105	logging.info("Gathering agMERRA from %s" % source_agmerra.as_posix())	×
106	infiles_agmerra = list()	×
107	for v in nasa_ag_variables:	×
108	infiles_agmerra.extend(list(sorted(source_agmerra.rglob(f"AgMERRA_*_{v}.nc4"))))	×
NEW 109	logging.info(	×
110	f"Found {len(infiles_agmerra)} files, totalling {report_file_size(infiles_agmerra)}."
111	)
UNCOV 112	return dict(cfsr=infiles_agmerra)	×
113
114
115	def gather_agcfsr(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
116	# agCFSR source data
117	source_agcfsr = Path(path)	×
118	logging.info("Gathering CFSR from %s" % source_agcfsr.as_posix())	×
119	infiles_agcfsr = list()	×
120	for v in nasa_ag_variables:	×
121	infiles_agcfsr.extend(list(sorted(source_agcfsr.rglob(f"AgCFSR_*_{v}.nc4"))))	×
NEW 122	logging.info(	×
123	f"Found {len(infiles_agcfsr)} files, totalling {report_file_size(infiles_agcfsr)}."
124	)
UNCOV 125	return dict(cfsr=infiles_agcfsr)	×
126
127
128	def gather_nrcan_gridded_obs(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
129	# NRCan Gridded Obs source data
130	source_nrcan = Path(path)	×
131	logging.info("Gathering NRCAN Gridded Obs from %s" % source_nrcan.as_posix())	×
NEW 132	infiles_nrcan = list()	×
NEW 133	for v in nrcan_variables:	×
NEW 134	infiles_nrcan.extend(list(sorted(source_nrcan.joinpath(v).glob(f"{v}_.nc"))))	×
NEW 135	logging.info(	×
136	f"Found {len(infiles_nrcan)} files, totalling {report_file_size(infiles_nrcan)}."
137	)
138	return dict(nrcan=infiles_nrcan)	×
139
140
141	def gather_wfdei_gem_capa(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
142	# WFDEI-GEM-CaPa source data
143	source_wfdei = Path(path)	×
144	logging.info("Gathering WFDEI-GEM_CaPa from %s" % source_wfdei.as_posix())	×
145	infiles_wfdei = list()	×
146	for v in wfdei_gem_capa_variables:	×
147	infiles_wfdei.extend(list(sorted(source_wfdei.rglob(f"{v}_*.nc"))))	×
NEW 148	logging.info(	×
149	f"Found {len(infiles_wfdei)} files, totalling {report_file_size(infiles_wfdei)}."
150	)
UNCOV 151	return {"wfdei-gem-capa": infiles_wfdei}	×
152
153
154	def gather_sc_earth(path: Union[str, os.PathLike]) -> Dict[str, List[Path]]:	3✔
155	# SC_Earth source data
156	source_sc_earth = Path(path)	×
157	logging.info("Gathering SC-Earth from %s" % source_sc_earth.as_posix())	×
158	infiles_sc_earth = list()	×
159	for v in sc_earth_variables:	×
160	infiles_sc_earth.extend(	×
161	list(sorted(source_sc_earth.rglob(f"SC-Earth_{v}_*.nc")))
162	)
NEW 163	logging.info(	×
164	f"Found {len(infiles_sc_earth)} files, totalling {report_file_size(infiles_sc_earth)}."
165	)
UNCOV 166	return {"wfdei-gem-capa": infiles_sc_earth}	×

Ouranosinc / miranda / 2117681512

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