467c297a-bd37-4a23-96fd-379ca31e196a

Committed 26 Aug 2025 03:44PM UTC coverage: 51.133% (+0.1%) from 50.985%

Build # 467c297a-bd37-4a23-96fd-379ca31e196a

Build Type

Pull #764

circleci

Committed by

nate-kean

Commit Message

Remove Python 3.8 support

- Update mentions of 3.8 (or 3.7) as minimum version to say 3.9

Pull Request Pull Request #764: Remove Python 3.8 support

Run Details

3204 of 6266 relevant lines covered (51.13%)

0.51 hits per line

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

60.94

/tools/RAiDER/cli/validators.py

import argparse
import datetime as dt
import importlib
import re
import sys
from pathlib import Path
from typing import Any, Optional, Union

import numpy as np
import pandas as pd


if sys.version_info >= (3,11):
    from typing import Self
else:
    Self = Any

from RAiDER.cli.types import (
    AOIGroupUnparsed,
    DateGroup,
    DateGroupUnparsed,
    HeightGroup,
    HeightGroupUnparsed,
    LOSGroupUnparsed,
    RuntimeGroup,
)
from RAiDER.llreader import AOI, BoundingBox, GeocodedFile, Geocube, RasterRDR, StationFile
from RAiDER.logger import logger
from RAiDER.losreader import LOS, Conventional, Zenith
from RAiDER.models.weatherModel import WeatherModel
from RAiDER.types import BB
from RAiDER.utilFcns import rio_extents, rio_profile


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


def parse_weather_model(weather_model_name: str, aoi: AOI) -> WeatherModel:
    weather_model_name = weather_model_name.upper().replace('-', '')
    try:
        _, Model = get_wm_by_name(weather_model_name)
    except ModuleNotFoundError:
        raise NotImplementedError(
            f'Model {weather_model_name} is not yet fully implemented, please contribute!'
        )

    # Check that the user-requested bounding box is within the weather model domain
    model: WeatherModel = Model()
    model.checkValidBounds(aoi.bounds())

    return model


def get_los(los_group: LOSGroupUnparsed) -> LOS:
    if los_group.orbit_file is not None:
        if los_group.ray_trace:
            from RAiDER.losreader import Raytracing
            los = Raytracing(los_group.orbit_file)
        else:
            los = Conventional(los_group.orbit_file)

    elif los_group.los_file is not None:
        if los_group.ray_trace:
            from RAiDER.losreader import Raytracing
            los = Raytracing(los_group.los_file, los_group.los_convention)
        else:
            los = Conventional(los_group.los_file, los_group.los_convention)

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

    return los


