2073326836

Build # 2073326836

Build Type

github

Committed by GitHub

Commit Message

Merge 2b44d4a80 into 66761433b

Pull Request Pull Request #27: Add cordex aws downloader function

Run Details

17 of 53 new or added lines in 3 files covered. (32.08%)

0 of 0 relevant lines covered (NaN%)

0.0 hits per line

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

29.41

/miranda/ncar/_aws_cordex.py

import ast
import json
import logging
import logging.config
from json.decoder import JSONDecodeError
from pathlib import Path
from typing import Dict, List, Union

import intake
import schema
import xarray as xr
from dask.diagnostics import ProgressBar

from miranda.scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)


_allowed_args = schema.Schema(
    {
        schema.Optional("variable"): schema.Schema(
            schema.Or(
                [
                    "hurs",
                    "huss",
                    "pr",
                    "prec",
                    "ps",
                    "rsds",
                    "sfcWind",
                    "tas",
                    "tasmax",
                    "tasmin",
                    "temp",
                    "tmax",
                    "tmin",
                    "uas",
                    "vas",
                ]
            )
        ),
        schema.Optional("frequency"): "day",
        schema.Optional("scenario"): schema.Schema(
            schema.Or(["eval", "hist", "rcp45", "rcp85", "hist-rcp45", "hist-rcp85"])
        ),
        schema.Optional("grid"): schema.Or(["NAM-22i", "NAM-44i"]),
        schema.Optional("bias_correction"): schema.Or(
            ["raw", "mbcn-Daymet", "mbcn-gridMET"]
        ),
    }
)


def cordex_aws_download(
    target_folder: Union[str, Path], *, search: Dict[str, Union[str, List[str]]]
):
    schema.Schema(_allowed_args).validate(search)

    # Define the catalog description file location.
    catalog_url = (
        "https://ncar-na-cordex.s3-us-west-2.amazonaws.com/catalogs/aws-na-cordex.json"
    )

    # Interpret the "na-cordex-models" column as a list of values.
    col = intake.open_esm_datastore(
        catalog_url,
        csv_kwargs={"converters": {"na-cordex-models": ast.literal_eval}},
    )

    col_subset = col.search(**search)

    dsets = col_subset.to_dataset_dict(
        zarr_kwargs={"consolidated": True}, storage_options={"anon": True}
    )
    logging.info(f"\nDataset dictionary keys:\n {dsets.keys()}")

    dds = list()
    for key in list(dsets.keys()):
        logging.info(f"Adding {key} to the search criteria.")
        dds.append(dsets[key])

    with ProgressBar():
        for ds in dds:
            scen = ds.attrs["experiment_id"]
            for i, member in enumerate(ds.member_id):
                for var in ds.variables:
                    if var in search["variable"]:
                        var_out = var

                new_attrs = dict()
                for key, vals in ds.attrs.items():
                    try:
                        mapped = json.loads(vals)
                        if isinstance(mapped, dict):
                            new_attrs[key] = mapped.get(str(member.values), "")
                    except JSONDecodeError:
                        new_attrs[key] = vals
                    except TypeError:
                        new_attrs[key] = vals[0]

                years, datasets = zip(*ds.isel(member_id=i).groupby("time.year"))
                for d in datasets:
                    d.attrs.update(new_attrs)

                out_folder = target_folder.joinpath(f"{member.values}_{scen}")
                out_folder.mkdir(exist_ok=True)

                file_name_pattern = f"{var_out}_{member.values}_day_{scen}_{search['grid']}_{search['bias_correction']} "

                logging.info(f"Writing out files for {file_name_pattern}.")
                paths = [
                    out_folder.joinpath(f"{file_name_pattern}_{y}.nc")
                    for y in years
                    if not out_folder.joinpath(f"{file_name_pattern}_{y}.nc").exists()
                ]
                xr.save_mfdataset(datasets, paths, format="NETCDF4_CLASSIC")

