5247671608

pending completion

Build # 5247671608

Build Type

push

github

Committed by

pre-commit-ci[bot]

Commit Message

[pre-commit.ci] auto fixes from pre-commit.com hooks

for more information, see https://pre-commit.ci

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

4 existing lines in 2 files now uncovered.

4477 of 5167 relevant lines covered (86.65%)

0.87 hits per line

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

97.37

/disdrodb/l0/issue.py

#!/usr/bin/env python3

# -----------------------------------------------------------------------------.
# Copyright (c) 2021-2022 DISDRODB developers
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program.  If not, see <http://www.gnu.org/licenses/>.
# -----------------------------------------------------------------------------.

import logging
import os

import numpy as np
import pandas as pd
import yaml

from disdrodb.utils.logger import log_error

logger = logging.getLogger(__name__)

####--------------------------------------------------------------------------.
#### Checks


def is_numpy_array_string(arr):
    """Check if the numpy array contains strings

    Parameters
    ----------
    arr : numpy array
        Numpy array to check.
    """

    dtype = arr.dtype.type
    return dtype == np.str_ or dtype == np.unicode_


def is_numpy_array_datetime(arr):
    """Check if the numpy array contains datetime64

    Parameters
    ----------
    arr : numpy array
        Numpy array to check.

    Returns
    -------
    numpy array
        Numpy array checked.
    """
    return arr.dtype.type == np.datetime64


def _check_timestep_datetime_accuracy(timesteps, unit="s"):
    """Check the accuracy of the numpy datetime array.

    Parameters
    ----------
    timesteps : numpy array
        Numpy array to check.
    unit : str, optional
        Unit, by default "s"

    Returns
    -------
    numpy array
        Numpy array checked.

    Raises
    ------
    ValueError
    """
    if not timesteps.dtype == f"<M8[{unit}]":
        msg = f"The timesteps does not have datetime64 {unit} accuracy."
        log_error(logger, msg=msg, verbose=False)
        raise ValueError(msg)
    return timesteps


def _check_timestep_string_second_accuracy(timesteps, n=19):
    """Check the timesteps string are provided with second accuracy.

    Note: it assumes the YYYY-mm-dd HH:MM:SS format
    """
    n_characters = np.char.str_len(timesteps)
    mispecified_timesteps = timesteps[n_characters != 19]
    if len(mispecified_timesteps) > 0:
        msg = (
            f"The following timesteps are mispecified: {mispecified_timesteps}. Expecting the YYYY-mm-dd HH:MM:SS"
            " format."
        )
        log_error(logger, msg=msg, verbose=False)
        raise ValueError(msg)
    return timesteps


def _check_timesteps_string(timesteps):
    """Check timesteps string validity.

    It expects a list of timesteps strings in YYYY-mm-dd HH:MM:SS format with second accuracy.
    """
    timesteps = np.asarray(timesteps)
    timesteps = _check_timestep_string_second_accuracy(timesteps)
    # Reformat as datetime64 with second accuracy
    new_timesteps = pd.to_datetime(timesteps, format="%Y-%m-%d %H:%M:%S", errors="coerce").astype("M8[s]")
    # Raise errors if timesteps are mispecified
    idx_mispecified = np.isnan(new_timesteps)
    mispecified_timesteps = timesteps[idx_mispecified].tolist()
    if len(mispecified_timesteps) > 0:
        msg = (
            f"The following timesteps are mispecified: {mispecified_timesteps}. Expecting the YYYY-mm-dd HH:MM:SS"
            " format."
        )
        log_error(logger, msg=msg, verbose=False)
        raise ValueError(msg)
    # Convert to numpy
    new_timesteps = new_timesteps.to_numpy()
    return new_timesteps


