b4dd4a1f-7b07-484b-bfa2-363b6b90e3f8

Committed 16 Nov 2023 07:01AM UTC coverage: 51.354% (+0.5%) from 50.87%

Build # b4dd4a1f-7b07-484b-bfa2-363b6b90e3f8

Build Type

push

circleci

Committed by

web-flow

Commit Message

Merge pull request #609 from jhkennedy/update_unit_tests

Update unit tests

Run Details

43 of 48 new or added lines in 5 files covered. (89.58%)

468 existing lines in 12 files now uncovered.

3090 of 6017 relevant lines covered (51.35%)

0.51 hits per line

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

61.99

/tools/RAiDER/cli/validators.py

from argparse import Action, ArgumentError, ArgumentTypeError

import importlib
import itertools
import os
import re

import pandas as pd
import numpy as np

from datetime import time, timedelta, datetime, date
from textwrap import dedent
from time import strptime

from RAiDER.llreader import BoundingBox, Geocube, RasterRDR, StationFile, GeocodedFile, Geocube
from RAiDER.losreader import Zenith, Conventional
from RAiDER.utilFcns import rio_extents, rio_profile
from RAiDER.logger import logger

_BUFFER_SIZE = 0.2 # default buffer size in lat/lon degrees

def enforce_wm(value, aoi):
    model = value.upper().replace("-", "")
    try:
        _, model_obj = modelName2Module(model)
    except ModuleNotFoundError:
        raise NotImplementedError(
            dedent('''
                Model {} is not yet fully implemented,
                please contribute!
                '''.format(model))
        )

    ## check the user requsted bounding box is within the weather model domain
    modObj = model_obj().checkValidBounds(aoi.bounds())

    return modObj


def get_los(args):
    if args.get('orbit_file'):
        if args.get('ray_trace'):
            from RAiDER.losreader import Raytracing
            los = Raytracing(args.orbit_file)
        else:
            los = Conventional(args.orbit_file)
    elif args.get('los_file'):
        if args.ray_trace:
            from RAiDER.losreader import Raytracing
            los = Raytracing(args.los_file, args.los_convention)
        else:
            los = Conventional(args.los_file, args.los_convention)

    elif args.get('los_cube'):
        raise NotImplementedError('LOS_cube is not yet implemented')
#        if args.ray_trace:
#            los = Raytracing(args.los_cube)
#        else:
#            los = Conventional(args.los_cube)
    else:
        los = Zenith()

    return los


def get_heights(args, out, station_file, bounding_box=None):
    '''
    Parse the Height info and download a DEM if needed
    '''
    dem_path = out

    out = {
            'dem': args.get('dem'),
            'height_file_rdr': None,
            'height_levels': None,
        }

    if args.get('dem'):
        if (station_file is not None):
            if 'Hgt_m' not in pd.read_csv(station_file):
                out['dem'] = os.path.join(dem_path, 'GLO30.dem')
        elif os.path.exists(args.dem):
            out['dem'] = args.dem
            # crop the DEM
            if bounding_box is not None:
                dem_bounds = rio_extents(rio_profile(args.dem))
                lats = dem_bounds[:2]
                lons = dem_bounds[2:]
                if isOutside(
                    bounding_box,
                    getBufferedExtent(
                        lats,
                        lons,
                        buf=_BUFFER_SIZE,
                    )
                ):
                    raise ValueError(
                                'Existing DEM does not cover the area of the input lat/lon '
                                'points; either move the DEM, delete it, or change the input '
                                'points.'
                            )
        else:
            pass # will download the dem later

    elif args.get('height_file_rdr'):
        out['height_file_rdr'] = args.height_file_rdr

    else:
        # download the DEM if needed
        out['dem'] = os.path.join(dem_path, 'GLO30.dem')

    if args.get('height_levels'):
        if isinstance(args.height_levels, str):
            l = re.findall('[-0-9]+', args.height_levels)
        else:
            l = args.height_levels

        out['height_levels'] = np.array([float(ll) for ll in l])
        if np.any(out['height_levels'] < 0):
            logger.warning('Weather model only extends to the surface topography; '
            'height levels below the topography will be interpolated from the surface '
            'and may be inaccurate.')

    return out