def get_heights(height_group: HeightGroupUnparsed, aoi_group: AOIGroupUnparsed, runtime_group: RuntimeGroup) -> HeightGroup:
    """Parse the Height info and download a DEM if needed."""
    result = HeightGroup(
        dem=height_group.dem,
        use_dem_latlon=height_group.use_dem_latlon,
        height_file_rdr=height_group.height_file_rdr,
        height_levels=None,
    )

    if height_group.dem is not None:
        if aoi_group.station_file is not None:
            station_data = pd.read_csv(aoi_group.station_file)
            if 'Hgt_m' not in station_data:
                result.dem = runtime_group.output_directory / 'GLO30.dem'
        elif Path(height_group.dem).exists():
            # crop the DEM
            if aoi_group.bounding_box is not None:
                dem_bounds = rio_extents(rio_profile(height_group.dem))
                lats: BB.SN = dem_bounds[:2]
                lons: BB.WE = dem_bounds[2:]
                if isOutside(
                    parse_bbox(aoi_group.bounding_box),
                    getBufferedExtent(
                        lats,
                        lons,
                        buffer_size=_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: will download the dem later

    elif height_group.height_file_rdr is None:
        # download the DEM if needed
        result.dem = runtime_group.output_directory / 'GLO30.dem'

    if height_group.height_levels is not None:
        if isinstance(height_group.height_levels, str):
            levels = re.findall('[-0-9]+', height_group.height_levels)
        else:
            levels = height_group.height_levels

        levels = np.array([float(level) for level in levels])
        if np.any(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.'
            )
        result.height_levels = list(levels)

    return result


def get_query_region(aoi_group: AOIGroupUnparsed, height_group: HeightGroupUnparsed, cube_spacing_in_m: float) -> AOI:
    """Parse the query region from inputs.
    
    This function determines the query region from the input parameters. It will return an AOI object that can be used
    to query the weather model.
    Note: both an AOI group and a height group are necessary in case a DEM is needed.
    """
    # Get bounds from the inputs
    # make sure this is first
    if height_group.use_dem_latlon:
        query = GeocodedFile(Path(height_group.dem), is_dem=True, cube_spacing_in_m=cube_spacing_in_m)

    elif aoi_group.lat_file is not None or aoi_group.lon_file is not None:
        if aoi_group.lat_file is None or aoi_group.lon_file is None:
            raise ValueError('A lon_file must be specified if a lat_file is specified')
        query = RasterRDR(
            aoi_group.lat_file, aoi_group.lon_file,
            height_group.height_file_rdr, height_group.dem,
            cube_spacing_in_m=cube_spacing_in_m
        )

    elif aoi_group.station_file is not None:
        query = StationFile(aoi_group.station_file, cube_spacing_in_m=cube_spacing_in_m)

    elif aoi_group.bounding_box is not None:
        bbox = parse_bbox(aoi_group.bounding_box)
        if np.min(bbox[0]) < -90 or 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, cube_spacing_in_m=cube_spacing_in_m)

    elif aoi_group.geocoded_file is not None:
        geocoded_file_path = Path(aoi_group.geocoded_file)
        filename = geocoded_file_path.name.upper()
        if filename.startswith('SRTM') or filename.startswith('GLO'):
            logger.debug('Using user DEM: %s', filename)
            is_dem = True
        else:
            is_dem = False
        query = GeocodedFile(geocoded_file_path, is_dem=is_dem, cube_spacing_in_m=cube_spacing_in_m)

    # untested
    elif aoi_group.geo_cube is not None:
        query = Geocube(aoi_group.geo_cube, cube_spacing_in_m)

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

    return query


def parse_bbox(bbox: Union[str, list[Union[int, float]], tuple]) -> BB.SNWE:
    """Parse a bounding box string input and ensure it is valid."""
    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 S, N, W, E


def parse_dates(date_group: DateGroupUnparsed) -> DateGroup:
    """Determine the requested dates from the input parameters."""
    if date_group.date_list is not None:
        if isinstance(date_group.date_list, str):
            unparsed_dates = re.findall('[0-9]+', date_group.date_list)
        elif isinstance(date_group.date_list, int):
            unparsed_dates = [date_group.date_list]
        else:
            unparsed_dates = date_group.date_list
        date_list = [coerce_into_date(d) for d in unparsed_dates]

    else:
        if date_group.date_start is None:
            raise ValueError('Inputs must include either date_list or date_start')
        start = coerce_into_date(date_group.date_start)

        if date_group.date_end is not None:
            end = coerce_into_date(date_group.date_end)
        else:
            end = start

        if date_group.date_step:
            step = int(date_group.date_step)
        else:
            step = 1

        date_list = [
            start + dt.timedelta(days=step)
            for step in range(0, (end - start).days + 1, step)
        ]
    
    return DateGroup(
        date_list=date_list,
    )


def coerce_into_date(val: Union[int, str]) -> dt.date:
    """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 dt.datetime.strptime(str(val), yf).date()
        except ValueError:
            pass

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


def get_wm_by_name(model_name: str) -> tuple[str, WeatherModel]:
    """
    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. It 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('-', '')
    module = importlib.import_module(module_name)
    Model = getattr(module, model_name.upper().replace('-', ''))
    return module_name, Model


def getBufferedExtent(lats: BB.SN, lons: BB.WE, buffer_size: float=0.0) -> BB.SNWE:
    """Get the bounding box around a set of lats/lons."""
    return (
        min(lats) - buffer_size,
        max(lats) + buffer_size,
        min(lons) - buffer_size,
        max(lons) + buffer_size
    )


def isOutside(extent1: BB.SNWE, extent2: BB.SNWE) -> bool:
    """Determine whether any of extent1 lies outside extent2.

    extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon] (SNWE).
    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]
    return any((t1, t2, t3, t4))


def isInside(extent1: BB.SNWE, extent2: BB.SNWE) -> bool:
    """Determine whether all of extent1 lies inside extent2.

    extent1/2 should be a list containing [lower_lat, upper_lat, left_lon, right_lon] (SNWE).
    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]
    return all((t1, t2, t3, t4))


