Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Change hazard object into functions #92

Merged
merged 3 commits into from
Jun 6, 2023
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions hydromt_fiat/fiat.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Implement fiat model class"""

from hydromt_fiat.workflows.vulnerability import Vulnerability
from hydromt_fiat.workflows.hazard import Hazard
from hydromt_fiat.workflows.hazard_old import Hazard
from hydromt.models.model_grid import GridModel
from hydromt_fiat.workflows.exposure_vector import ExposureVector
import logging
Expand Down Expand Up @@ -263,7 +263,7 @@ def _configread(self, fn):
return config.load_file(fn)

def check_path_exists(self, fn):
"""TODO: decide to use this or another function (check_file_exist in validation.py)"""
"""TODO: decide to use this or another function (check_file_exist in py)"""
path = Path(fn)
self.logger.debug(f"Reading file {str(path.name)}")
if not fn.is_file():
Expand Down
232 changes: 128 additions & 104 deletions hydromt_fiat/validation.py
Original file line number Diff line number Diff line change
@@ -1,120 +1,144 @@
from pathlib import Path

### TO BE UPDATED ###
class Validation:
"""CONTROL FUNCTIONS"""
def check_dir_exist(dir, name=None):
""" """

def check_dir_exist(self, dir, name=None):
""" """
if not isinstance(dir, Path):
raise TypeError(
f"The directory indicated by the '{name}' parameter does not exist."
)

if not isinstance(dir, Path):
raise TypeError(
f"The directory indicated by the '{name}' parameter does not exist."
)

def check_file_exist(self, root, param_lst, name=None, input_dir=None):
""" """
root = Path(root)
param_lst = [Path(p) for p in param_lst]
for param_idx, param in enumerate(param_lst):
if isinstance(param, dict):
fn_lst = list(param.values())
else:
fn_lst = [param]
for fn_idx, fn in enumerate(fn_lst):
if not Path(fn).is_file():
if root.joinpath(fn).is_file():
def check_file_exist(model, param_lst, name=None, input_dir=None):
""" """
root = Path(model.root)
param_lst = [Path(p) for p in param_lst]
for param_idx, param in enumerate(param_lst):
if isinstance(param, dict):
fn_lst = list(param.values())
else:
fn_lst = [param]
for fn_idx, fn in enumerate(fn_lst):
if not Path(fn).is_file():
if root.joinpath(fn).is_file():
if isinstance(param, dict):
param_lst[param_idx][
list(param.keys())[fn_idx]
] = root.joinpath(fn)
else:
param_lst[param_idx] = root.joinpath(fn)
if input_dir is not None:
if model.get_config(input_dir).joinpath(fn).is_file():
if isinstance(param, dict):
param_lst[param_idx][
list(param.keys())[fn_idx]
] = root.joinpath(fn)
] = model.get_config(input_dir).joinpath(fn)
else:
param_lst[param_idx] = root.joinpath(fn)
if input_dir is not None:
if self.get_config(input_dir).joinpath(fn).is_file():
if isinstance(param, dict):
param_lst[param_idx][
list(param.keys())[fn_idx]
] = self.get_config(input_dir).joinpath(fn)
else:
param_lst[param_idx] = self.get_config(
input_dir
).joinpath(fn)
param_lst[param_idx] = model.get_config(
input_dir
).joinpath(fn)
else:
if isinstance(param, dict):
param_lst[param_idx][list(param.keys())[fn_idx]] = Path(fn)
else:
if isinstance(param, dict):
param_lst[param_idx][list(param.keys())[fn_idx]] = Path(fn)
else:
param_lst[param_idx] = Path(fn)
try:
if isinstance(param, dict):
assert isinstance(
param_lst[param_idx][list(param.keys())[fn_idx]], Path
)
else:
assert isinstance(param_lst[param_idx], Path)
except AssertionError:
if input_dir is None:
raise TypeError(
f"The file indicated by the '{name}' parameter does not"
f" exist in the directory '{root}'."
)
else:
raise TypeError(
f"The file indicated by the '{name}' parameter does not"
f" exist in either of the directories '{root}' or "
f"'{self.get_config(input_dir)}'."
)

def check_uniqueness(self, *args, file_type=None, filename=None):
"""TODO: this function should be checked"""

args = list(args)
if len(args) == 1 and "." in args[0]:
args = args[0].split(".") + args[1:]
branch = args.pop(-1)
for key in args[::-1]:
branch = {key: branch}

if args[0].get_config(args[0], args[1]):
for key in self.staticmaps.data_vars:
if filename == key:
raise ValueError(
f"The filenames of the {file_type} maps should be unique."
param_lst[param_idx] = Path(fn)
try:
if isinstance(param, dict):
assert isinstance(
param_lst[param_idx][list(param.keys())[fn_idx]], Path
)
else:
assert isinstance(param_lst[param_idx], Path)
except AssertionError:
if input_dir is None:
raise TypeError(
f"The file indicated by the '{name}' parameter does not"
f" exist in the directory '{root}'."
)
else:
raise TypeError(
f"The file indicated by the '{name}' parameter does not"
f" exist in either of the directories '{root}' or "
f"'{model.get_config(input_dir)}'."
)
if (
self.get_config(args[0], args[1], key)
== list(branch[args[0]][args[1]].values())[0]
):
raise ValueError("Each model input layers must be unique.")

def check_param_type(self, param, name=None, types=None):
""" """
# def check_uniqueness(model, *args, file_type=None, filename=None):
# """TODO: this function should be checked"""

if not isinstance(param, list):
raise TypeError(
f"The '{name}_lst' parameter should be a of {list}, received a "
f"{type(param)} instead."
)
for i in param:
if not isinstance(i, types):
if isinstance(types, tuple):
types = " or ".join([str(j) for j in types])
else:
types = types
raise TypeError(
f"The '{name}' parameter should be a of {types}, received a "
f"{type(i)} instead."
# args = list(args)
# if len(args) == 1 and "." in args[0]:
# args = args[0].split(".") + args[1:]
# branch = args.pop(-1)
# for key in args[::-1]:
# branch = {key: branch}

# if args[0].get_config(args[0], args[1]):
# for key in model.staticmaps.data_vars:
# if filename == key:
# raise ValueError(
# f"The filenames of the {file_type} maps should be unique."
# )
# if (
# model.get_config(args[0], args[1], key)
# == list(branch[args[0]][args[1]].values())[0]
# ):
# raise ValueError("Each model input layers must be unique.")


def check_uniqueness(model, *args, file_type=None, filename=None):
""" """

args = list(args)
if len(args) == 1 and "." in args[0]:
args = args[0].split(".") + args[1:]
branch = args.pop(-1)
for key in args[::-1]:
branch = {key: branch}

if model.get_config(args[0], args[1]):
for key in model.staticmaps.data_vars:
if filename == key:
raise ValueError(
f"The filenames of the {file_type} maps should be unique."
)
if (
model.get_config(args[0], args[1], key)
== list(branch[args[0]][args[1]].values())[0]
):
raise ValueError(f"Each model input layers must be unique.")


def get_param(self, param_lst, map_fn_lst, file_type, filename, i, param_name):
""" """

if len(param_lst) == 1:
return param_lst[0]
elif len(param_lst) != 1 and len(map_fn_lst) == len(param_lst):
return param_lst[i]
elif len(param_lst) != 1 and len(map_fn_lst) != len(param_lst):
raise IndexError(
f"Could not derive the {param_name} parameter for {file_type} "
f"map: {filename}."
def check_param_type(param, name=None, types=None):
""" """

if not isinstance(param, list):
raise TypeError(
f"The '{name}_lst' parameter should be a of {list}, received a "
f"{type(param)} instead."
)
for i in param:
if not isinstance(i, types):
if isinstance(types, tuple):
types = " or ".join([str(j) for j in types])
else:
types = types
raise TypeError(
f"The '{name}' parameter should be a of {types}, received a "
f"{type(i)} instead."
)




def get_param(param_lst, map_fn_lst, file_type, filename, i, param_name):
""" """

if len(param_lst) == 1:
return param_lst[0]
elif len(param_lst) != 1 and len(map_fn_lst) == len(param_lst):
return param_lst[i]
elif len(param_lst) != 1 and len(map_fn_lst) != len(param_lst):
raise IndexError(
f"Could not derive the {param_name} parameter for {file_type} "
f"map: {filename}."
)
Loading