def get_query_region(args):
    '''
    Parse the query region from inputs
    '''
    # Get bounds from the inputs
    # make sure this is first
    if args.get('use_dem_latlon'):
        query = GeocodedFile(args.dem, is_dem=True)

    elif args.get('lat_file'):
        hgt_file = args.get('height_file_rdr') # only get it if exists
        dem_file = args.get('dem')
        query    = RasterRDR(args.lat_file, args.lon_file, hgt_file, dem_file)

    elif args.get('station_file'):
        query = StationFile(args.station_file)

    elif args.get('bounding_box'):
        bbox = enforce_bbox(args.bounding_box)
        if (np.min(bbox[0]) < -90) | (np.max(bbox[1]) > 90):
            raise ValueError('Lats are out of N/S bounds; are your lat/lon coordinates switched? Should be SNWE')
        query = BoundingBox(bbox)

    elif args.get('geocoded_file'):
        gfile  = os.path.basename(args.geocoded_file).upper()
        if (gfile.startswith('SRTM') or gfile.startswith('GLO')):
            logger.debug('Using user DEM: %s', gfile)
            is_dem = True
        else:
            is_dem = False

        query  = GeocodedFile(args.geocoded_file, is_dem=is_dem)

    ## untested
    elif args.get('geo_cube'):
        query = Geocube(args.geo_cube)

    else:
        # TODO: Need to incorporate the cube
        raise ValueError('No valid query points or bounding box found in the configuration file')

    return query


def enforce_bbox(bbox):
    """
    Enforce a valid bounding box
    """
    if isinstance(bbox, str):
        bbox = [float(d) for d in bbox.strip().split()]
    else:
        bbox = [float(d) for d in bbox]

    # Check the bbox
    if len(bbox) != 4:
        raise ValueError("bounding box must have 4 elements!")
    S, N, W, E = bbox

    if N <= S or E <= W:
        raise ValueError('Bounding box has no size; make sure you use "S N W E"')

    for sn in (S, N):
        if sn < -90 or sn > 90:
            raise ValueError('Lats are out of S/N bounds (-90 to 90).')

    for we in (W, E):
        if we < -180 or we > 180:
            raise ValueError('Lons are out of W/E bounds (-180 to 180); Lons in the format of (0 to 360) are not supported.')

    return bbox


def parse_dates(arg_dict):
    '''
    Determine the requested dates from the input parameters
    '''

    if arg_dict.get('date_list'):
        l = arg_dict['date_list']
        if isinstance(l, str):
            l = re.findall('[0-9]+', l)
        elif isinstance(l, int):
            l = [l]
        L = [enforce_valid_dates(d) for d in l]

    else:
        try:
            start = arg_dict['date_start']
        except KeyError:
            raise ValueError('Inputs must include either date_list or date_start')
        start = enforce_valid_dates(start)

        if arg_dict.get('date_end'):
            end = arg_dict['date_end']
            end = enforce_valid_dates(end)
        else:
           end = start

        if arg_dict.get('date_step'):
            step = int(arg_dict['date_step'])
        else:
            step = 1

        L = [start + timedelta(days=step) for step in range(0, (end - start).days + 1, step)]

    return L


def enforce_valid_dates(arg):
    """
    Parse a date from a string in pseudo-ISO 8601 format.
    """
    year_formats = (
        '%Y-%m-%d',
        '%Y%m%d',
        '%d',
        '%j',
    )

    for yf in year_formats:
        try:
            return datetime.strptime(str(arg), yf)
        except ValueError:
            pass


    raise ValueError(
        'Unable to coerce {} to a date. Try %Y-%m-%d'.format(arg)
    )


def enforce_time(arg_dict):
    '''
    Parse an input time (required to be ISO 8601)
    '''
    try:
        arg_dict['time'] = convert_time(arg_dict['time'])
    except KeyError:
        raise ValueError('You must specify a "time" in the input config file')

    if 'end_time' in arg_dict.keys():
        arg_dict['end_time'] = convert_time(arg_dict['end_time'])
    return arg_dict


def convert_time(inp):
    time_formats = (
        '',
        'T%H:%M:%S.%f',
        'T%H%M%S.%f',
        '%H%M%S.%f',
        'T%H:%M:%S',
        '%H:%M:%S',
        'T%H%M%S',
        '%H%M%S',
        'T%H:%M',
        'T%H%M',
        '%H:%M',
        'T%H',
    )
    timezone_formats = (
        '',
        'Z',
        '%z',
    )
    all_formats = map(
        ''.join,
        itertools.product(time_formats, timezone_formats)
    )

    for tf in all_formats:
        try:
            return time(*strptime(inp, tf)[3:6])
        except ValueError:
            pass

    raise ValueError(
                'Unable to coerce {} to a time.'+
                'Try T%H:%M:%S'.format(inp)
        )


def modelName2Module(model_name):
    """Turn an arbitrary string into a module name.
    Takes as input a model name, which hopefully looks like ERA-I, and
    converts it to a module name, which will look like erai. I doesn't
    always produce a valid module name, but that's not the goal. The
    goal is just to handle common cases.
    Inputs:
       model_name  - Name of an allowed weather model (e.g., 'era-5')
    Outputs:
       module_name - Name of the module
       wmObject    - callable, weather model object
    """
    module_name = 'RAiDER.models.' + model_name.lower().replace('-', '')
    model_module = importlib.import_module(module_name)
    wmObject = getattr(model_module, model_name.upper().replace('-', ''))
    return module_name, wmObject


