Source code for gridmet.config

"""
This module contains classes and enums used to configure
gridMET processing and specify its parameters
"""


#  Copyright (c) 2021. Harvard University
#
#  Developed by Research Software Engineering,
#  Faculty of Arts and Sciences, Research Computing (FAS RC)
#  Author: Michael A Bouzinier
#
#  Licensed under the Apache License, Version 2.0 (the "License");
#  you may not use this file except in compliance with the License.
#  You may obtain a copy of the License at
#
#         http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS,
#  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#  See the License for the specific language governing permissions and
#  limitations under the License.
#

import datetime
from enum import Enum
from typing import Optional

from nsaph_gis.constants import Geography, RasterizationStrategy
from nsaph_utils.utils.context import Context, Argument, Cardinality

var_doc_string = """
        Gridmet bands or variables. 
        :ref: `doc/bands`
"""


[docs]class DateFilter: """ Class, implementing filtering by dates. Primarily used for debugging and testing purposes to avoid long running calculations. The condition can be specified in one of the following ways: - Range: `YYYY-MM-DD:YYYY-MM-DD` only dates falling in the given range will be accepted. Example: '2009-12-30:2010-01-03' means that only 5 days between the 30-th of December of 2009 and January 3, 2010 will be accepted - Day of Month: `dayofmonth:DD`, example: 'dayofmonth:12' means that only dates corresponding to the 12-th of every month will be accepted - Month: `month:MM` only dates in the given month will be accepted - Month and day of a year: `date:MM-DD`, example: 'date:03-14' means that only March 14 for each year will be accepted. """ def __init__(self, value: str): self.min = None self.max = None self.ftype = None self.values = [] if not value: return if ':' not in value: raise ValueError("Filter spec must include ':'") bounds = value.split(':') if bounds[0].lower() in ["dayofmonth", "month", "date"]: self.ftype = bounds[0].lower() self.values = [v.strip() for v in bounds[1].split(',')] else: self.ftype = "range" self.min = datetime.date.fromisoformat(bounds[0]) self.max = datetime.date.fromisoformat(bounds[1])
[docs] def accept(self, day: datetime.date): if self.ftype == "dayofmonth": dom = str(day.day) if dom in self.values: return True return False elif self.ftype == "month": mnth = str(day.month) if mnth in self.values: return True return False elif self.ftype == "date": dt = day.strftime("%m-%d") if dt in self.values: return True if dt.strip('0') in self.values: return True return False if self.min and day < self.min: return False if self.max and day > self.max: return False return True
[docs]class Shape(Enum): """Type of shape""" point = "point" """Point""" polygon = "polygon" """Polygon"""
[docs]class GridmetVariable(Enum): """ `GridMET Bands <https://developers.google.com/earth-engine/datasets/catalog/IDAHO_EPSCOR_GRIDMET#bands>`_ and additional exposure variable types """ bi = "bi" """Burning index: NFDRS fire danger index""" erc = "erc" """Energy release component: NFDRS fire danger index""" etr = "etr" """Daily reference evapotranspiration: Alfalfa, mm""" fm100 = "fm100" """100-hour dead fuel moisture: %""" fm1000 = "fm1000" """1000-hour dead fuel moisture: %""" pet = "pet" """Potential evapotranspiration""" pr = "pr" """Precipitation amount: mm, daily total """ rmax = "rmax" """Maximum relative humidity: %""" rmin = "rmin" """Minimum relative humidity: %""" sph = "sph" """Specific humididy: kg/kg""" srad = "srad" """Surface downward shortwave radiation: W/m^2""" th = "th" """Wind direction: Degrees clockwise from North""" tmmn = "tmmn" """Minimum temperature: K""" tmmx = "tmmx" """Maximum temperature: K""" vpd = "vpd" """Mean vapor pressure deficit: kPa""" vs = "vs" """Wind velocity at 10m: m/s"""
[docs]class OutputType(Enum): """ Type of teh output that the tool should produce """ aggregation = "aggregation" data_dictionary = "data_dictionary"
[docs]class GridContext(Context): """ Defines a configuration object to process aggregations and other tasks over data grids """ _variables = Argument("variables", help="Gridmet bands or variables", aliases=["var"], cardinality=Cardinality.multiple) _strategy = Argument("strategy", aliases=['s'], default=RasterizationStrategy.default.value, help="Rasterization Strategy", valid_values=[v.value for v in RasterizationStrategy]) _destination = Argument("destination", aliases=['dest', 'd'], cardinality=Cardinality.single, default="data/processed", help="Destination directory for the processed files" ) _raw_downloads = Argument("raw_downloads", cardinality=Cardinality.single, default="data/downloads", help="Directory for downloaded raw files" ) _geography = Argument("geography", cardinality = Cardinality.single, default = "zip", help = "The type of geographic area over " + "which we aggregate data", valid_values=[v.value for v in Geography] ) _shapes_dir = Argument("shapes_dir", default="shapes", help="Directory containing shape files for" + " geographies. Directory structure is" + " expected to be: " + ".../${year}/${geo_type}/{point|polygon}/") _shapes = Argument("shapes", cardinality=Cardinality.multiple, default=[Shape.polygon.value], help="Type of shapes to aggregate over", valid_values=[v.value for v in Shape] ) _points = Argument("points", cardinality=Cardinality.single, default="", help="Path to CSV file containing points") _coordinates = Argument("coordinates", aliases=["xy", "coord"], cardinality=Cardinality.multiple, default="", help="Column names for coordinates") _metadata = Argument("metadata", aliases=["m", "meta"], cardinality=Cardinality.multiple, default="", help="Column names for metadata") _extra_columns = Argument("extra_columns", aliases=["e", "extra"], cardinality=Cardinality.multiple, default="", help="Columns with constant values to be added to the output file" ) _statistics = Argument("statistics", cardinality=Cardinality.single, default="mean", help="Type of statistics" ) _dates = Argument("dates", help="Filter dates, can be used " + "to paralellize computations " + "(e.g., over months) and " + "for debugging purposes", required=False) _shape_files = Argument("shape_files", cardinality=Cardinality.multiple, default="", help="Path to shape files", ) _description = Argument("description", cardinality=Cardinality.single, default="Dorieh data model for aggregations of netCDF data", help="Description to be added to data dictionary" ) _table = Argument("table", help = "Name of the table where the aggregated data will be stored", type = str, required = False, aliases = ["t"], default = None, cardinality = Cardinality.single ) _output = Argument("output", aliases=['o'], cardinality=Cardinality.multiple, default=[OutputType.aggregation.value], help="What the tool should output", valid_values=[v.value for v in OutputType]) _ram = Argument("ram", cardinality=Cardinality.single, help="Runtime memory available to the process", default="2G" ) def __init__(self, subclass = None, doc = None): """ Constructor :param doc: Optional argument, specifying what to print as documentation """ self.variables = None """ Gridmet bands or variables :type: List[GridmetVariable] """ self.strategy = None """ Rasterization strategy :type: RasterizationStrategy """ self.destination = None '''Destination directory for the processed files''' self.raw_downloads = None '''Directory for downloaded raw files''' self.geography = None """ The type of geographic area over which we aggregate data :type: Geography """ self.shapes_dir = None '''Directory containing shape files for geographies''' self.shapes = None """ Type of shapes to aggregate over, e.g. points, polygons :type: List[Shape] """ self.shape_files = None self.points = None '''Path to CSV file containing points''' self.coordinates = None '''Column names for coordinates''' self.metadata = None '''Column names for metadata''' self.extra_columns = None '''Columns with constant values to be added to the output file''' self.dates: Optional[DateFilter] = None '''Filter on dates - for debugging purposes only''' self.statistics = None '''Type of statistics''' self.description = None '''Description to be added to data dictionary''' self.table = None '''Name of the table where the aggregated data will be stored''' self.output = None '''Type of the output the tool should produce''' self.ram = None '''Runtime memory available to the process''' if subclass is None: super().__init__(GridContext, doc, include_default = True) else: super().__init__(subclass, doc, include_default = True) self._attrs += [ attr[1:] for attr in GridContext.__dict__ if attr[0] == '_' and attr[1] != '_' ]
[docs] def validate(self, attr, value): value = super().validate(attr, value) if attr == self._shapes.name: return [Shape(v) for v in value] if attr == self._geography.name: return Geography[value] if attr == self._strategy.name: return RasterizationStrategy[value] if attr == self._output.name: return [OutputType[v] for v in value] if attr == self._dates.name: if value: return DateFilter(value) if attr == self._ram.name: value = value.strip().lower() nv = "" postfix = "" for c in value: if c.isdigit(): nv += c else: postfix += c n = int(nv) m = { "": 1, "k": 1000, "m": 1000000, "g": 1000000000, "t": 1000000000000 }.get(postfix[0]) if m is None: raise ValueError("Invalid value for RAM: " + value) value = n * m return value
[docs]class GridMETContext(GridContext): """ Defines a configuration object to process aggregations and other tasks over data grids containing gridMET data. Includes validation that a correct gridMET band is provided """ # _variables = Argument("variables", # help="Gridmet bands or variables", # aliases=["var"], # cardinality=Cardinality.multiple, # valid_values=[v.value for v in GridmetVariable]) GridContext._variables.valid_values=[v.value for v in GridmetVariable] def __init__(self, doc = None): super().__init__(GridMETContext, doc)
[docs] def validate(self, attr, value): value = super().validate(attr, value) if attr == self._variables.name: return [GridmetVariable(v) for v in value] return value