1883563084

Build Type

github

Committed by GitHub

Commit Message

Merge 4555deb62 into 4a2a34e15

Pull Request Pull Request #24: Add CMIP file structure - WIP

Run Details

41 of 361 new or added lines in 19 files covered. (11.36%)

3 existing lines in 2 files now uncovered.

627 of 2829 relevant lines covered (22.16%)

0.66 hits per line

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

21.74

/miranda/ecmwf/_ecmwf.py

import functools
import logging.config
import multiprocessing
import os
from datetime import date
from datetime import datetime as dt
from pathlib import Path
from typing import List, Mapping, Optional, Tuple, Union

from cdsapi import Client

from miranda.gis.subset import subsetting_domains
from miranda.scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = ["request_ecmwf"]


def request_ecmwf(
    variables: Optional[Mapping[str, str]],
    projects: List[str],
    *,
    domain: str = "AMNO",
    output_folder: Optional[Union[str, os.PathLike]] = None,
    year_start: Union[str, int] = 1950,
    year_end: Optional[Union[str, int]] = None,
    processes: int = 10,
) -> None:
    """Request ERA5/ERA5-Land from Copernicus Data Store in NetCDF4 format.

    Parameters
    ----------
    variables: Mapping[str, str]
    projects : List[{"era5", "era5-land"}]
    domain : {"GLOBAL", "AMNO", "CAN", "QC", "MTL"}
    output_folder : str or os.PathLike, optional
    year_start : int
    year_end : int, optional
    processes : int

    Returns
    -------
    None
    """
    # Variables of interest
    variable_reference = dict()
    variable_reference["era5-land"] = dict(
        tp="total_precipitation",
        v10="10m_v_component_of_wind",
        u10="10m_u_component_of_wind",
        d2m="2m_dewpoint_temperature",
        t2m="2m_temperature",
        pev="potential evaporation",
        sde="snow_depth",
        sd="snow_depth_water_equivalent",
        sf="snowfall",
    )
    variable_reference["era5"] = dict(
        tp="total_precipitation",
        v10="10m_v_component_of_wind",
        u10="10m_u_component_of_wind",
        d2m="2m_dewpoint_temperature",
        t2m="2m_temperature",
        pev="potential evaporation",
        # sde= Not available for era5
        sd="snow_depth",  # note difference in name vs era5-land cf_variable == snw
        sf="snowfall",
    )

    if year_end is None:
        year_end = date.today().year
    years = range(int(year_start), int(year_end) + 1)

    months = [str(d).zfill(2) for d in range(1, 13)]
    yearmonth = list()
    for y in years:
        for m in months:
            yearmonth.append((y, m))

    project_names = dict()
    if "era5" in projects:
        project_names["era5"] = "reanalysis-era5-single-levels"
        # project_names.append("reanalysis-era5-single-levels")
    if "era5-land" in projects:
        project_names["era5-land"] = "reanalysis-era5-land"
    # product = project_names[0].split("-")[0]

    if output_folder is None:
        target = Path().cwd().joinpath("downloaded")
    else:
        target = output_folder
    Path(target).mkdir(exist_ok=True)
    os.chdir(target)

    for key, p in project_names.items():
        product = p.split("-")[0]
        v_requested = dict()
        if variables:
            for v in variables:
                if v in variable_reference[key]:
                    v_requested[v] = variable_reference[key][v]
        else:
            v_requested = variable_reference[key]
        proc = multiprocessing.Pool(processes=processes)
        func = functools.partial(_request_direct_era, v_requested, p, domain, product)

        logging.info([func, dt.now().strftime("%Y-%m-%d %X")])

        proc.map(func, yearmonth)
        proc.close()
        proc.join()


def _request_direct_era(
    variables: Mapping[str, str],
    project: str,
    domain: str,
    product: str,
    yearmonth: Tuple[int, str],
):
    """Launch formatted request."""
    year, month = yearmonth
    days = [str(d).zfill(2) for d in range(32)]
    times = [f"{str(t).zfill(2)}:00" for t in range(24)]

    if domain.upper() == "AMNO":
        domain = "NAM"
    region = subsetting_domains(domain)

    c = Client()

    if project in ["reanalysis-era5-single-levels", "reanalysis-era5-land"]:
        timestep = "hourly"
    else:
        raise NotImplementedError(project)

    for var in variables.keys():
        netcdf_name = (
            f"{var}_{timestep}_ecmwf_{'-'.join(project.split('-')[1:])}"
            f"_{product}_{domain.upper()}_{year}{month}.nc"
        )

        if Path(netcdf_name).exists():
            logging.info("Dataset %s already exists. Continuing..." % netcdf_name)
            continue

        request_kwargs = dict(
            variable=variables[var],
            year=year,
            month=month,
            day=days,
            time=times,
            area=region,
            format="netcdf",
        )

        if project == "reanalysis-era5-single-levels":
            request_kwargs.update(dict(product_type=product))

        c.retrieve(
            project,
            request_kwargs,
            netcdf_name,
        )