def getBufferedExtent(lats, lons=None, buf=0.):
    '''
    get the bounding box around a set of lats/lons
    '''
    if lons is None:
        lats, lons = lats[..., 0], lons[..., 1]

    try:
        if (lats.size == 1) & (lons.size == 1):
            out = [lats - buf, lats + buf, lons - buf, lons + buf]
        elif (lats.size > 1) & (lons.size > 1):
            out = [np.nanmin(lats), np.nanmax(lats), np.nanmin(lons), np.nanmax(lons)]
        elif lats.size == 1:
            out = [lats - buf, lats + buf, np.nanmin(lons), np.nanmax(lons)]
        elif lons.size == 1:
            out = [np.nanmin(lats), np.nanmax(lats), lons - buf, lons + buf]
    except AttributeError:
        if (isinstance(lats, tuple) or isinstance(lats, list)) and len(lats) == 2:
            out = [min(lats) - buf, max(lats) + buf, min(lons) - buf, max(lons) + buf]
    except Exception as e:
        raise RuntimeError('Not a valid lat/lon shape or variable')

    return np.array(out)


def isOutside(extent1, extent2):
    '''
    Determine whether any of extent1  lies outside extent2
    extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon]
    Equal extents are considered "inside"
    '''
    t1 = extent1[0] < extent2[0]
    t2 = extent1[1] > extent2[1]
    t3 = extent1[2] < extent2[2]
    t4 = extent1[3] > extent2[3]
    if np.any([t1, t2, t3, t4]):
        return True
    return False


def isInside(extent1, extent2):
    '''
    Determine whether all of extent1 lies inside extent2
    extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon].
    Equal extents are considered "inside"
    '''
    t1 = extent1[0] <= extent2[0]
    t2 = extent1[1] >= extent2[1]
    t3 = extent1[2] <= extent2[2]
    t4 = extent1[3] >= extent2[3]
    if np.all([t1, t2, t3, t4]):
        return True
    return False


## below are for downloadGNSSDelays
def date_type(arg):
    """
    Parse a date from a string in pseudo-ISO 8601 format.
    """
    year_formats = (
        '%Y-%m-%d',
        '%Y%m%d',
        '%d',
        '%j',
    )

    for yf in year_formats:
        try:
            return date(*strptime(arg, yf)[0:3])
        except ValueError:
            pass

    raise ArgumentTypeError(
        'Unable to coerce {} to a date. Try %Y-%m-%d'.format(arg)
    )


class MappingType(object):
    """
    A type that maps arguments to constants.

    # Example
    ```
    mapping = MappingType(foo=42, bar="baz").default(None)
    assert mapping("foo") == 42
    assert mapping("bar") == "baz"
    assert mapping("hello") is None
    ```
    """
    UNSET = object()

    def __init__(self, **kwargs):
        self.mapping = kwargs
        self._default = self.UNSET

    def default(self, default):
        """Set a default value if no mapping is found"""
        self._default = default
        return self

    def __call__(self, arg):
        if arg in self.mapping:
            return self.mapping[arg]

        if self._default is self.UNSET:
            raise KeyError(
                "Invalid choice '{}', must be one of {}".format(
                    arg, list(self.mapping.keys())
                )
            )

        return self._default


class IntegerType(object):
    """
    A type that converts arguments to integers.

    # Example
    ```
    integer = IntegerType(0, 100)
    assert integer("0") == 0
    assert integer("100") == 100
    integer("-10")  # Raises exception
    ```
    """

    def __init__(self, lo=None, hi=None):
        self.lo = lo
        self.hi = hi

    def __call__(self, arg):
        integer = int(arg)

        if self.lo is not None and integer < self.lo:
            raise ArgumentTypeError("Must be greater than {}".format(self.lo))
        if self.hi is not None and integer > self.hi:
            raise ArgumentTypeError("Must be less than {}".format(self.hi))

        return integer


class IntegerMappingType(MappingType, IntegerType):
    """
    An integer type that converts non-integer types through a mapping.

    # Example
    ```
    integer = IntegerMappingType(0, 100, random=42)
    assert integer("0") == 0
    assert integer("100") == 100
    assert integer("random") == 42
    ```
    """

    def __init__(self, lo=None, hi=None, mapping={}, **kwargs):
        IntegerType.__init__(self, lo, hi)
        kwargs.update(mapping)
        MappingType.__init__(self, **kwargs)

    def __call__(self, arg):
        try:
            return IntegerType.__call__(self, arg)
        except ValueError:
            return MappingType.__call__(self, arg)