1	import ast	3✔
2	import json	3✔
3	import logging	3✔
4	import logging.config	3✔
5	from json.decoder import JSONDecodeError	3✔
6	from pathlib import Path	3✔
7	from typing import Dict, List, Union	3✔
8
9	import intake	3✔
10	import schema	3✔
11	import xarray as xr	3✔
12	from dask.diagnostics import ProgressBar	3✔
13
14	from miranda.scripting import LOGGING_CONFIG	3✔
15
16	logging.config.dictConfig(LOGGING_CONFIG)	3✔
17
18
19	_allowed_args = schema.Schema(	3✔
20	{
21	schema.Optional("variable"): schema.Schema(
22	schema.Or(
23	[
24	"hurs",
25	"huss",
26	"pr",
27	"prec",
28	"ps",
29	"rsds",
30	"sfcWind",
31	"tas",
32	"tasmax",
33	"tasmin",
34	"temp",
35	"tmax",
36	"tmin",
37	"uas",
38	"vas",
39	]
40	)
41	),
42	schema.Optional("frequency"): "day",
43	schema.Optional("scenario"): schema.Schema(
44	schema.Or(["eval", "hist", "rcp45", "rcp85", "hist-rcp45", "hist-rcp85"])
45	),
46	schema.Optional("grid"): schema.Or(["NAM-22i", "NAM-44i"]),
47	schema.Optional("bias_correction"): schema.Or(
48	["raw", "mbcn-Daymet", "mbcn-gridMET"]
49	),
50	}
51	)
52
53
54	def cordex_aws_download(	3✔
55	target_folder: Union[str, Path], *, search: Dict[str, Union[str, List[str]]]
56	):
NEW 57	schema.Schema(_allowed_args).validate(search)	×
58
59	# Define the catalog description file location.
NEW 60	catalog_url = (	×
61	"https://ncar-na-cordex.s3-us-west-2.amazonaws.com/catalogs/aws-na-cordex.json"
62	)
63
64	# Interpret the "na-cordex-models" column as a list of values.
NEW 65	col = intake.open_esm_datastore(	×
66	catalog_url,
67	csv_kwargs={"converters": {"na-cordex-models": ast.literal_eval}},
68	)
69
NEW 70	col_subset = col.search(**search)	×
71
NEW 72	dsets = col_subset.to_dataset_dict(	×
73	zarr_kwargs={"consolidated": True}, storage_options={"anon": True}
74	)
NEW 75	logging.info(f"\nDataset dictionary keys:\n {dsets.keys()}")	×
76
NEW 77	dds = list()	×
NEW 78	for key in list(dsets.keys()):	×
NEW 79	logging.info(f"Adding {key} to the search criteria.")	×
NEW 80	dds.append(dsets[key])	×
81
NEW 82	with ProgressBar():	×
NEW 83	for ds in dds:	×
NEW 84	scen = ds.attrs["experiment_id"]	×
NEW 85	for i, member in enumerate(ds.member_id):	×
NEW 86	for var in ds.variables:	×
NEW 87	if var in search["variable"]:	×
NEW 88	var_out = var	×
89
NEW 90	new_attrs = dict()	×
NEW 91	for key, vals in ds.attrs.items():	×
NEW 92	try:	×
NEW 93	mapped = json.loads(vals)	×
NEW 94	if isinstance(mapped, dict):	×
NEW 95	new_attrs[key] = mapped.get(str(member.values), "")	×
NEW 96	except JSONDecodeError:	×
NEW 97	new_attrs[key] = vals	×
NEW 98	except TypeError:	×
NEW 99	new_attrs[key] = vals[0]	×
100
NEW 101	years, datasets = zip(*ds.isel(member_id=i).groupby("time.year"))	×
NEW 102	for d in datasets:	×
NEW 103	d.attrs.update(new_attrs)	×
104
NEW 105	out_folder = target_folder.joinpath(f"{member.values}_{scen}")	×
NEW 106	out_folder.mkdir(exist_ok=True)	×
107
NEW 108	file_name_pattern = f"{var_out}_{member.values}_day_{scen}_{search['grid']}_{search['bias_correction']} "	×
109
NEW 110	logging.info(f"Writing out files for {file_name_pattern}.")	×
NEW 111	paths = [	×
112	out_folder.joinpath(f"{file_name_pattern}_{y}.nc")
113	for y in years
114	if not out_folder.joinpath(f"{file_name_pattern}_{y}.nc").exists()
115	]
NEW 116	xr.save_mfdataset(datasets, paths, format="NETCDF4_CLASSIC")	×

Ouranosinc / miranda / 2073326836

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