2151466472

Build Type

github

Committed by GitHub

Commit Message

Merge fc8b4c32c into bf78f91b7

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

Run Details

241 of 1100 new or added lines in 35 files covered. (21.91%)

13 existing lines in 4 files now uncovered.

735 of 3244 relevant lines covered (22.66%)

0.68 hits per line

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

18.31

/miranda/ecmwf/_tigge.py

import functools
import logging.config
import multiprocessing
import os
from datetime import datetime as dt
from datetime import timedelta as td
from pathlib import Path
from typing import List, Optional

from ecmwfapi import ECMWFDataServer

from miranda.scripting import LOGGING_CONFIG

logging.config.dictConfig(LOGGING_CONFIG)

__all__ = ["request_tigge"]


def request_tigge(
    variables: List[str],
    providers: Optional[List[str]] = None,
    *,
    forecast_type: str = "pf",
    times: Optional[List[str]] = None,
    dates: Optional[List[str]] = None,
    date_start: Optional[str] = None,
    date_end: Optional[str] = None,
    output_folder: Optional[os.PathLike] = None,
    processes: int = 4,
) -> None:
    """Request tigge data from ECMWF in grib format.

    Parameters
    ----------
    variables : List[str]
    providers : List[str], optional
    forecast_type: {"pf", "cf"}
    times : List[str], optional
    dates : List[str]. optional
    date_start : str, optional
    date_end : str, optional
    output_folder : os.PathLike, optional
    processes : int

    Returns
    -------
    None
    """

    def _tigge_request(
        variable_name: str,
        variable_code: str,
        time: str,
        fc_type: str,
        provider: str,
        nums: Optional[int],
        date: str,
    ):
        """Launch formatted request."""
        number_range = ""
        if nums:
            number_range = "/".join([str(n) for n in range(1, nums + 1)])
        output_name = (
            f"{variable_name}_{provider}_{'-'.join(time.split('/'))}_tigge_reanalysis_6h_"
            f"{date.split('/')[0]}_{date.split('/')[-1]}.grib2"
        )

        # Note: This is only valid for ECMWF at the moment.
        steps = (
            "0/6/12/18/24/30/36/42/48/54/60/66/72/78/84/90/96/102/108/114/120/126/132/138/144/150/156/162/168/"
            "174/180/186/192/198/204/210/216/222/228/234/240/246/252/258/264/270/276/282/288/294/300/306/312/318/324/"
            "330/336/342/348/354/360"
        )

        # Remove steps 0 for tasmax and tasmin
        if variable_code in [121, 122]:
            steps = steps[2:]

        request = {
            "class": "ti",
            "dataset": "tigge",
            "date": date,
            "expver": "prod",
            "grid": "0.5/0.5",
            "levtype": "sfc",
            "origin": provider,
            "param": variable_code,
            "step": steps,
            "time": time,
            "type": fc_type,
            "target": output_name,
        }
        if nums:
            request.update({"number": number_range})

        server = ECMWFDataServer()
        server.retrieve(request)

    # Providers of interest
    if providers is None:
        providers = ["ecmf"]

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

    if not times:
        times = ["00/12"]

    if date_start and date_end:
        start = (dt.today() - td(days=4)).strftime("%Y-%m-%d")
        finish = (dt.today() - td(days=3)).strftime("%Y-%m-%d")
        date_range = f"{start}/to/{finish}"
        times.append(date_range)
    elif dates:
        pass
    else:
        raise ValueError()

    tigge_variables = dict()
    tigge_variables["uas"] = 165
    tigge_variables["vas"] = 166
    tigge_variables["tas"] = 167
    tigge_variables["tds"] = 168
    tigge_variables["pr"] = 222
    tigge_variables["swe"] = 228144
    tigge_variables["tasmax"] = 121
    tigge_variables["tasmin"] = 122

    if variables is None:
        variables = tigge_variables.keys()

    project_members = dict(
        dems=44,
        kwbc=26,
        cwao=20,
        ecmf=50,
        babj=30,
        egrr=23,
        rksl=24,
        rjtd=50,
        edzw=40,
        lfpw=34,
        ammc=32,
    )

    for v in variables:
        var_num = tigge_variables[v]
        for t in times:
            for d in dates:
                for p in providers:
                    proc = multiprocessing.Pool(processes=processes)
                    config = dict(
                        variable_name=v,
                        variable_code=var_num,
                        time=t,
                        provider=p,
                        dates=d,
                    )
                    num_dict = dict()
                    if forecast_type == "pf":
                        numbers = project_members[p]
                        num_dict[numbers] = numbers

                    func = functools.partial(_tigge_request, **config, **num_dict)

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

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