1	import functools	3✔
2	import logging.config	3✔
3	import multiprocessing	3✔
4	import os	3✔
5	from datetime import date	3✔
6	from datetime import datetime as dt	3✔
7	from pathlib import Path	3✔
8	from typing import List, Mapping, Optional, Tuple, Union	3✔
9
10	from cdsapi import Client	3✔
11
12	from miranda.gis.subset import subsetting_domains	3✔
13	from miranda.scripting import LOGGING_CONFIG	3✔
14
15	logging.config.dictConfig(LOGGING_CONFIG)	3✔
16
17	__all__ = ["request_ecmwf"]	3✔
18
19
20	def request_ecmwf(	3✔
21	variables: Optional[Mapping[str, str]],
22	projects: List[str],
23	*,
24	domain: str = "AMNO",
25	output_folder: Optional[Union[str, os.PathLike]] = None,
26	year_start: Union[str, int] = 1950,
27	year_end: Optional[Union[str, int]] = None,
28	processes: int = 10,
29	) -> None:
30	"""Request ERA5/ERA5-Land from Copernicus Data Store in NetCDF4 format.
31
32	Parameters
33	----------
34	variables: Mapping[str, str]
35	projects : List[{"era5", "era5-land"}]
36	domain : {"GLOBAL", "AMNO", "CAN", "QC", "MTL"}
37	output_folder : str or os.PathLike, optional
38	year_start : int
39	year_end : int, optional
40	processes : int
41
42	Returns
43	-------
44	None
45	"""
46	# Variables of interest
47	variable_reference = dict()	×
48	variable_reference["era5-land"] = dict(	×
49	tp="total_precipitation",
50	v10="10m_v_component_of_wind",
51	u10="10m_u_component_of_wind",
52	d2m="2m_dewpoint_temperature",
53	t2m="2m_temperature",
54	pev="potential evaporation",
55	sde="snow_depth",
56	sd="snow_depth_water_equivalent",
57	sf="snowfall",
58	)
59	variable_reference["era5"] = dict(	×
60	tp="total_precipitation",
61	v10="10m_v_component_of_wind",
62	u10="10m_u_component_of_wind",
63	d2m="2m_dewpoint_temperature",
64	t2m="2m_temperature",
65	pev="potential evaporation",
66	# sde= Not available for era5
67	sd="snow_depth", # note difference in name vs era5-land cf_variable == snw
68	sf="snowfall",
69	)
70
71	if year_end is None:	×
72	year_end = date.today().year	×
73	years = range(int(year_start), int(year_end) + 1)	×
74
75	months = [str(d).zfill(2) for d in range(1, 13)]	×
76	yearmonth = list()	×
77	for y in years:	×
78	for m in months:	×
79	yearmonth.append((y, m))	×
80
81	project_names = dict()	×
82	if "era5" in projects:	×
83	project_names["era5"] = "reanalysis-era5-single-levels"	×
84	# project_names.append("reanalysis-era5-single-levels")
85	if "era5-land" in projects:	×
86	project_names["era5-land"] = "reanalysis-era5-land"	×
87	# product = project_names[0].split("-")[0]
88
89	if output_folder is None:	×
90	target = Path().cwd().joinpath("downloaded")	×
91	else:
92	target = output_folder	×
93	Path(target).mkdir(exist_ok=True)	×
94	os.chdir(target)	×
95
96	for key, p in project_names.items():	×
97	product = p.split("-")[0]	×
98	v_requested = dict()	×
99	if variables:	×
100	for v in variables:	×
101	if v in variable_reference[key]:	×
102	v_requested[v] = variable_reference[key][v]	×
103	else:
104	v_requested = variable_reference[key]	×
105	proc = multiprocessing.Pool(processes=processes)	×
106	func = functools.partial(_request_direct_era, v_requested, p, domain, product)	×
107
108	logging.info([func, dt.now().strftime("%Y-%m-%d %X")])	×
109
110	proc.map(func, yearmonth)	×
111	proc.close()	×
112	proc.join()	×
113
114
115	def _request_direct_era(	3✔
116	variables: Mapping[str, str],
117	project: str,
118	domain: str,
119	product: str,
120	yearmonth: Tuple[int, str],
121	):
122	"""Launch formatted request."""
123	year, month = yearmonth	×
124	days = [str(d).zfill(2) for d in range(32)]	×
NEW 125	times = [f"{str(t).zfill(2)}:00" for t in range(24)]	×
126
127	if domain.upper() == "AMNO":	×
128	domain = "NAM"	×
129	region = subsetting_domains(domain)	×
130
131	c = Client()	×
132
133	if project in ["reanalysis-era5-single-levels", "reanalysis-era5-land"]:	×
134	timestep = "hourly"	×
135	else:
136	raise NotImplementedError(project)	×
137
138	for var in variables.keys():	×
139	netcdf_name = (	×
140	f"{var}_{timestep}_ecmwf_{'-'.join(project.split('-')[1:])}"
141	f"_{product}_{domain.upper()}_{year}{month}.nc"
142	)
143
144	if Path(netcdf_name).exists():	×
145	logging.info("Dataset %s already exists. Continuing..." % netcdf_name)	×
146	continue	×
147
148	request_kwargs = dict(	×
149	variable=variables[var],
150	year=year,
151	month=month,
152	day=days,
153	time=times,
154	area=region,
155	format="netcdf",
156	)
157
158	if project == "reanalysis-era5-single-levels":	×
159	request_kwargs.update(dict(product_type=product))	×
160
161	c.retrieve(	×
162	project,
163	request_kwargs,
164	netcdf_name,
165	)

Ouranosinc / miranda / 1883563084

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