class DateListAction(Action):
    """An Action that parses and stores a list of dates"""

    def __init__(
        self,
        option_strings,
        dest,
        nargs=None,
        const=None,
        default=None,
        type=None,
        choices=None,
        required=False,
        help=None,
        metavar=None
    ):
        if type is not date_type:
            raise ValueError("type must be `date_type`!")

        super().__init__(
            option_strings=option_strings,
            dest=dest,
            nargs=nargs,
            const=const,
            default=default,
            type=type,
            choices=choices,
            required=required,
            help=help,
            metavar=metavar
        )

    def __call__(self, parser, namespace, values, option_string=None):
        if len(values) > 3 or not values:
            raise ArgumentError(self, "Only 1, 2 dates, or 2 dates and interval may be supplied")

        if len(values) == 2:
            start, end = values
            values = [start + timedelta(days=k) for k in range(0, (end - start).days + 1, 1)]
        elif len(values) == 3:
            start, end, stepsize = values

            if not isinstance(stepsize.day, int):
                raise ArgumentError(self, "The stepsize should be in integer days")

            new_year = date(year=stepsize.year, month=1, day=1)
            stepsize = (stepsize - new_year).days + 1

            values = [start + timedelta(days=k)
                      for k in range(0, (end - start).days + 1, stepsize)]

        setattr(namespace, self.dest, values)


class BBoxAction(Action):
    """An Action that parses and stores a valid bounding box"""

    def __init__(
        self,
        option_strings,
        dest,
        nargs=None,
        const=None,
        default=None,
        type=None,
        choices=None,
        required=False,
        help=None,
        metavar=None
    ):
        if nargs != 4:
            raise ValueError("nargs must be 4!")

        super().__init__(
            option_strings=option_strings,
            dest=dest,
            nargs=nargs,
            const=const,
            default=default,
            type=type,
            choices=choices,
            required=required,
            help=help,
            metavar=metavar
        )

    def __call__(self, parser, namespace, values, option_string=None):
        S, N, W, E = values

        if N <= S or E <= W:
            raise ArgumentError(self, 'Bounding box has no size; make sure you use "S N W E"')

        for sn in (S, N):
            if sn < -90 or sn > 90:
                raise ArgumentError(self, 'Lats are out of S/N bounds (-90 to 90).')

        for we in (W, E):
            if we < -180 or we > 180:
                raise ArgumentError(self, 'Lons are out of W/E bounds (-180 to 180); Lons in the format of (0 to 360) are not supported.')

        setattr(namespace, self.dest, values)