1	import functools	3✔
2	import logging.config	3✔
3	import multiprocessing	3✔
4	import os	3✔
5	from datetime import datetime as dt	3✔
6	from datetime import timedelta as td	3✔
7	from pathlib import Path	3✔
8	from typing import List, Optional	3✔
9
10	from ecmwfapi import ECMWFDataServer	3✔
11
12	from miranda.scripting import LOGGING_CONFIG	3✔
13
14	logging.config.dictConfig(LOGGING_CONFIG)	3✔
15
16	__all__ = ["request_tigge"]	3✔
17
18
19	def request_tigge(	3✔
20	variables: List[str],
21	providers: Optional[List[str]] = None,
22	*,
23	forecast_type: str = "pf",
24	times: Optional[List[str]] = None,
25	dates: Optional[List[str]] = None,
26	date_start: Optional[str] = None,
27	date_end: Optional[str] = None,
28	output_folder: Optional[os.PathLike] = None,
29	processes: int = 4,
30	) -> None:
31	"""Request tigge data from ECMWF in grib format.
32
33	Parameters
34	----------
35	variables : List[str]
36	providers : List[str], optional
37	forecast_type: {"pf", "cf"}
38	times : List[str], optional
39	dates : List[str]. optional
40	date_start : str, optional
41	date_end : str, optional
42	output_folder : os.PathLike, optional
43	processes : int
44
45	Returns
46	-------
47	None
48	"""
49
50	def _tigge_request(	×
51	variable_name: str,
52	variable_code: str,
53	time: str,
54	fc_type: str,
55	provider: str,
56	nums: Optional[int],
57	date: str,
58	):
59	"""Launch formatted request."""
60	number_range = ""	×
NEW 61	if nums:	×
NEW 62	number_range = "/".join([str(n) for n in range(1, nums + 1)])	×
63	output_name = (	×
64	f"{variable_name}_{provider}_{'-'.join(time.split('/'))}_tigge_reanalysis_6h_"
65	f"{date.split('/')[0]}_{date.split('/')[-1]}.grib2"
66	)
67
68	# Note: This is only valid for ECMWF at the moment.
69	steps = (	×
70	"0/6/12/18/24/30/36/42/48/54/60/66/72/78/84/90/96/102/108/114/120/126/132/138/144/150/156/162/168/"
71	"174/180/186/192/198/204/210/216/222/228/234/240/246/252/258/264/270/276/282/288/294/300/306/312/318/324/"
72	"330/336/342/348/354/360"
73	)
74
75	# Remove steps 0 for tasmax and tasmin
76	if variable_code in [121, 122]:	×
77	steps = steps[2:]	×
78
79	request = {	×
80	"class": "ti",
81	"dataset": "tigge",
82	"date": date,
83	"expver": "prod",
84	"grid": "0.5/0.5",
85	"levtype": "sfc",
86	"origin": provider,
87	"param": variable_code,
88	"step": steps,
89	"time": time,
90	"type": fc_type,
91	"target": output_name,
92	}
NEW 93	if nums:	×
94	request.update({"number": number_range})	×
95
96	server = ECMWFDataServer()	×
97	server.retrieve(request)	×
98
99	# Providers of interest
100	if providers is None:	×
101	providers = ["ecmf"]	×
102
103	if output_folder is None:	×
104	target = Path().cwd().joinpath("download")	×
105	else:
106	target = Path(output_folder)	×
107	Path(target).mkdir(exist_ok=True)	×
108	os.chdir(target)	×
109
110	if not times:	×
111	times = ["00/12"]	×
112
113	if date_start and date_end:	×
114	start = (dt.today() - td(days=4)).strftime("%Y-%m-%d")	×
115	finish = (dt.today() - td(days=3)).strftime("%Y-%m-%d")	×
116	date_range = f"{start}/to/{finish}"	×
117	times.append(date_range)	×
118	elif dates:	×
119	pass	×
120	else:
121	raise ValueError()	×
122
123	tigge_variables = dict()	×
124	tigge_variables["uas"] = 165	×
125	tigge_variables["vas"] = 166	×
126	tigge_variables["tas"] = 167	×
127	tigge_variables["tds"] = 168	×
128	tigge_variables["pr"] = 222	×
129	tigge_variables["swe"] = 228144	×
130	tigge_variables["tasmax"] = 121	×
131	tigge_variables["tasmin"] = 122	×
132
133	if variables is None:	×
134	variables = tigge_variables.keys()	×
135
136	project_members = dict(	×
137	dems=44,
138	kwbc=26,
139	cwao=20,
140	ecmf=50,
141	babj=30,
142	egrr=23,
143	rksl=24,
144	rjtd=50,
145	edzw=40,
146	lfpw=34,
147	ammc=32,
148	)
149
150	for v in variables:	×
151	var_num = tigge_variables[v]	×
152	for t in times:	×
153	for d in dates:	×
154	for p in providers:	×
155	proc = multiprocessing.Pool(processes=processes)	×
156	config = dict(	×
157	variable_name=v,
158	variable_code=var_num,
159	time=t,
160	provider=p,
161	dates=d,
162	)
163	num_dict = dict()	×
164	if forecast_type == "pf":	×
165	numbers = project_members[p]	×
166	num_dict[numbers] = numbers	×
167
168	func = functools.partial(_tigge_request, config, num_dict)	×
169
170	logging.info([func, dt.now().strftime("%Y-%m-%d %X")])	×
171
172	proc.map(func, dates)	×
173	proc.close()	×
174	proc.join()	×

Ouranosinc / miranda / 2151466472

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