def check_timesteps(timesteps):
    """Check timesteps validity.

    It expects timesteps string in YYYY-mm-dd HH:MM:SS format with second accuracy.
    If timesteps is None, return None.
    """
    if isinstance(timesteps, type(None)):
        return None
    if isinstance(timesteps, str):
        timesteps = [timesteps]
    # Set as numpy array
    timesteps = np.array(timesteps)
    # If strings, check accordingly
    if is_numpy_array_string(timesteps):
        timesteps = _check_timesteps_string(timesteps)
    # If numpy datetime64, check accordingly
    elif is_numpy_array_datetime(timesteps):
        timesteps = _check_timestep_datetime_accuracy(timesteps, unit="s")
    else:
        raise TypeError("Unvalid timesteps input.")
    return timesteps


def _check_time_period_nested_list_format(time_periods):
    """Check that the time_periods is a list of list of length 2."""

    if not isinstance(time_periods, list):
        msg = "'time_periods' must be a list'"
        log_error(logger, msg=msg, verbose=False)
        raise TypeError(msg)

    for time_period in time_periods:
        if not isinstance(time_period, (list, np.ndarray)) or len(time_period) != 2:
            msg = "Every time period of time_periods must be a list of length 2."
            log_error(logger, msg=msg, verbose=False)
            raise ValueError(msg)
    return None


def check_time_periods(time_periods):
    """Check time_periods validity."""
    # Return None if None
    if isinstance(time_periods, type(None)):
        return None
    # Check time_period format
    _check_time_period_nested_list_format(time_periods)
    # Convert each time period to datetime64
    new_time_periods = []
    for time_period in time_periods:
        time_period = check_timesteps(timesteps=time_period)
        new_time_periods.append(time_period)
    # Check time period start occur before end
    for time_period in new_time_periods:
        if time_period[0] > time_period[1]:
            msg = f"The {time_period} time_period is unvalid. Start time occurs after end time."
            log_error(logger, msg=msg, verbose=False)
            raise ValueError(msg)
    return new_time_periods


def _get_issue_timesteps(issue_dict):
    """Get timesteps from issue dictionary."""
    timesteps = issue_dict.get("timesteps", None)
    # Check validity
    timesteps = check_timesteps(timesteps)
    # Sort
    timesteps.sort()
    return timesteps


def _get_issue_time_periods(issue_dict):
    """Get time_periods from issue dictionary."""
    time_periods = issue_dict.get("time_periods", None)
    time_periods = check_time_periods(time_periods)
    return time_periods


def check_issue_dict(issue_dict):
    """Check validity of the issue dictionary"""
    # Check is empty
    if len(issue_dict) == 0:
        return issue_dict
    # Check there are only timesteps and time_periods keys
    valid_keys = ["timesteps", "time_periods"]
    keys = list(issue_dict.keys())
    unvalid_keys = [k for k in keys if k not in valid_keys]
    if len(unvalid_keys) > 0:
        msg = f"Unvalid {unvalid_keys} keys. The issue YAML file accept only {valid_keys}"
        log_error(logger, msg=msg, verbose=False)
        raise ValueError(msg)

    # Check timesteps
    timesteps = _get_issue_timesteps(issue_dict)
    # Check time periods
    time_periods = _get_issue_time_periods(issue_dict)
    # Recreate issue dict
    issue_dict["timesteps"] = timesteps
    issue_dict["time_periods"] = time_periods

    return issue_dict


def check_issue_file(fpath: str) -> None:
    """Check issue YAML file validity.

    Parameters
    ----------
    fpath : str
        Issue YAML file path.

    """
    issue_dict = load_yaml_without_date_parsing(fpath)
    issue_dict = check_issue_dict(issue_dict)
    return None


####--------------------------------------------------------------------------.
#### Writer


def _write_issue_docs(f):
    """Provide template for issue.yml"""
    f.write("""# This file is used to store timesteps/time periods with wrong/corrupted observation.
# The specified timesteps are dropped during the L0 processing.
# The time format used is the isoformat : YYYY-mm-dd HH:MM:SS.
# The 'timesteps' key enable to specify the list of timesteps to be discarded.
# The 'time_period' key enable to specify the time periods to be dropped.
# Example:
#
# timesteps:
# - 2018-12-07 14:15:00
# - 2018-12-07 14:17:00
# - 2018-12-07 14:19:00
# - 2018-12-07 14:25:00
# time_period:
# - ['2018-08-01 12:00:00', '2018-08-01 14:00:00']
# - ['2018-08-01 15:44:30', '2018-08-01 15:59:31']
# - ['2018-08-02 12:44:30', '2018-08-02 12:59:31'] \n
""")
    return None