1	from argparse import Action, ArgumentError, ArgumentTypeError	1✔
2
3	import importlib	1✔
4	import itertools	1✔
5	import os	1✔
6	import re	1✔
7
8	import pandas as pd	1✔
9	import numpy as np	1✔
10
11	from datetime import time, timedelta, datetime, date	1✔
12	from textwrap import dedent	1✔
13	from time import strptime	1✔
14
15	from RAiDER.llreader import BoundingBox, Geocube, RasterRDR, StationFile, GeocodedFile, Geocube	1✔
16	from RAiDER.losreader import Zenith, Conventional	1✔
17	from RAiDER.utilFcns import rio_extents, rio_profile	1✔
18	from RAiDER.logger import logger	1✔
19
20	_BUFFER_SIZE = 0.2 # default buffer size in lat/lon degrees	1✔
21
22	def enforce_wm(value, aoi):	1✔
23	model = value.upper().replace("-", "")	1✔
24	try:	1✔
25	_, model_obj = modelName2Module(model)	1✔
26	except ModuleNotFoundError:	1✔
27	raise NotImplementedError(	1✔
28	dedent('''
29	Model {} is not yet fully implemented,
30	please contribute!
31	'''.format(model))
32	)
33
34	## check the user requsted bounding box is within the weather model domain
35	modObj = model_obj().checkValidBounds(aoi.bounds())	1✔
36
37	return modObj	1✔
38
39
40	def get_los(args):	1✔
41	if args.get('orbit_file'):	1✔
42	if args.get('ray_trace'):	1✔
43	from RAiDER.losreader import Raytracing	1✔
44	los = Raytracing(args.orbit_file)	1✔
45	else:
46	los = Conventional(args.orbit_file)	1✔
47	elif args.get('los_file'):	1✔
48	if args.ray_trace:	1✔
49	from RAiDER.losreader import Raytracing	×
50	los = Raytracing(args.los_file, args.los_convention)	×
51	else:
52	los = Conventional(args.los_file, args.los_convention)	1✔
53
54	elif args.get('los_cube'):	1✔
55	raise NotImplementedError('LOS_cube is not yet implemented')	×
56	# if args.ray_trace:
57	# los = Raytracing(args.los_cube)
58	# else:
59	# los = Conventional(args.los_cube)
60	else:
61	los = Zenith()	1✔
62
63	return los	1✔
64
65
66	def get_heights(args, out, station_file, bounding_box=None):	1✔
67	'''
68	Parse the Height info and download a DEM if needed
69	'''
70	dem_path = out	1✔
71
72	out = {	1✔
73	'dem': args.get('dem'),
74	'height_file_rdr': None,
75	'height_levels': None,
76	}
77
78	if args.get('dem'):	1✔
79	if (station_file is not None):	×
80	if 'Hgt_m' not in pd.read_csv(station_file):	×
81	out['dem'] = os.path.join(dem_path, 'GLO30.dem')	×
82	elif os.path.exists(args.dem):	×
83	out['dem'] = args.dem	×
84	# crop the DEM
85	if bounding_box is not None:	×
86	dem_bounds = rio_extents(rio_profile(args.dem))	×
87	lats = dem_bounds[:2]	×
88	lons = dem_bounds[2:]	×
89	if isOutside(	×
90	bounding_box,
91	getBufferedExtent(
92	lats,
93	lons,
94	buf=_BUFFER_SIZE,
95	)
96	):
97	raise ValueError(	×
98	'Existing DEM does not cover the area of the input lat/lon '
99	'points; either move the DEM, delete it, or change the input '
100	'points.'
101	)
102	else:
103	pass # will download the dem later	×
104
105	elif args.get('height_file_rdr'):	1✔
106	out['height_file_rdr'] = args.height_file_rdr	×
107
108	else:
109	# download the DEM if needed
110	out['dem'] = os.path.join(dem_path, 'GLO30.dem')	1✔
111
112	if args.get('height_levels'):	1✔
113	if isinstance(args.height_levels, str):	1✔
114	l = re.findall('[-0-9]+', args.height_levels)	1✔
115	else:
116	l = args.height_levels	1✔
117
118	out['height_levels'] = np.array([float(ll) for ll in l])	1✔
119	if np.any(out['height_levels'] < 0):	1✔
120	logger.warning('Weather model only extends to the surface topography; '	1✔
121	'height levels below the topography will be interpolated from the surface '
122	'and may be inaccurate.')
123
124	return out	1✔
125
126
127	def get_query_region(args):	1✔
128	'''
129	Parse the query region from inputs
130	'''
131	# Get bounds from the inputs
132	# make sure this is first
133	if args.get('use_dem_latlon'):	1✔
134	query = GeocodedFile(args.dem, is_dem=True)	×
135
136	elif args.get('lat_file'):	1✔
137	hgt_file = args.get('height_file_rdr') # only get it if exists	1✔
138	dem_file = args.get('dem')	1✔
139	query = RasterRDR(args.lat_file, args.lon_file, hgt_file, dem_file)	1✔
140
141	elif args.get('station_file'):	1✔
142	query = StationFile(args.station_file)	1✔
143
144	elif args.get('bounding_box'):	1✔
145	bbox = enforce_bbox(args.bounding_box)	1✔
146	if (np.min(bbox[0]) < -90) \| (np.max(bbox[1]) > 90):	1✔
147	raise ValueError('Lats are out of N/S bounds; are your lat/lon coordinates switched? Should be SNWE')	×
148	query = BoundingBox(bbox)	1✔
149
150	elif args.get('geocoded_file'):	×
151	gfile = os.path.basename(args.geocoded_file).upper()	×
152	if (gfile.startswith('SRTM') or gfile.startswith('GLO')):	×
153	logger.debug('Using user DEM: %s', gfile)	×
154	is_dem = True	×
155	else:
156	is_dem = False	×
157
158	query = GeocodedFile(args.geocoded_file, is_dem=is_dem)	×
159
160	## untested
161	elif args.get('geo_cube'):	×
162	query = Geocube(args.geo_cube)	×
163
164	else:
165	# TODO: Need to incorporate the cube
166	raise ValueError('No valid query points or bounding box found in the configuration file')	×
167
168	return query	1✔
169
170
171	def enforce_bbox(bbox):	1✔
172	"""
173	Enforce a valid bounding box
174	"""
175	if isinstance(bbox, str):	1✔
176	bbox = [float(d) for d in bbox.strip().split()]	1✔
177	else:
178	bbox = [float(d) for d in bbox]	1✔
179
180	# Check the bbox
181	if len(bbox) != 4:	1✔
UNCOV 182	raise ValueError("bounding box must have 4 elements!")	×
183	S, N, W, E = bbox	1✔
184
185	if N <= S or E <= W:	1✔
186	raise ValueError('Bounding box has no size; make sure you use "S N W E"')	1✔
187
188	for sn in (S, N):	1✔
189	if sn < -90 or sn > 90:	1✔
190	raise ValueError('Lats are out of S/N bounds (-90 to 90).')	1✔
191
192	for we in (W, E):	1✔
193	if we < -180 or we > 180:	1✔
194	raise ValueError('Lons are out of W/E bounds (-180 to 180); Lons in the format of (0 to 360) are not supported.')	1✔
195
196	return bbox	1✔
197
198
199	def parse_dates(arg_dict):	1✔
200	'''
201	Determine the requested dates from the input parameters
202	'''
203
204	if arg_dict.get('date_list'):	1✔
205	l = arg_dict['date_list']	1✔
206	if isinstance(l, str):	1✔
207	l = re.findall('[0-9]+', l)	1✔
208	elif isinstance(l, int):	1✔
UNCOV 209	l = [l]	×
210	L = [enforce_valid_dates(d) for d in l]	1✔
211
212	else:
213	try:	1✔
214	start = arg_dict['date_start']	1✔
UNCOV 215	except KeyError:	×
216	raise ValueError('Inputs must include either date_list or date_start')	×
217	start = enforce_valid_dates(start)	1✔
218
219	if arg_dict.get('date_end'):	1✔
220	end = arg_dict['date_end']	1✔
221	end = enforce_valid_dates(end)	1✔
222	else:
223	end = start	1✔
224
225	if arg_dict.get('date_step'):	1✔
226	step = int(arg_dict['date_step'])	1✔
227	else:
228	step = 1	1✔
229
230	L = [start + timedelta(days=step) for step in range(0, (end - start).days + 1, step)]	1✔
231
232	return L	1✔
233
234
235	def enforce_valid_dates(arg):	1✔
236	"""
237	Parse a date from a string in pseudo-ISO 8601 format.
238	"""
239	year_formats = (	1✔
240	'%Y-%m-%d',
241	'%Y%m%d',
242	'%d',
243	'%j',
244	)
245
246	for yf in year_formats:	1✔
247	try:	1✔
248	return datetime.strptime(str(arg), yf)	1✔
249	except ValueError:	1✔
250	pass	1✔
251
252
253	raise ValueError(	1✔
254	'Unable to coerce {} to a date. Try %Y-%m-%d'.format(arg)
255	)
256
257
258	def enforce_time(arg_dict):	1✔
259	'''
260	Parse an input time (required to be ISO 8601)
261	'''
262	try:	1✔
263	arg_dict['time'] = convert_time(arg_dict['time'])	1✔
UNCOV 264	except KeyError:	×
265	raise ValueError('You must specify a "time" in the input config file')	×
266
267	if 'end_time' in arg_dict.keys():	1✔
UNCOV 268	arg_dict['end_time'] = convert_time(arg_dict['end_time'])	×
269	return arg_dict	1✔
270
271
272	def convert_time(inp):	1✔
273	time_formats = (	1✔
274	'',
275	'T%H:%M:%S.%f',
276	'T%H%M%S.%f',
277	'%H%M%S.%f',
278	'T%H:%M:%S',
279	'%H:%M:%S',
280	'T%H%M%S',
281	'%H%M%S',
282	'T%H:%M',
283	'T%H%M',
284	'%H:%M',
285	'T%H',
286	)
287	timezone_formats = (	1✔
288	'',
289	'Z',
290	'%z',
291	)
292	all_formats = map(	1✔
293	''.join,
294	itertools.product(time_formats, timezone_formats)
295	)
296
297	for tf in all_formats:	1✔
298	try:	1✔
299	return time(*strptime(inp, tf)[3:6])	1✔
300	except ValueError:	1✔
301	pass	1✔
302
303	raise ValueError(	1✔
304	'Unable to coerce {} to a time.'+
305	'Try T%H:%M:%S'.format(inp)
306	)
307
308
309	def modelName2Module(model_name):	1✔
310	"""Turn an arbitrary string into a module name.
311	Takes as input a model name, which hopefully looks like ERA-I, and
312	converts it to a module name, which will look like erai. I doesn't
313	always produce a valid module name, but that's not the goal. The
314	goal is just to handle common cases.
315	Inputs:
316	model_name - Name of an allowed weather model (e.g., 'era-5')
317	Outputs:
318	module_name - Name of the module
319	wmObject - callable, weather model object
320	"""
321	module_name = 'RAiDER.models.' + model_name.lower().replace('-', '')	1✔
322	model_module = importlib.import_module(module_name)	1✔
323	wmObject = getattr(model_module, model_name.upper().replace('-', ''))	1✔
324	return module_name, wmObject	1✔
325
326
327	def getBufferedExtent(lats, lons=None, buf=0.):	1✔
328	'''
329	get the bounding box around a set of lats/lons
330	'''
331	if lons is None:	1✔
UNCOV 332	lats, lons = lats[..., 0], lons[..., 1]	×
333
334	try:	1✔
335	if (lats.size == 1) & (lons.size == 1):	1✔
UNCOV 336	out = [lats - buf, lats + buf, lons - buf, lons + buf]	×
337	elif (lats.size > 1) & (lons.size > 1):	1✔
338	out = [np.nanmin(lats), np.nanmax(lats), np.nanmin(lons), np.nanmax(lons)]	1✔
UNCOV 339	elif lats.size == 1:	×
340	out = [lats - buf, lats + buf, np.nanmin(lons), np.nanmax(lons)]	×
341	elif lons.size == 1:	×
342	out = [np.nanmin(lats), np.nanmax(lats), lons - buf, lons + buf]	×
343	except AttributeError:	1✔
344	if (isinstance(lats, tuple) or isinstance(lats, list)) and len(lats) == 2:	1✔
345	out = [min(lats) - buf, max(lats) + buf, min(lons) - buf, max(lons) + buf]	1✔
UNCOV 346	except Exception as e:	×
347	raise RuntimeError('Not a valid lat/lon shape or variable')	×
348
349	return np.array(out)	1✔
350
351
352	def isOutside(extent1, extent2):	1✔
353	'''
354	Determine whether any of extent1 lies outside extent2
355	extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon]
356	Equal extents are considered "inside"
357	'''
358	t1 = extent1[0] < extent2[0]	1✔
359	t2 = extent1[1] > extent2[1]	1✔
360	t3 = extent1[2] < extent2[2]	1✔
361	t4 = extent1[3] > extent2[3]	1✔
362	if np.any([t1, t2, t3, t4]):	1✔
363	return True	1✔
364	return False	1✔
365
366
367	def isInside(extent1, extent2):	1✔
368	'''
369	Determine whether all of extent1 lies inside extent2
370	extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon].
371	Equal extents are considered "inside"
372	'''
373	t1 = extent1[0] <= extent2[0]	1✔
374	t2 = extent1[1] >= extent2[1]	1✔
375	t3 = extent1[2] <= extent2[2]	1✔
376	t4 = extent1[3] >= extent2[3]	1✔
377	if np.all([t1, t2, t3, t4]):	1✔
378	return True	1✔
379	return False	1✔
380
381
382	## below are for downloadGNSSDelays
383	def date_type(arg):	1✔
384	"""
385	Parse a date from a string in pseudo-ISO 8601 format.
386	"""
UNCOV 387	year_formats = (	×
388	'%Y-%m-%d',
389	'%Y%m%d',
390	'%d',
391	'%j',
392	)
393
UNCOV 394	for yf in year_formats:	×
395	try:	×
396	return date(*strptime(arg, yf)[0:3])	×
397	except ValueError:	×
398	pass	×
399
UNCOV 400	raise ArgumentTypeError(	×
401	'Unable to coerce {} to a date. Try %Y-%m-%d'.format(arg)
402	)
403
404
405	class MappingType(object):	1✔
406	"""
407	A type that maps arguments to constants.
408
409	# Example
410	```
411	mapping = MappingType(foo=42, bar="baz").default(None)
412	assert mapping("foo") == 42
413	assert mapping("bar") == "baz"
414	assert mapping("hello") is None
415	```
416	"""
417	UNSET = object()	1✔
418
419	def __init__(self, **kwargs):	1✔
UNCOV 420	self.mapping = kwargs	×
421	self._default = self.UNSET	×
422
423	def default(self, default):	1✔
424	"""Set a default value if no mapping is found"""
UNCOV 425	self._default = default	×
426	return self	×
427
428	def __call__(self, arg):	1✔
UNCOV 429	if arg in self.mapping:	×
430	return self.mapping[arg]	×
431
UNCOV 432	if self._default is self.UNSET:	×
433	raise KeyError(	×
434	"Invalid choice '{}', must be one of {}".format(
435	arg, list(self.mapping.keys())
436	)
437	)
438
UNCOV 439	return self._default	×
440
441
442	class IntegerType(object):	1✔
443	"""
444	A type that converts arguments to integers.
445
446	# Example
447	```
448	integer = IntegerType(0, 100)
449	assert integer("0") == 0
450	assert integer("100") == 100
451	integer("-10") # Raises exception
452	```
453	"""
454
455	def __init__(self, lo=None, hi=None):	1✔
UNCOV 456	self.lo = lo	×
457	self.hi = hi	×
458
459	def __call__(self, arg):	1✔
UNCOV 460	integer = int(arg)	×
461
UNCOV 462	if self.lo is not None and integer < self.lo:	×
463	raise ArgumentTypeError("Must be greater than {}".format(self.lo))	×
464	if self.hi is not None and integer > self.hi:	×
465	raise ArgumentTypeError("Must be less than {}".format(self.hi))	×
466
UNCOV 467	return integer	×
468
469
470	class IntegerMappingType(MappingType, IntegerType):	1✔
471	"""
472	An integer type that converts non-integer types through a mapping.
473
474	# Example
475	```
476	integer = IntegerMappingType(0, 100, random=42)
477	assert integer("0") == 0
478	assert integer("100") == 100
479	assert integer("random") == 42
480	```
481	"""
482
483	def __init__(self, lo=None, hi=None, mapping={}, **kwargs):	1✔
UNCOV 484	IntegerType.__init__(self, lo, hi)	×
485	kwargs.update(mapping)	×
486	MappingType.__init__(self, **kwargs)	×
487
488	def __call__(self, arg):	1✔
UNCOV 489	try:	×
490	return IntegerType.__call__(self, arg)	×
491	except ValueError:	×
492	return MappingType.__call__(self, arg)	×
493
494
495	class DateListAction(Action):	1✔
496	"""An Action that parses and stores a list of dates"""
497
498	def __init__(	1✔
499	self,
500	option_strings,
501	dest,
502	nargs=None,
503	const=None,
504	default=None,
505	type=None,
506	choices=None,
507	required=False,
508	help=None,
509	metavar=None
510	):
UNCOV 511	if type is not date_type:	×
512	raise ValueError("type must be `date_type`!")	×
513
UNCOV 514	super().__init__(	×
515	option_strings=option_strings,
516	dest=dest,
517	nargs=nargs,
518	const=const,
519	default=default,
520	type=type,
521	choices=choices,
522	required=required,
523	help=help,
524	metavar=metavar
525	)
526
527	def __call__(self, parser, namespace, values, option_string=None):	1✔
UNCOV 528	if len(values) > 3 or not values:	×
529	raise ArgumentError(self, "Only 1, 2 dates, or 2 dates and interval may be supplied")	×
530
UNCOV 531	if len(values) == 2:	×
532	start, end = values	×
533	values = [start + timedelta(days=k) for k in range(0, (end - start).days + 1, 1)]	×
534	elif len(values) == 3:	×
535	start, end, stepsize = values	×
536
UNCOV 537	if not isinstance(stepsize.day, int):	×
538	raise ArgumentError(self, "The stepsize should be in integer days")	×
539
UNCOV 540	new_year = date(year=stepsize.year, month=1, day=1)	×
541	stepsize = (stepsize - new_year).days + 1	×
542
UNCOV 543	values = [start + timedelta(days=k)	×
544	for k in range(0, (end - start).days + 1, stepsize)]
545
UNCOV 546	setattr(namespace, self.dest, values)	×
547
548
549	class BBoxAction(Action):	1✔
550	"""An Action that parses and stores a valid bounding box"""
551
552	def __init__(	1✔
553	self,
554	option_strings,
555	dest,
556	nargs=None,
557	const=None,
558	default=None,
559	type=None,
560	choices=None,
561	required=False,
562	help=None,
563	metavar=None
564	):
UNCOV 565	if nargs != 4:	×
566	raise ValueError("nargs must be 4!")	×
567
UNCOV 568	super().__init__(	×
569	option_strings=option_strings,
570	dest=dest,
571	nargs=nargs,
572	const=const,
573	default=default,
574	type=type,
575	choices=choices,
576	required=required,
577	help=help,
578	metavar=metavar
579	)
580
581	def __call__(self, parser, namespace, values, option_string=None):	1✔
UNCOV 582	S, N, W, E = values	×
583
UNCOV 584	if N <= S or E <= W:	×
585	raise ArgumentError(self, 'Bounding box has no size; make sure you use "S N W E"')	×
586
UNCOV 587	for sn in (S, N):	×
588	if sn < -90 or sn > 90:	×
589	raise ArgumentError(self, 'Lats are out of S/N bounds (-90 to 90).')	×
590
UNCOV 591	for we in (W, E):	×
592	if we < -180 or we > 180:	×
593	raise ArgumentError(self, 'Lons are out of W/E bounds (-180 to 180); Lons in the format of (0 to 360) are not supported.')	×
594
UNCOV 595	setattr(namespace, self.dest, values)	×

dbekaert / RAiDER / b4dd4a1f-7b07-484b-bfa2-363b6b90e3f8

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