import contextlib
import inspect
import os
import sys
import textwrap
from collections import defaultdict
from functools import partial
import numpy as np
import rich_click as click
from landlab import (
FramedVoronoiGrid,
HexModelGrid,
ModelGrid,
RadialModelGrid,
RasterModelGrid,
VoronoiDelaunayGrid,
)
GRIDS = [
ModelGrid,
RasterModelGrid,
VoronoiDelaunayGrid,
HexModelGrid,
RadialModelGrid,
FramedVoronoiGrid,
]
CATEGORIES = {
"boundary-condition",
"connectivity",
"deprecated",
"field-add",
"field-io",
"gradient",
"info-cell",
"info-corner",
"info-face",
"info-field",
"info-grid",
"info-link",
"info-node",
"info-patch",
"map",
"quantity",
"subset",
"surface",
"uncategorized",
}
click.rich_click.ERRORS_SUGGESTION = (
"Try running the '--help' flag for more information."
)
click.rich_click.ERRORS_EPILOGUE = (
"To find out more, visit https://github.com/landlab/landlab"
)
click.rich_click.STYLE_ERRORS_SUGGESTION = "yellow italic"
click.rich_click.SHOW_ARGUMENTS = True
click.rich_click.GROUP_ARGUMENTS_OPTIONS = False
click.rich_click.SHOW_METAVARS_COLUMN = True
click.rich_click.USE_MARKDOWN = True
out = partial(click.secho, bold=True, file=sys.stderr)
err = partial(click.secho, fg="red", file=sys.stderr)
@click.group() # chain=True)
@click.version_option()
@click.option(
"--cd",
default=".",
type=click.Path(exists=True, file_okay=False, dir_okay=True, readable=True),
help="chage to directory, then execute",
)
@click.option(
"-s",
"--silent",
is_flag=True,
help="Suppress status status messages, including the progress bar.",
)
@click.option(
"-v", "--verbose", is_flag=True, help="Also emit status messages to stderr."
)
def landlab(cd, silent, verbose) -> None:
os.chdir(cd)
@landlab.command(name="list")
def _list():
for cls in get_all_components():
print(cls.__name__)
@landlab.command()
@click.argument("component", type=str, nargs=-1)
def used_by(component):
for name in _used_by(get_components(component)):
print(name)
@landlab.command()
@click.argument("component", type=str, nargs=-1)
def provided_by(component):
for name in _provided_by(get_components(component)):
print(name)
@landlab.command()
@click.argument("var", type=str)
def uses(var):
for name in get_users_of(var):
print(name)
@landlab.command()
@click.argument("var", type=str)
def provides(var):
for name in get_providers_of(var):
print(name)
@landlab.command()
@click.argument("component", type=str, nargs=-1)
def validate(component):
failures = 0
classes = get_components(component)
for cls in classes:
out(cls.__name__)
errors = _validate_component(cls)
if errors:
failures += 1
for error in errors:
err(f"Error: {cls.__name__}: {error}")
if failures:
click.Abort()
else:
out("💥 All good! 💥")
@landlab.group(chain=True)
@click.pass_context
def index(ctx):
pass
@index.command()
@click.pass_context
def grids(ctx):
verbose = ctx.parent.parent.params["verbose"]
silent = ctx.parent.parent.params["silent"]
index = {"grids": {}}
for cls in GRIDS:
index["grids"][cls.__name__] = _categorize_class(cls)
index["grids"][cls.__name__]["field-io"] += [
f"{cls.__module__}.{cls.__name__}.at_node",
f"{cls.__module__}.{cls.__name__}.at_link",
f"{cls.__module__}.{cls.__name__}.at_patch",
f"{cls.__module__}.{cls.__name__}.at_corner",
f"{cls.__module__}.{cls.__name__}.at_face",
f"{cls.__module__}.{cls.__name__}.at_cell",
]
print("# Generated using `landlab index grids`")
for grid, cats in index["grids"].items():
print(f"[grids.{grid}]")
for cat, funcs in cats.items():
print(f"{cat} = [")
print(
textwrap.indent(
os.linesep.join([repr(f) + "," for f in sorted(funcs)]), " "
)
)
print("]")
print("")
if verbose and not silent:
summary = defaultdict(int)
for cats in index["grids"].values():
for cat, funcs in cats.items():
summary[cat] += len(funcs)
out("[summary]")
out(f"grids = [{', '.join(sorted(index['grids']))}]")
out(f"entries = {sum(summary.values())}")
out("")
out("[summary.categories]")
for cat in sorted(summary):
out(f"{cat} = {summary[cat]}")
@index.command()
@click.pass_context
def components(ctx):
verbose = ctx.parent.parent.params["verbose"]
silent = ctx.parent.parent.params["silent"]
from sphinx.util.docstrings import prepare_docstring
index = {"components": {}}
for cls in get_all_components():
if verbose and not silent:
out(f"indexing: {cls.__name__}")
index["components"][cls.__name__] = {
"name": f"{cls.__module__}.{cls.__name__}",
"unit_agnostic": cls._unit_agnostic,
"info": cls._info,
"summary": prepare_docstring(cls.__doc__)[0],
}
print("# Generated using `landlab index components`")
for component, info in index["components"].items():
print("")
print(f"[components.{component}]")
print(f"name = {info['name']!r}")
print(f"unit_agnostic = {'true' if info['unit_agnostic'] else 'false'}")
print(f"summary = {info['summary']!r}")
for name, values in info["info"].items():
print("")
print(f"[components.{component}.info.{name}]")
print(f"dtype = {str(np.dtype(values['dtype']))!r}")
print(f"intent = {values['intent']!r}")
print(f"optional = {'true' if values['optional'] else 'false'}")
print(f"units = {values['units']!r}")
print(f"mapping = {values['mapping']!r}")
print(f"doc = {values['doc']!r}")
if not silent:
out("[summary]")
out(f"count = {len(index['components'])}")
@index.command()
@click.pass_context
def fields(ctx):
verbose = ctx.parent.parent.params["verbose"]
silent = ctx.parent.parent.params["silent"]
fields = defaultdict(lambda: defaultdict(list))
for cls in get_all_components():
if verbose and not silent:
out(f"checking {cls.__name__}... {len(cls._info)} fields")
for name, desc in cls._info.items():
fields[name]["desc"].append(desc["doc"])
if desc["intent"].startswith("in"):
fields[name]["used_by"].append(f"{cls.__module__}.{cls.__name__}")
if desc["intent"].endswith("out"):
fields[name]["provided_by"].append(f"{cls.__module__}.{cls.__name__}")
print("# Generated using `landlab index fields`")
print("[fields]")
for field, info in fields.items():
print("")
print(f"[fields.{field}]")
print(f"desc = {info['desc'][0]!r}")
if info["used_by"]:
# used_by = [repr(f) for f in info["used_by"]]
# print(f"used_by = [{', '.join(used_by)}]")
print("used_by = [")
for component in info["used_by"]:
print(f" {component!r},")
print("]")
else:
print("used_by = []")
if info["provided_by"]:
print("provided_by = [")
for component in info["provided_by"]:
print(f" {component!r},")
print("]")
else:
print("provided_by = []")
if not silent:
out("[summary]")
out(f"count = {len(fields)}")
[docs]def get_all_components():
from landlab.components import COMPONENTS
from landlab.core.model_component import Component
components = []
for cls in COMPONENTS:
if issubclass(cls, Component):
components.append(cls)
return components
[docs]def get_all_components_by_name():
return {cls.__name__: cls for cls in get_all_components()}
[docs]def get_components(*args):
"""Get components by name.
Parameters
----------
names : list of str, optional
Component names.
Returns
-------
list of class
Components with any of the given names.
"""
if len(args) == 0 or len(args[0]) == 0:
components = get_all_components()
else:
components_by_name = get_all_components_by_name()
components = []
for name in args[0]:
try:
components.append(components_by_name[name])
except KeyError:
print(f"{name}: not a component", file=sys.stderr)
return components
[docs]def get_users_of(var):
"""Get components that use a variable."""
users = []
for cls in get_all_components():
try:
if var in cls.input_var_names:
users.append(cls.__name__)
except (AttributeError, TypeError):
print(
f"Warning: {cls.__name__}: unable to get input vars",
file=sys.stderr,
)
return users
[docs]def get_providers_of(var):
"""Get components that provide a variable."""
providers = []
for cls in get_all_components():
try:
if var in cls.output_var_names:
providers.append(cls.__name__)
except (AttributeError, TypeError):
print(
f"Warning: {cls.__name__}: unable to get output vars",
file=sys.stderr,
)
return providers
def _used_by(classes):
"""Get variables used by components."""
used = []
for cls in classes:
with contextlib.suppress(TypeError):
used += cls.input_var_names
return used
def _provided_by(classes):
"""Get variables provided by components."""
provided = []
for cls in classes:
with contextlib.suppress(TypeError):
provided += cls.output_var_names
return provided
def _test_input_var_names(cls):
errors = []
try:
names = cls.input_var_names
except AttributeError:
errors.append("no input_var_names attribute")
else:
if not isinstance(names, tuple):
errors.append("input_var_names is not a tuple")
return errors
def _test_output_var_names(cls):
errors = []
try:
names = cls.output_var_names
except AttributeError:
errors.append("no output_var_names attribute")
else:
if not isinstance(names, tuple):
errors.append("output_var_names is not a tuple")
return errors
def _validate_component(cls):
from landlab.core.model_component import Component
errors = []
if not issubclass(cls, Component):
errors.append("not a subclass of Component")
else:
errors += _test_input_var_names(cls)
errors += _test_output_var_names(cls)
return errors
def _validate(args):
failures = 0
classes = get_components(args.name)
for cls in classes:
errors = _validate_component(cls)
if errors:
failures += 1
for error in errors:
print(f"Error: {cls.__name__}: {error}")
return failures
def _categorize_class(cls):
funcs = {cat: [] for cat in CATEGORIES}
for name, func in inspect.getmembers(cls):
if not name.startswith("_"):
full_name = ".".join([cls.__module__, cls.__name__, name])
for cat in _extract_landlab_category(inspect.getdoc(func)):
funcs[cat].append(full_name)
return funcs
def _extract_landlab_category(s: str):
from sphinx.util.docstrings import separate_metadata
return [
cat.strip() or "uncategorized"
for cat in separate_metadata(s)[1].get("landlab", "").split(",")
]