# below are for downloadGNSSDelays
def date_type(val: Union[int, str]) -> dt.date:
    """Parse a date from a string in pseudo-ISO 8601 format."""
    try:
        return coerce_into_date(val)
    except ValueError as exc:
        raise argparse.ArgumentTypeError(str(exc))


class MappingType:
    """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()
    _default: Union[object, Any]

    def __init__(self, **kwargs: dict[str, Any]) -> None:
        self.mapping = kwargs
        self._default = self.UNSET

    def default(self, default: Any) -> Self:  # noqa: ANN401
        """Set a default value if no mapping is found."""
        self._default = default
        return self

    def __call__(self, arg: str) -> Any:  # noqa: ANN401
        if arg in self.mapping:
            return self.mapping[arg]

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

        return self._default


class IntegerOnRangeType:
    """A type that converts arguments to integers and enforces that they are on a certain range.

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

    def __init__(self, lo: Optional[int]=None, hi: Optional[int]=None) -> None:
        self.lo = lo
        self.hi = hi

    def __call__(self, arg: Any) -> int:  # noqa: ANN401
        integer = int(arg)

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

        return integer


class IntegerMappingType(MappingType, IntegerOnRangeType):
    """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: Optional[int]=None, hi: Optional[int]=None, mapping: Optional[dict[str, Any]]={}, **kwargs: dict[str, Any]) -> None:
        IntegerOnRangeType.__init__(self, lo, hi)
        kwargs.update(mapping)
        MappingType.__init__(self, **kwargs)

    def __call__(self, arg: Any) -> Union[int, Any]:  # noqa: ANN401
        try:
            return IntegerOnRangeType.__call__(self, arg)
        except ValueError:
            return MappingType.__call__(self, arg)


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

    def __init__(
        self,
        option_strings,  # noqa: ANN001 -- see argparse.Action.__init__
        dest,  # noqa: ANN001
        nargs=None,  # noqa: ANN001
        const=None,  # noqa: ANN001
        default=None,  # noqa: ANN001
        type=None,  # noqa: ANN001
        choices=None,  # noqa: ANN001
        required=False,  # noqa: ANN001
        help=None,  # noqa: ANN001
        metavar=None,  # noqa: ANN001
    ) -> 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, _, namespace, values, __=None):  # noqa: ANN001, ANN204 -- see argparse.Action.__call__
        if len(values) > 3 or not values:
            raise argparse.ArgumentError(self, 'Only 1, 2 dates, or 2 dates and interval may be supplied')

        if len(values) == 2:
            start, end = values
            values = [start + dt.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 argparse.ArgumentError(self, 'The stepsize should be in integer days')

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

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

        setattr(namespace, self.dest, values)


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

    def __init__(
        self,
        option_strings,  # noqa: ANN001 -- see argparse.Action.__init__
        dest,  # noqa: ANN001
        nargs=None,  # noqa: ANN001
        const=None,  # noqa: ANN001
        default=None,  # noqa: ANN001
        type=None,  # noqa: ANN001
        choices=None,  # noqa: ANN001
        required=False,  # noqa: ANN001
        help=None,  # noqa: ANN001
        metavar=None,  # noqa: ANN001
    ) -> 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, _, namespace, values, __=None):  # noqa: ANN001, ANN204 -- see argparse.Action.__call__
        S, N, W, E = values

        if N <= S or E <= W:
            raise argparse.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 argparse.ArgumentError(self, 'Lats are out of S/N bounds (-90 to 90).')

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

dbekaert / RAiDER / 467c297a-bd37-4a23-96fd-379ca31e196a

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