diff --git a/src/natcap/invest/datastack.py b/src/natcap/invest/datastack.py index 266dbfe53c..6d445bb059 100644 --- a/src/natcap/invest/datastack.py +++ b/src/natcap/invest/datastack.py @@ -34,6 +34,7 @@ from natcap.invest import spec from natcap.invest import utils from natcap.invest.utils import base_model_id +import pandas LOGGER = logging.getLogger(__name__) DATASTACK_EXTENSION = '.invest.tar.gz' @@ -123,6 +124,14 @@ def get_datastack_info(filepath, extract_path=None): return 'logfile', extract_parameters_from_logfile(filepath) +class Datastack(): + + def __init__(self, target_dir): + self.target_dir = target_dir + self.files_found = {} + self.args = {} + + def build_datastack_archive(args, model_id, datastack_path): """Build an InVEST datastack from an arguments dict. @@ -154,199 +163,37 @@ def build_datastack_archive(args, model_id, datastack_path): archive_filehandler.setLevel(logging.NOTSET) logging.getLogger().addHandler(archive_filehandler) - # For tracking existing files so we don't copy files in twice - files_found = {} LOGGER.debug(f'Keys: {sorted(args.keys())}') - spatial_types = {spec.SingleBandRasterInput, spec.VectorInput, - spec.RasterOrVectorInput} - file_based_types = spatial_types.union({ - spec.CSVInput, spec.FileInput, spec.WorkspaceInput}) - rewritten_args = {} + datastack = Datastack(data_dir) for key in args: - # Allow the model to override specific arguments in datastack archive - # prep. This is useful for tables (like HRA) that are too complicated - # to describe in the MODEL_SPEC format, but use a common specification - # for the other args keys. - override_funcname = f'_override_datastack_archive_{key}' - if hasattr(module, override_funcname): - LOGGER.debug(f'Using model override function for key {key}') - # Notes about the override function: - # * Function may modify files_found - # * If this function copies data into the data dir, it _should_ - # be within its own folder (e.g. - # {data_dir}/criteria_table_path_data/) to minimize chances of - # stomping on other data. But this is up to the function to - # decide. - # * The override function is responsible for logging whatever is - # useful to include in the logfile. - rewritten_args[key] = getattr(module, override_funcname)( - args[key], data_dir, files_found) - continue - LOGGER.info(f'Starting to archive arg "{key}": {args[key]}') # Possible that a user might pass an args key that doesn't belong to # this model. Skip if so. if key not in module.MODEL_SPEC.inputs: LOGGER.info(f'Skipping arg {key}; not in model MODEL_SPEC') - input_spec = module.MODEL_SPEC.get_input(key) - # We don't want to accidentally archive a user's complete workspace - # directory, complete with prior runs there. - if isinstance(input_spec, spec.WorkspaceInput): - LOGGER.debug( - f"Skipping workspace directory: {args['workspace_dir']}") - continue - - if type(input_spec) in file_based_types: - if args[key] in {None, ''}: - LOGGER.info( - f'Skipping key {key}, value is empty and cannot point to ' - 'a file.') - rewritten_args[key] = '' - continue - - # Python can't handle mixed file separators, so let's just - # standardize on linux filepaths. - source_path = args[key].replace('\\', '/') - - # If we already know about the parameter, then we can just reuse it - # and skip the file copying. - if source_path in files_found: - LOGGER.debug( - f'Key {key} is known: using {files_found[source_path]}') - rewritten_args[key] = files_found[source_path] - continue - - if type(input_spec) is spec.CSVInput: - # check the CSV for columns that may be spatial. - # But also, the columns specification might not be listed, so don't - # require that 'columns' exists in the MODEL_SPEC. - spatial_columns = [] - if input_spec.columns: - for col_spec in input_spec.columns: - if type(col_spec) in spatial_types: - spatial_columns.append(col_spec.id) - - LOGGER.debug(f'Detected spatial columns: {spatial_columns}') - - csv_dir = os.path.join(data_dir, f'{key}_csv') - os.makedirs(csv_dir) - - target_csv_path = os.path.join( - csv_dir, os.path.basename(source_path)) - if not spatial_columns: - LOGGER.debug( - f'No spatial columns, copying to {target_csv_path}') - shutil.copyfile(source_path, target_csv_path) - else: - contained_files_dir = os.path.join( - csv_dir, f'{key}_csv_data') - - dataframe = input_spec.get_validated_dataframe(source_path) - csv_source_dir = os.path.abspath(os.path.dirname(source_path)) - for spatial_column_name in spatial_columns: - # Iterate through the spatial columns, identify the set of - # unique files and copy them out. - # if a string is not a filepath, assume it's supposed to be - # there and skip it - for row_index, column_value in dataframe[ - spatial_column_name.lower()].items(): - if ((isinstance(column_value, float) and - math.isnan(column_value)) or - column_value == ''): - # The table cell is blank, so skip it. - # We can't compare nan values directly in a way - # that also works for strings, so skip it. - continue - - source_filepath = None - for possible_filepath in ( - column_value, - os.path.join(csv_source_dir, column_value)): - if os.path.exists(possible_filepath): - source_filepath = possible_filepath - break - - # If we didn't end up finding a valid source filepath - # for the field value, assume it's supposed to be that - # way and leave it alone. - if not source_filepath: - continue - - try: - # This path is already relative to the data - # directory - target_filepath = files_found[source_filepath] - except KeyError: - basename = os.path.splitext( - os.path.basename(source_filepath))[0] - target_dir = os.path.join( - contained_files_dir, - f'{row_index}_{basename}') - target_filepath = utils.copy_spatial_files( - source_filepath, target_dir) - target_filepath = os.path.relpath( - target_filepath, csv_dir) - - LOGGER.debug( - 'Spatial file in CSV copied from ' - f'{source_filepath} --> {target_filepath}') - dataframe.at[ - row_index, spatial_column_name] = target_filepath - files_found[source_filepath] = target_filepath - - LOGGER.debug( - f'Rewritten spatial CSV written to {target_csv_path}') - dataframe.to_csv(target_csv_path) - - target_arg_value = target_csv_path - files_found[source_path] = target_arg_value - - elif type(input_spec) is spec.FileInput: - target_filepath = os.path.join( - data_dir, f'{key}_file') - shutil.copyfile(source_path, target_filepath) - target_arg_value = target_filepath - files_found[source_path] = target_arg_value - - elif type(input_spec) in spatial_types: - # Create a directory with a readable name, something like - # "aoi_path_vector" or "lulc_cur_path_raster". - spatial_dir = os.path.join(data_dir, f'{key}_{input_spec.type}') - target_arg_value = utils.copy_spatial_files( - source_path, spatial_dir) - files_found[source_path] = target_arg_value - - else: - LOGGER.debug( - f"Type {type(input_spec)} is not filesystem-based; " - "recording value directly") - # not a filesystem-based type - # Record the value directly - target_arg_value = args[key] - rewritten_args[key] = target_arg_value + LOGGER.info(f'Starting to archive arg "{key}": {args[key]}') + module.MODEL_SPEC.get_input(key).archive_for_datastack(args[key], datastack) LOGGER.info('Args preprocessing complete') - - LOGGER.debug(f'found files: \n{pprint.pformat(files_found)}') - LOGGER.debug(f'new arguments: \n{pprint.pformat(rewritten_args)}') + LOGGER.debug(f'found files: \n{pprint.pformat(datastack.files_found)}') + LOGGER.debug(f'new arguments: \n{pprint.pformat(datastack.args)}') # write parameters to a new json file in the temp workspace param_file_uri = os.path.join(temp_workspace, 'parameters' + PARAMETER_SET_EXTENSION) parameter_set = build_parameter_set( - rewritten_args, model_id, param_file_uri, relative=True) + datastack.args, model_id, param_file_uri, relative=True) # write metadata for all files in args - keywords = [module.MODEL_SPEC.model_id, 'InVEST'] for k, v in args.items(): if isinstance(v, str) and os.path.isfile(v): - this_arg_spec = module.MODEL_SPEC.get_input(k) # write metadata file to target location (in temp dir) - subdir = os.path.dirname(parameter_set['args'][k]) - target_location = os.path.join(temp_workspace, subdir) - spec.write_metadata_file(v, this_arg_spec, keywords, - out_workspace=target_location) + module.MODEL_SPEC.get_input(k).write_metadata_file( + datasource_path=v, + keywords_list=[module.MODEL_SPEC.model_id, 'InVEST'], + out_workspace=os.path.join( + temp_workspace, os.path.dirname(parameter_set['args'][k]))) # Remove the handler before archiving the working dir (and the logfile) archive_filehandler.close() @@ -354,10 +201,9 @@ def build_datastack_archive(args, model_id, datastack_path): # archive the workspace. with tempfile.TemporaryDirectory() as temp_dir: - temp_archive = os.path.join(temp_dir, 'invest_archive') archive_name = shutil.make_archive( - temp_archive, 'gztar', root_dir=temp_workspace, - logger=LOGGER, verbose=True) + os.path.join(temp_dir, 'invest_archive'), 'gztar', + root_dir=temp_workspace, logger=LOGGER, verbose=True) shutil.move(archive_name, datastack_path) diff --git a/src/natcap/invest/hra/hra.py b/src/natcap/invest/hra/hra.py index 96b4bd206b..0960e00d62 100644 --- a/src/natcap/invest/hra/hra.py +++ b/src/natcap/invest/hra/hra.py @@ -45,6 +45,85 @@ 'TILED=YES', 'BIGTIFF=YES', 'COMPRESS=DEFLATE', 'BLOCKXSIZE=256', 'BLOCKYSIZE=256') +class HRACriteriaTableInput(spec.CSVInput): + + def archive_for_datastack(self, value, datastack): + """Prepare the HRA criteria_table_path input for archiving in a datastack. + + This function rewrites a criteria table (which may contain spatial ratings + layers), copying any spatial layers into the data directory provided, which + will be included in the datastack archive. + + Args: + value (string): The path to the criteria table provided by the user. + datastack (natcap.invest.datastack.Datastack): datastack object to be + updated with the archived value + + Returns: + None + """ + criteria_table_array = utils.read_csv_to_dataframe( + value, header=None).to_numpy() + contained_data_dir = os.path.join( + datastack.target_dir, f'{self.id}_csv', f'{self.id}_csv_data') + + known_rating_cols = set() + for row in range(1, len(criteria_table_array)): # skip named habitats + # When we encounter an empty row, reset the known ratings columns in + # case one of the sub-tables changes the order around. + try: + if numpy.all(numpy.isnan( + criteria_table_array[row].astype(numpy.float32))): + known_rating_cols = set() + continue + except ValueError: + # ValueError when there are any string values in the row + pass + + if not known_rating_cols: + for col in range(1, len(criteria_table_array[0])): + if criteria_table_array[row, col] == 'RATING': + known_rating_cols.add(col) + continue # skip the RATING headers row. + + for col in known_rating_cols: + cell_value = criteria_table_array[row, col] + try: + float(cell_value) + continue + except ValueError: + # When value is obviously not a number. + pass + + # Expand the path if it's not absolute + cell_value = utils.expand_path(cell_value, value) + if not os.path.exists(cell_value): + LOGGER.warning(f'File not found: {cell_value}') + continue + if cell_value in datastack.files_found: + LOGGER.info( + f"File {cell_value} already known, perhaps from another " + f"cell or table. Reusing {datastack.files_found[cell_value]}") + criteria_table_array[row, col] = datastack.files_found[cell_value] + else: + dir_for_this_spatial_data = os.path.join( + contained_data_dir, + os.path.splitext(os.path.basename(cell_value))[0]) + LOGGER.info(f"Copying spatial file {cell_value} --> " + f"{dir_for_this_spatial_data}") + new_path = utils.copy_spatial_files( + cell_value, dir_for_this_spatial_data) + criteria_table_array[row, col] = new_path + datastack.files_found[cell_value] = new_path + + target_output_path = os.path.join(datastack.target_dir, f'{self.id}_csv', + os.path.basename(value)) + os.makedirs(os.path.dirname(target_output_path), exist_ok=True) + numpy.savetxt(target_output_path, criteria_table_array, delimiter=',', + fmt="%s", encoding="UTF-8") + datastack.args[self.id] = target_output_path + + MODEL_SPEC = spec.ModelSpec( model_id="habitat_risk_assessment", model_title=gettext("Habitat Risk Assessment"), @@ -127,7 +206,7 @@ index_col="name", na_allowed=["stressor buffer (meters)"] ), - spec.CSVInput( + HRACriteriaTableInput( id="criteria_table_path", name=gettext("criteria scores table"), about=gettext("A table of criteria scores for all habitats and stressors."), @@ -2367,91 +2446,6 @@ def _sum_op(*array_list): _sum_op, target_result_path, target_datatype, target_nodata) -def _override_datastack_archive_criteria_table_path( - criteria_table_path, data_dir, known_files): - """Prepare the HRA criteria_table_path input for archiving in a datastack. - - This function rewrites a criteria table (which may contain spatial ratings - layers), copying any spatial layers into the data directory provided, which - will be included in the datastack archive. - - Args: - criteria_table_path (string): The path to the criteria table provided - by the user. - data_dir (string): the path to where data files will be written. - known_files (dict): a dict mapping original source files to their - location within the ``data_dir``. - - Note: - The ``known_files`` dict may be modified by this function. - - Returns: - The path to where the rewritten criteria table path is located within - the data dir. - """ - args_key = 'criteria_table_path' - criteria_table_array = utils.read_csv_to_dataframe( - criteria_table_path, header=None).to_numpy() - contained_data_dir = os.path.join(data_dir, f'{args_key}_csv', - f'{args_key}_csv_data') - - known_rating_cols = set() - for row in range(1, len(criteria_table_array)): # skip named habitats - # When we encounter an empty row, reset the known ratings columns in - # case one of the sub-tables changes the order around. - try: - if numpy.all(numpy.isnan( - criteria_table_array[row].astype(numpy.float32))): - known_rating_cols = set() - continue - except ValueError: - # ValueError when there are any string values in the row - pass - - if not known_rating_cols: - for col in range(1, len(criteria_table_array[0])): - if criteria_table_array[row, col] == 'RATING': - known_rating_cols.add(col) - continue # skip the RATING headers row. - - for col in known_rating_cols: - value = criteria_table_array[row, col] - try: - float(value) - continue - except ValueError: - # When value is obviously not a number. - pass - - # Expand the path if it's not absolute - value = utils.expand_path(value, criteria_table_path) - if not os.path.exists(value): - LOGGER.warning(f'File not found: {value}') - continue - if value in known_files: - LOGGER.info( - f"File {value} already known, perhaps from another " - f"cell or table. Reusing {known_files[value]}") - criteria_table_array[row, col] = known_files[value] - else: - dir_for_this_spatial_data = os.path.join( - contained_data_dir, - os.path.splitext(os.path.basename(value))[0]) - LOGGER.info(f"Copying spatial file {value} --> " - f"{dir_for_this_spatial_data}") - new_path = utils.copy_spatial_files( - value, dir_for_this_spatial_data) - criteria_table_array[row, col] = new_path - known_files[value] = new_path - - target_output_path = os.path.join(data_dir, f'{args_key}_csv', - os.path.basename(criteria_table_path)) - os.makedirs(os.path.dirname(target_output_path), exist_ok=True) - numpy.savetxt(target_output_path, criteria_table_array, delimiter=',', - fmt="%s", encoding="UTF-8") - return target_output_path - - @validation.invest_validator def validate(args, limit_to=None): """Validate args to ensure they conform to ``execute``'s contract. diff --git a/src/natcap/invest/spec.py b/src/natcap/invest/spec.py index 4bd69cc024..f191fcafe6 100644 --- a/src/natcap/invest/spec.py +++ b/src/natcap/invest/spec.py @@ -8,6 +8,7 @@ import pprint import queue import re +import shutil import threading import types import typing @@ -190,10 +191,87 @@ class ImmutableBaseModel(BaseModel): class IOModel(ImmutableBaseModel): """Base class for both `Input` and `Output`.""" + id: str + """Input/output identifier that should be unique within a model""" + + about: typing.Union[str, None] = None + """User-facing description of the input/output""" + model_config = ConfigDict(arbitrary_types_allowed=True) """Allow fields to have arbitrary types that don't inherit from BaseModel (needed for pint.Unit).""" + def write_metadata_file(self, datasource_path, keywords_list, + lineage_statement='', out_workspace=None): + """Write a metadata sidecar file for an invest dataset. + + Create metadata for invest model inputs or outputs, taking care to + preserve existing human-modified attributes. + + Note: We do not want to overwrite any existing metadata so if there is + invalid metadata for the datasource (i.e., doesn't pass geometamaker + validation in ``describe``), this function will NOT create new metadata. + + Args: + datasource_path (str) - filepath to the data to describe + keywords_list (list) - sequence of strings + lineage_statement (str, optional) - string to describe origin of + the dataset + out_workspace (str, optional) - where to write metadata if different + from data location + Returns: + None + + """ + try: + resource = geometamaker.describe(datasource_path, compute_stats=True) + except ValueError as e: + # Don't want function to fail bc can't create metadata due to invalid filetype + LOGGER.debug(f"Skipping metadata creation for {datasource_path}: {e}") + return None + resource.set_lineage(lineage_statement) + # a pre-existing metadata doc could have keywords + words = resource.get_keywords() + resource.set_keywords(set(words + keywords_list)) + + if self.about: + resource.set_description(self.about) + attr_specs = None + if hasattr(self, 'columns') and self.columns: + attr_specs = self.columns + if hasattr(self, 'fields') and self.fields: + attr_specs = self.fields + if attr_specs: + # field names in attr_spec might not match the case of the + # actual fieldname in the data because + # invest does not require case-sensitive fieldnames + field_lookup = { + field.name.lower(): field for field in resource._get_fields()} + for nested_spec in attr_specs: + try: + field_metadata = field_lookup[nested_spec.id.lower()] + # Field description only gets set if its empty, i.e. '' + if len(field_metadata.description.strip()) < 1: + resource.set_field_description( + field_metadata.name, description=nested_spec.about) + # units only get set if empty + if len(field_metadata.units.strip()) < 1: + units = format_unit(nested_spec.units) if hasattr( + nested_spec, 'units') else '' + resource.set_field_description( + field_metadata.name, units=units) + except KeyError as error: + # fields that are in the spec but missing + # from model results because they are conditional. + LOGGER.debug(error) + if isinstance(self, SingleBandRasterInput) or isinstance( + self, SingleBandRasterOutput): + if len(resource.get_band_description(1).units) < 1: + units = format_unit(self.units) + resource.set_band_description(1, units=units) + + resource.write(workspace=out_workspace) + class Input(IOModel): """A data input, or parameter, of an invest model. @@ -202,9 +280,6 @@ class Input(IOModel): input field in the InVEST workbench. This does not store the value of the parameter for a specific run of the model. """ - id: str - """Input identifier that should be unique within a model""" - name: typing.Union[str, None] = None """The user-facing name of the input. The workbench UI displays this property as a label for each input. The name should be as short as @@ -217,9 +292,6 @@ class Input(IOModel): Bad examples: ``PRECIPITATION``, ``kc_factor``, ``table of valuation parameters`` """ - about: typing.Union[str, None] = None - """User-facing description of the input""" - required: typing.Union[bool, str] = True """Whether the input is required to be provided. Defaults to True. Set to False if the input is always optional. If the input is conditionally @@ -303,6 +375,34 @@ def describe_rst(self): return [rst_line] + def archive_for_datastack(self, value, datastack): + """Archive a given value of this input into a datastack. + + This method can be overridden to handle specific types of input data, + or even for specific model inputs that need custom handling. + This is useful for tables (like HRA) that are too complicated + to describe in the MODEL_SPEC format, but use a common specification + for the other args keys. + Notes about overriding this method: + - should add the archived value to datastack.args + - should update datastack.files_found if any new files are included + - if this function copies data into datastack.target_dir, it _should_ + be within its own folder (e.g. + {data_dir}/criteria_table_path_data/) to minimize chances of + stomping on other data. But this is up to the function to + decide. + - The override function is responsible for logging whatever is + useful to include in the logfile. + + Args: + value (object): value of this input + datastack (natcap.invest.datastack.Datastack): Datastack instance + + Returns: + None + """ + datastack.args[self.id] = value + class Output(IOModel): """A data output, or result, of an invest model. @@ -311,11 +411,6 @@ class Output(IOModel): an invest model. This does not store the value of the output for a specific run of the model. """ - id: str - """Output identifier that should be unique within a model""" - - about: typing.Union[str, None] = None - """User-facing description of the output""" created_if: typing.Union[bool, str] = True """Defaults to True. If the input is only created under a certain condition @@ -392,6 +487,33 @@ def format_path(p): return col.apply(format_path).astype(pandas.StringDtype()) + def archive_for_datastack(self, value, datastack): + + if value in {None, ''}: + LOGGER.info( + f'Skipping key {self.id}, value is empty and cannot point to ' + 'a file.') + datastack.args[self.id] = '' + return + + # Python can't handle mixed file separators, so let's just + # standardize on linux filepaths. + source_path = value.replace('\\', '/') + + # If we already know about the parameter, then we can just reuse it + # and skip the file copying. + if source_path in datastack.files_found: + LOGGER.debug( + f'Key {self.id} is known: using {datastack.files_found[source_path]}') + datastack.args[self.id] = datastack.files_found[source_path] + return + + target_filepath = os.path.join(datastack.target_dir, f'{self.id}_file') + shutil.copyfile(source_path, target_filepath) + datastack.args[self.id] = target_filepath + datastack.files_found[source_path] = target_filepath + + class SpatialFileInput(FileInput): """Base class for raster and vector spatial inputs.""" @@ -454,6 +576,32 @@ def format_path(p): return col.apply(format_path).astype(pandas.StringDtype()) + def archive_for_datastack(self, value, datastack): + + if value in {None, ''}: + datastack.args[self.id] = '' + return + + # Python can't handle mixed file separators, so let's just + # standardize on linux filepaths. + source_path = value.replace('\\', '/') + + # If we already know about the parameter, then we can just reuse it + # and skip the file copying. + if source_path in datastack.files_found: + LOGGER.debug( + f'Key {self.id} is known: using {datastack.files_found[source_path]}') + datastack.args[self.id] = datastack.files_found[source_path] + return + + # Create a directory with a readable name, something like + # "aoi_path_vector" or "lulc_cur_path_raster". + spatial_dir = os.path.join(datastack.target_dir, f'{self.id}_{self.type}') + target_arg_value = utils.copy_spatial_files( + source_path, spatial_dir) + datastack.args[self.id] = target_arg_value + datastack.files_found[source_path] = target_arg_value + class RasterBand(IOModel): """A representation of a single raster band.""" @@ -1065,6 +1213,98 @@ def describe_rst(self): return [rst_line] + def archive_for_datastack(self, value, datastack): + + if value in {None, ''}: + datastack.args[self.id] = '' + return + + # Python can't handle mixed file separators, so let's just + # standardize on linux filepaths. + source_path = value.replace('\\', '/') + + # If we already know about the parameter, then we can just reuse it + # and skip the file copying. + if source_path in datastack.files_found: + LOGGER.debug( + f'Key {self.id} is known: using {datastack.files_found[source_path]}') + datastack.args[self.id] = datastack.files_found[source_path] + return + + # check the CSV for columns that may be spatial. + # But also, the columns specification might not be listed, so don't + # require that 'columns' exists in the MODEL_SPEC. + + csv_dir = os.path.join(datastack.target_dir, f'{self.id}_csv') + os.makedirs(csv_dir) + target_csv_path = os.path.join( + csv_dir, os.path.basename(source_path)) + + if self.columns: + dataframe = self.get_validated_dataframe(source_path) + + for column_spec in [c for c in self.columns if isinstance(c, FileInput)]: + + # Iterate through the columns, identify the set of + # unique files and copy them out. + # if a string is not a filepath, assume it's supposed to be + # there and skip it + for row_index, value in dataframe[column_spec.id.lower()].items(): + if pandas.isna(value) or value == '': + continue # skip empty cells + + # file paths in a csv may be absolute, or relative to the + # csv location + if os.path.isabs(value): + source_filepath = value + else: + source_filepath = os.path.join(os.path.abspath( + os.path.dirname(source_path)), value) + + # If the file path doesn't exist, assume it's supposed to be + # that way and leave it alone. + if not os.path.exists(source_filepath): + continue + + if source_filepath in datastack.files_found: + # the file was already copied into the target dir, + # so refer to the existing copy + target_filepath = datastack.files_found[source_filepath] + else: + basename = os.path.splitext( + os.path.basename(source_filepath))[0] + target_dir = os.path.join( + csv_dir, f'{self.id}_csv_data', + f'{row_index}_{basename}') + os.makedirs(target_dir) + if isinstance(column_spec, SpatialFileInput): + target_filepath = utils.copy_spatial_files( + source_filepath, target_dir) + target_filepath = os.path.relpath( + target_filepath, csv_dir) + else: + target_filepath = os.path.join(target_dir, + os.path.basename(source_filepath)) + shutil.copyfile(source_filepath, target_filepath) + target_filepath = os.path.relpath( + target_filepath, csv_dir) + + + LOGGER.debug( + 'File referenced in CSV copied from ' + f'{source_filepath} --> {target_filepath}') + dataframe.at[ + row_index, column_spec.id] = target_filepath + datastack.files_found[source_filepath] = target_filepath + + LOGGER.debug( + f'Rewritten spatial CSV written to {target_csv_path}') + dataframe.to_csv(target_csv_path) + else: + shutil.copyfile(source_path, target_csv_path) + datastack.args[self.id] = target_csv_path + datastack.files_found[source_path] = target_csv_path + class WorkspaceInput(Input): """A workspace directory path input to an invest model. @@ -1142,6 +1382,10 @@ def serialize(self, handler): 'about': self.about } + def archive_for_datastack(self, value, datastack): + """Skip the workspace directory when building a datastack archive""" + pass + class NumberInput(Input): """A floating-point number input, or parameter, of an invest model. @@ -2073,9 +2317,8 @@ def _generate_metadata(root_key, value): if 'taskgraph.db' in value: return try: - write_metadata_file( - value, self.get_output(root_key), - keywords, lineage_statement) + self.get_output(root_key).write_metadata_file( + value, keywords, lineage_statement) except ValueError as error: # Some unsupported file formats, e.g. html LOGGER.debug(error) @@ -2519,76 +2762,3 @@ def format_type_string(_input): f'`{_input._single_band_raster_input.display_name} <{INPUT_TYPES_HTML_FILE}#{SingleBandRasterInput.rst_section}>`__ or ' f'`{_input._vector_input.display_name} <{INPUT_TYPES_HTML_FILE}#{VectorInput.rst_section}>`__') return f'`{_input.display_name} <{INPUT_TYPES_HTML_FILE}#{_input.rst_section}>`__' - - -def write_metadata_file(datasource_path, spec, keywords_list, - lineage_statement='', out_workspace=None): - """Write a metadata sidecar file for an invest dataset. - - Create metadata for invest model inputs or outputs, taking care to - preserve existing human-modified attributes. - - Note: We do not want to overwrite any existing metadata so if there is - invalid metadata for the datasource (i.e., doesn't pass geometamaker - validation in ``describe``), this function will NOT create new metadata. - - Args: - datasource_path (str) - filepath to the data to describe - spec (dict) - the invest specification for ``datasource_path`` - keywords_list (list) - sequence of strings - lineage_statement (str, optional) - string to describe origin of - the dataset - out_workspace (str, optional) - where to write metadata if different - from data location - Returns: - None - - """ - try: - resource = geometamaker.describe(datasource_path, compute_stats=True) - except ValueError as e: - # Don't want function to fail bc can't create metadata due to invalid filetype - LOGGER.debug(f"Skipping metadata creation for {datasource_path}: {e}") - return None - resource.set_lineage(lineage_statement) - # a pre-existing metadata doc could have keywords - words = resource.get_keywords() - resource.set_keywords(set(words + keywords_list)) - - if spec.about: - resource.set_description(spec.about) - attr_specs = None - if hasattr(spec, 'columns') and spec.columns: - attr_specs = spec.columns - if hasattr(spec, 'fields') and spec.fields: - attr_specs = spec.fields - if attr_specs: - # field names in attr_spec might not match the case of the - # actual fieldname in the data because - # invest does not require case-sensitive fieldnames - field_lookup = { - field.name.lower(): field for field in resource._get_fields()} - for nested_spec in attr_specs: - try: - field_metadata = field_lookup[nested_spec.id.lower()] - # Field description only gets set if its empty, i.e. '' - if len(field_metadata.description.strip()) < 1: - resource.set_field_description( - field_metadata.name, description=nested_spec.about) - # units only get set if empty - if len(field_metadata.units.strip()) < 1: - units = format_unit(nested_spec.units) if hasattr( - nested_spec, 'units') else '' - resource.set_field_description( - field_metadata.name, units=units) - except KeyError as error: - # fields that are in the spec but missing - # from model results because they are conditional. - LOGGER.debug(error) - if isinstance(spec, SingleBandRasterInput) or isinstance( - spec, SingleBandRasterOutput): - if len(resource.get_band_description(1).units) < 1: - units = format_unit(spec.units) - resource.set_band_description(1, units=units) - - resource.write(workspace=out_workspace) diff --git a/tests/test_datastack.py b/tests/test_datastack.py index 02c6cbf77f..f8c4392bcd 100644 --- a/tests/test_datastack.py +++ b/tests/test_datastack.py @@ -381,20 +381,11 @@ def test_archive_extraction(self): textfile.write('hello world!') with open(params['spatial_table'], 'w') as spatial_csv: - # copy existing DEM - # copy existing watersheds - # new raster - # new vector - spatial_csv.write('ID,path\n') - spatial_csv.write(f"1,{params['raster']}\n") - spatial_csv.write(f"2,{params['vector']}\n") - # Create a raster only referenced by the CSV target_csv_raster_path = os.path.join( self.workspace, 'new_raster.tif') pygeoprocessing.new_raster_from_base( params['raster'], target_csv_raster_path, gdal.GDT_UInt16, [0]) - spatial_csv.write(f'3,{target_csv_raster_path}\n') # Create a vector only referenced by the CSV target_csv_vector_path = os.path.join( @@ -406,7 +397,20 @@ def test_archive_extraction(self): params['raster'])['projection_wkt'], 'GeoJSON', ogr_geom_type=ogr.wkbPoint) - spatial_csv.write(f'4,{target_csv_vector_path}\n') + + # Create a json file only referenced by the CSV + nonspatial_path = os.path.join(self.workspace, 'nonspatial.json') + with open(nonspatial_path, 'w') as f: + f.write('{"foo": 1}') + # copy existing DEM + # copy existing watersheds + # new raster + # new vector + spatial_csv.write('ID,spatial_path,nonspatial_path\n') + spatial_csv.write(f"1,{params['raster']},{nonspatial_path}\n") + spatial_csv.write(f"2,{params['vector']},{nonspatial_path}\n") + spatial_csv.write(f'3,{target_csv_raster_path},{nonspatial_path}\n') + spatial_csv.write(f'4,{target_csv_vector_path},{nonspatial_path}\n') archive_path = os.path.join(self.workspace, 'archive.invs.tar.gz') with patch('natcap.invest.datastack.models') as p: @@ -446,7 +450,8 @@ def test_archive_extraction(self): index_col='id', columns=[ spec.IntegerInput(id='id'), - spec.FileInput(id='path')] + spec.FileInput(id='spatial_path'), + spec.FileInput(id='nonspatial_path')] ).get_validated_dataframe( archive_params['spatial_table'] ).to_dict(orient='index') @@ -454,17 +459,20 @@ def test_archive_extraction(self): # assert paths inside CSV are relative to CSV folder contained_files_dir = os.path.join( spatial_csv_dir, 'spatial_table_csv_data') - spatial_file_from_csv = spatial_csv_dict[3]['path'] + spatial_file_from_csv = spatial_csv_dict[3]['spatial_path'] self.assertTrue(os.path.exists(spatial_file_from_csv)) self.assertIn(contained_files_dir, spatial_file_from_csv) + nonspatial_file_from_csv = spatial_csv_dict[3]['nonspatial_path'] + self.assertTrue(os.path.exists(nonspatial_file_from_csv)) + self.assertIn(contained_files_dir, nonspatial_file_from_csv) numpy.testing.assert_allclose( pygeoprocessing.raster_to_numpy_array( - os.path.join(spatial_csv_dir, spatial_csv_dict[3]['path'])), + os.path.join(spatial_csv_dir, spatial_csv_dict[3]['spatial_path'])), pygeoprocessing.raster_to_numpy_array( target_csv_raster_path)) utils._assert_vectors_equal( - os.path.join(spatial_csv_dir, spatial_csv_dict[4]['path']), + os.path.join(spatial_csv_dir, spatial_csv_dict[4]['spatial_path']), target_csv_vector_path) def test_relative_path_failure(self): diff --git a/tests/test_datastack_modules/archive_extraction.py b/tests/test_datastack_modules/archive_extraction.py index 69a5376011..04c0ec1567 100644 --- a/tests/test_datastack_modules/archive_extraction.py +++ b/tests/test_datastack_modules/archive_extraction.py @@ -15,11 +15,12 @@ columns=[ spec.IntegerInput(id='ID'), spec.RasterOrVectorInput( - id='path', + id='spatial_path', fields=[], geometry_types={'POINT', 'POLYGON'}, units=None - ) + ), + spec.FileInput(id='nonspatial_path') ] )], outputs=[], diff --git a/tests/test_hra.py b/tests/test_hra.py index efb66e7663..0c523cae6a 100644 --- a/tests/test_hra.py +++ b/tests/test_hra.py @@ -909,6 +909,7 @@ def test_sum_rasters(self): def test_datastack_criteria_table_override(self): """HRA: verify we store all data referenced in the criteria table.""" + from natcap.invest.datastack import Datastack from natcap.invest.hra import hra criteria_table_path = os.path.join( @@ -955,12 +956,12 @@ def test_datastack_criteria_table_override(self): data_dir = os.path.join(self.workspace_dir, 'datastack_data') csv_dir = os.path.join(data_dir, 'criteria_table_path_csv') - known_files = {} + datastack = Datastack(data_dir) - new_csv_path = hra._override_datastack_archive_criteria_table_path( - criteria_table_path, data_dir, known_files) + hra.MODEL_SPEC.get_input('criteria_table_path').archive_for_datastack( + criteria_table_path, datastack) self.assertEqual( - new_csv_path, + datastack.args['criteria_table_path'], os.path.join(csv_dir, os.path.basename(criteria_table_path)) ) output_criteria_data_dir = os.path.join( @@ -968,7 +969,7 @@ def test_datastack_criteria_table_override(self): self.maxDiff = None self.assertEqual( - known_files, { + datastack.files_found, { eelgrass_path: os.path.join( output_criteria_data_dir, 'eelgrass_connectivity', 'eelgrass_connectivity.shp'), @@ -978,7 +979,7 @@ def test_datastack_criteria_table_override(self): output_criteria_data_dir, 'mgmt2', 'mgmt2.tif') } ) - for copied_filepath in known_files.values(): + for copied_filepath in datastack.files_found.values(): self.assertEqual(True, os.path.exists(copied_filepath)) try: spatial_file = gdal.OpenEx(copied_filepath)