def _write_issue(fpath: str, timesteps: list = None, time_periods: list = None) -> None:
    """Write the issue YAML file.

    Parameters
    ----------
    fpath : str
        Filepath of the issue YAML to write.
    timesteps : list, optional
        List of timesteps (to be dropped in L0 processing).
        The default is None.
    time_periods : list, optional
        A list of time periods (to be dropped in L0 processing).
        The default is None.
    """
    # Preprocess timesteps and time_periods (to plain list of strings)
    if timesteps is not None:
        timesteps = timesteps.astype(str).tolist()

    if time_periods is not None:
        new_periods = []
        for i, time_period in enumerate(time_periods):
            new_periods.append(time_period.astype(str).tolist())
        time_periods = new_periods

    # Write the issue YAML file
    logger.info(f"Creating issue YAML file at {fpath}")
    with open(fpath, "w") as f:
        # Write the docs for the issue.yml
        _write_issue_docs(f)

        # Write timesteps if provided
        if timesteps is not None:
            timesteps_dict = {"timesteps": timesteps}
            yaml.dump(timesteps_dict, f, default_flow_style=False)

        # Write time_periods if provided
        if time_periods is not None:
            time_periods_dict = {"time_periods": time_periods}
            yaml.dump(time_periods_dict, f, default_flow_style=None)
    return None


def write_issue_dict(fpath: str, issue_dict: dict) -> None:
    """Write the issue YAML file.

    Parameters
    ----------
    fpath : str
        Filepath of the issue YAML to write.
    issue_dict : dict
        Issue dictionary
    """
    _write_issue(
        fpath=fpath,
        timesteps=issue_dict.get("timesteps", None),
        time_periods=issue_dict.get("time_periods", None),
    )


def write_default_issue(fpath: str) -> None:
    """Write an empty issue YAML file.

    Parameters
    ----------
    fpath : str
        Filepath of the issue YAML to write.
    """
    _write_issue(fpath=fpath)
    return None


####--------------------------------------------------------------------------.
#### Reader


class NoDatesSafeLoader(yaml.SafeLoader):
    @classmethod
    def remove_implicit_resolver(cls, tag_to_remove):
        """
        Remove implicit resolvers for a particular tag

        Takes care not to modify resolvers in super classes.

        We want to load datetimes as strings, not dates, because we
        go on to serialise as json which doesn't have the advanced types
        of yaml, and leads to incompatibilities down the track.
        """
        if "yaml_implicit_resolvers" not in cls.__dict__:
            cls.yaml_implicit_resolvers = cls.yaml_implicit_resolvers.copy()

        for first_letter, mappings in cls.yaml_implicit_resolvers.items():
            cls.yaml_implicit_resolvers[first_letter] = [
                (tag, regexp) for tag, regexp in mappings if tag != tag_to_remove
            ]


def load_yaml_without_date_parsing(filepath):
    "Read a YAML file without converting automatically date string to datetime."
    NoDatesSafeLoader.remove_implicit_resolver("tag:yaml.org,2002:timestamp")
    with open(filepath) as f:
        dictionary = yaml.load(f, Loader=NoDatesSafeLoader)
    # Return empty dictionary if nothing to be read in the file
    if isinstance(dictionary, type(None)):
        dictionary = {}
    return dictionary


def read_issue(raw_dir: str, station_name: str) -> dict:
    """Read YAML issue file.

    Parameters
    ----------
    raw_dir : str
        Path of the campaign raw directory.
    station_name : int
        Station name.

    Returns
    -------
    dict
        Issue dictionary.
    """

    issue_fpath = os.path.join(raw_dir, "issue", station_name + ".yml")
    issue_dict = load_yaml_without_date_parsing(issue_fpath)
    issue_dict = check_issue_dict(issue_dict)
    return issue_dict


def read_issue_file(fpath: str) -> dict:
    """Read YAML issue file.

    Parameters
    ----------
    fpath : str
        Filepath of the issue YAML.

    Returns
    -------
    dict
        Issue dictionary.
    """
    issue_dict = load_yaml_without_date_parsing(fpath)
    issue_dict = check_issue_dict(issue_dict)
    return issue_dict


####--------------------------------------------------------------------------.

1	#!/usr/bin/env python3
2
3	# -----------------------------------------------------------------------------.
4	# Copyright (c) 2021-2022 DISDRODB developers
5	#
6	# This program is free software: you can redistribute it and/or modify
7	# it under the terms of the GNU General Public License as published by
8	# the Free Software Foundation, either version 3 of the License, or
9	# (at your option) any later version.
10	#
11	# This program is distributed in the hope that it will be useful,
12	# but WITHOUT ANY WARRANTY; without even the implied warranty of
13	# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
14	# GNU General Public License for more details.
15	#
16	# You should have received a copy of the GNU General Public License
17	# along with this program. If not, see <http://www.gnu.org/licenses/>.
18	# -----------------------------------------------------------------------------.
19
20	import logging	1✔
21	import os	1✔
22
23	import numpy as np	1✔
24	import pandas as pd	1✔
25	import yaml	1✔
26
27	from disdrodb.utils.logger import log_error	1✔
28
29	logger = logging.getLogger(__name__)	1✔
30
31	####--------------------------------------------------------------------------.
32	#### Checks
33
34
35	def is_numpy_array_string(arr):	1✔
36	"""Check if the numpy array contains strings
37
38	Parameters
39	----------
40	arr : numpy array
41	Numpy array to check.
42	"""
43
44	dtype = arr.dtype.type	1✔
45	return dtype == np.str_ or dtype == np.unicode_	1✔
46
47
48	def is_numpy_array_datetime(arr):	1✔
49	"""Check if the numpy array contains datetime64
50
51	Parameters
52	----------
53	arr : numpy array
54	Numpy array to check.
55
56	Returns
57	-------
58	numpy array
59	Numpy array checked.
60	"""
61	return arr.dtype.type == np.datetime64	1✔
62
63
64	def _check_timestep_datetime_accuracy(timesteps, unit="s"):	1✔
65	"""Check the accuracy of the numpy datetime array.
66
67	Parameters
68	----------
69	timesteps : numpy array
70	Numpy array to check.
71	unit : str, optional
72	Unit, by default "s"
73
74	Returns
75	-------
76	numpy array
77	Numpy array checked.
78
79	Raises
80	------
81	ValueError
82	"""
83	if not timesteps.dtype == f"<M8[{unit}]":	1✔
84	msg = f"The timesteps does not have datetime64 {unit} accuracy."	1✔
85	log_error(logger, msg=msg, verbose=False)	1✔
86	raise ValueError(msg)	1✔
87	return timesteps	1✔
88
89
90	def _check_timestep_string_second_accuracy(timesteps, n=19):	1✔
91	"""Check the timesteps string are provided with second accuracy.
92
93	Note: it assumes the YYYY-mm-dd HH:MM:SS format
94	"""
95	n_characters = np.char.str_len(timesteps)	1✔
96	mispecified_timesteps = timesteps[n_characters != 19]	1✔
97	if len(mispecified_timesteps) > 0:	1✔
98	msg = (	1✔
99	f"The following timesteps are mispecified: {mispecified_timesteps}. Expecting the YYYY-mm-dd HH:MM:SS"
100	" format."
101	)
102	log_error(logger, msg=msg, verbose=False)	1✔
103	raise ValueError(msg)	1✔
104	return timesteps	1✔
105
106
107	def _check_timesteps_string(timesteps):	1✔
108	"""Check timesteps string validity.
109
110	It expects a list of timesteps strings in YYYY-mm-dd HH:MM:SS format with second accuracy.
111	"""
112	timesteps = np.asarray(timesteps)	1✔
113	timesteps = _check_timestep_string_second_accuracy(timesteps)	1✔
114	# Reformat as datetime64 with second accuracy
115	new_timesteps = pd.to_datetime(timesteps, format="%Y-%m-%d %H:%M:%S", errors="coerce").astype("M8[s]")	1✔
116	# Raise errors if timesteps are mispecified
117	idx_mispecified = np.isnan(new_timesteps)	1✔
118	mispecified_timesteps = timesteps[idx_mispecified].tolist()	1✔
119	if len(mispecified_timesteps) > 0:	1✔
120	msg = (	×
121	f"The following timesteps are mispecified: {mispecified_timesteps}. Expecting the YYYY-mm-dd HH:MM:SS"
122	" format."
123	)
124	log_error(logger, msg=msg, verbose=False)	×
125	raise ValueError(msg)	×
126	# Convert to numpy
127	new_timesteps = new_timesteps.to_numpy()	1✔
128	return new_timesteps	1✔
129
130
131	def check_timesteps(timesteps):	1✔
132	"""Check timesteps validity.
133
134	It expects timesteps string in YYYY-mm-dd HH:MM:SS format with second accuracy.
135	If timesteps is None, return None.
136	"""
137	if isinstance(timesteps, type(None)):	1✔
138	return None	1✔
139	if isinstance(timesteps, str):	1✔
140	timesteps = [timesteps]	1✔
141	# Set as numpy array
142	timesteps = np.array(timesteps)	1✔
143	# If strings, check accordingly
144	if is_numpy_array_string(timesteps):	1✔
145	timesteps = _check_timesteps_string(timesteps)	1✔
146	# If numpy datetime64, check accordingly
147	elif is_numpy_array_datetime(timesteps):	1✔
148	timesteps = _check_timestep_datetime_accuracy(timesteps, unit="s")	1✔
149	else:
150	raise TypeError("Unvalid timesteps input.")	1✔
151	return timesteps	1✔
152
153
154	def _check_time_period_nested_list_format(time_periods):	1✔
155	"""Check that the time_periods is a list of list of length 2."""
156
157	if not isinstance(time_periods, list):	1✔
158	msg = "'time_periods' must be a list'"	1✔
159	log_error(logger, msg=msg, verbose=False)	1✔
160	raise TypeError(msg)	1✔
161
162	for time_period in time_periods:	1✔
163	if not isinstance(time_period, (list, np.ndarray)) or len(time_period) != 2:	1✔
164	msg = "Every time period of time_periods must be a list of length 2."	1✔
165	log_error(logger, msg=msg, verbose=False)	1✔
166	raise ValueError(msg)	1✔
167	return None	1✔
168
169
170	def check_time_periods(time_periods):	1✔
171	"""Check time_periods validity."""
172	# Return None if None
173	if isinstance(time_periods, type(None)):	1✔
174	return None	1✔
175	# Check time_period format
176	_check_time_period_nested_list_format(time_periods)	1✔
177	# Convert each time period to datetime64
178	new_time_periods = []	1✔
179	for time_period in time_periods:	1✔
180	time_period = check_timesteps(timesteps=time_period)	1✔
181	new_time_periods.append(time_period)	1✔
182	# Check time period start occur before end
183	for time_period in new_time_periods:	1✔
184	if time_period[0] > time_period[1]:	1✔
185	msg = f"The {time_period} time_period is unvalid. Start time occurs after end time."	1✔
186	log_error(logger, msg=msg, verbose=False)	1✔
187	raise ValueError(msg)	1✔
188	return new_time_periods	1✔
189
190
191	def _get_issue_timesteps(issue_dict):	1✔
192	"""Get timesteps from issue dictionary."""
193	timesteps = issue_dict.get("timesteps", None)	1✔
194	# Check validity
195	timesteps = check_timesteps(timesteps)	1✔
196	# Sort
197	timesteps.sort()	1✔
198	return timesteps	1✔
199
200
201	def _get_issue_time_periods(issue_dict):	1✔
202	"""Get time_periods from issue dictionary."""
203	time_periods = issue_dict.get("time_periods", None)	1✔
204	time_periods = check_time_periods(time_periods)	1✔
205	return time_periods	1✔
206
207
208	def check_issue_dict(issue_dict):	1✔
209	"""Check validity of the issue dictionary"""
210	# Check is empty
211	if len(issue_dict) == 0:	1✔
212	return issue_dict	1✔
213	# Check there are only timesteps and time_periods keys
214	valid_keys = ["timesteps", "time_periods"]	1✔
215	keys = list(issue_dict.keys())	1✔
216	unvalid_keys = [k for k in keys if k not in valid_keys]	1✔
217	if len(unvalid_keys) > 0:	1✔
218	msg = f"Unvalid {unvalid_keys} keys. The issue YAML file accept only {valid_keys}"	1✔
219	log_error(logger, msg=msg, verbose=False)	1✔
220	raise ValueError(msg)	1✔
221
222	# Check timesteps
223	timesteps = _get_issue_timesteps(issue_dict)	1✔
224	# Check time periods
225	time_periods = _get_issue_time_periods(issue_dict)	1✔
226	# Recreate issue dict
227	issue_dict["timesteps"] = timesteps	1✔
228	issue_dict["time_periods"] = time_periods	1✔
229
230	return issue_dict	1✔
231
232
233	def check_issue_file(fpath: str) -> None:	1✔
234	"""Check issue YAML file validity.
235
236	Parameters
237	----------
238	fpath : str
239	Issue YAML file path.
240
241	"""
242	issue_dict = load_yaml_without_date_parsing(fpath)	1✔
243	issue_dict = check_issue_dict(issue_dict)	1✔
244	return None	1✔
245
246
247	####--------------------------------------------------------------------------.
248	#### Writer
249
250
251	def _write_issue_docs(f):	1✔
252	"""Provide template for issue.yml"""
253	f.write("""# This file is used to store timesteps/time periods with wrong/corrupted observation.	1✔
254	# The specified timesteps are dropped during the L0 processing.
255	# The time format used is the isoformat : YYYY-mm-dd HH:MM:SS.
256	# The 'timesteps' key enable to specify the list of timesteps to be discarded.
257	# The 'time_period' key enable to specify the time periods to be dropped.
258	# Example:
259	#
260	# timesteps:
261	# - 2018-12-07 14:15:00
262	# - 2018-12-07 14:17:00
263	# - 2018-12-07 14:19:00
264	# - 2018-12-07 14:25:00
265	# time_period:
266	# - ['2018-08-01 12:00:00', '2018-08-01 14:00:00']
267	# - ['2018-08-01 15:44:30', '2018-08-01 15:59:31']
268	# - ['2018-08-02 12:44:30', '2018-08-02 12:59:31'] \n
269	""")
270	return None	1✔
271
272
273	def _write_issue(fpath: str, timesteps: list = None, time_periods: list = None) -> None:	1✔
274	"""Write the issue YAML file.
275
276	Parameters
277	----------
278	fpath : str
279	Filepath of the issue YAML to write.
280	timesteps : list, optional
281	List of timesteps (to be dropped in L0 processing).
282	The default is None.
283	time_periods : list, optional
284	A list of time periods (to be dropped in L0 processing).
285	The default is None.
286	"""
287	# Preprocess timesteps and time_periods (to plain list of strings)
288	if timesteps is not None:	1✔
289	timesteps = timesteps.astype(str).tolist()	1✔
290
291	if time_periods is not None:	1✔
292	new_periods = []	1✔
293	for i, time_period in enumerate(time_periods):	1✔
294	new_periods.append(time_period.astype(str).tolist())	1✔
295	time_periods = new_periods	1✔
296
297	# Write the issue YAML file
298	logger.info(f"Creating issue YAML file at {fpath}")	1✔
299	with open(fpath, "w") as f:	1✔
300	# Write the docs for the issue.yml
301	_write_issue_docs(f)	1✔
302
303	# Write timesteps if provided
304	if timesteps is not None:	1✔
305	timesteps_dict = {"timesteps": timesteps}	1✔
306	yaml.dump(timesteps_dict, f, default_flow_style=False)	1✔
307
308	# Write time_periods if provided
309	if time_periods is not None:	1✔
310	time_periods_dict = {"time_periods": time_periods}	1✔
311	yaml.dump(time_periods_dict, f, default_flow_style=None)	1✔
312	return None	1✔
313
314
315	def write_issue_dict(fpath: str, issue_dict: dict) -> None:	1✔
316	"""Write the issue YAML file.
317
318	Parameters
319	----------
320	fpath : str
321	Filepath of the issue YAML to write.
322	issue_dict : dict
323	Issue dictionary
324	"""
UNCOV 325	_write_issue(	×
326	fpath=fpath,
327	timesteps=issue_dict.get("timesteps", None),
328	time_periods=issue_dict.get("time_periods", None),
329	)
330
331
332	def write_default_issue(fpath: str) -> None:	1✔
333	"""Write an empty issue YAML file.
334
335	Parameters
336	----------
337	fpath : str
338	Filepath of the issue YAML to write.
339	"""
340	_write_issue(fpath=fpath)	1✔
341	return None	1✔
342
343
344	####--------------------------------------------------------------------------.
345	#### Reader
346
347
348	class NoDatesSafeLoader(yaml.SafeLoader):	1✔
349	@classmethod	1✔
350	def remove_implicit_resolver(cls, tag_to_remove):	1✔
351	"""
352	Remove implicit resolvers for a particular tag
353
354	Takes care not to modify resolvers in super classes.
355
356	We want to load datetimes as strings, not dates, because we
357	go on to serialise as json which doesn't have the advanced types
358	of yaml, and leads to incompatibilities down the track.
359	"""
360	if "yaml_implicit_resolvers" not in cls.__dict__:	1✔
361	cls.yaml_implicit_resolvers = cls.yaml_implicit_resolvers.copy()	1✔
362
363	for first_letter, mappings in cls.yaml_implicit_resolvers.items():	1✔
364	cls.yaml_implicit_resolvers[first_letter] = [	1✔
365	(tag, regexp) for tag, regexp in mappings if tag != tag_to_remove
366	]
367
368
369	def load_yaml_without_date_parsing(filepath):	1✔
370	"Read a YAML file without converting automatically date string to datetime."
371	NoDatesSafeLoader.remove_implicit_resolver("tag:yaml.org,2002:timestamp")	1✔
372	with open(filepath) as f:	1✔
373	dictionary = yaml.load(f, Loader=NoDatesSafeLoader)	1✔
374	# Return empty dictionary if nothing to be read in the file
375	if isinstance(dictionary, type(None)):	1✔
376	dictionary = {}	1✔
377	return dictionary	1✔
378
379
380	def read_issue(raw_dir: str, station_name: str) -> dict:	1✔
381	"""Read YAML issue file.
382
383	Parameters
384	----------
385	raw_dir : str
386	Path of the campaign raw directory.
387	station_name : int
388	Station name.
389
390	Returns
391	-------
392	dict
393	Issue dictionary.
394	"""
395
396	issue_fpath = os.path.join(raw_dir, "issue", station_name + ".yml")	1✔
397	issue_dict = load_yaml_without_date_parsing(issue_fpath)	1✔
398	issue_dict = check_issue_dict(issue_dict)	1✔
399	return issue_dict	1✔
400
401
402	def read_issue_file(fpath: str) -> dict:	1✔
403	"""Read YAML issue file.
404
405	Parameters
406	----------
407	fpath : str
408	Filepath of the issue YAML.
409
410	Returns
411	-------
412	dict
413	Issue dictionary.
414	"""
415	issue_dict = load_yaml_without_date_parsing(fpath)	1✔
416	issue_dict = check_issue_dict(issue_dict)	1✔
417	return issue_dict	1✔
418
419
420	####--------------------------------------------------------------------------.

ltelab / disdrodb / 5247671608

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