import pandas as pd
from typing import List
from canopy.core.field import Field
from canopy.core.grid import GridEmpty
from canopy.util.checks import check_spatial_coords_match
[docs]
def unite(fields: List[Field], checks: str = '') -> Field:
"""
Concatenate fields with common layers along the time or spatial axes
The fields must fulfill the following basic requirements:
- Their time series must have the same frequency
- Their grids must be of the same type *and* compatible (see documentation for grid compatibility)
- All fields must have at least one overlapping layer
Parameters
----------
fields : List[Field]
A list of fields to unite
checks : str
A string specifying what checks to run. Each character represents a check:
't': check that fields identical time series
'c': check that fields have contiguous time series
'g': check that all gridcells overlap between fields
'd': (disjoint) check that there are no common gridcells
Returns
-------
A field object with the layers common to all passed fields concatenated along the coordinate axes
Notes
-----
In case of overlapping coordinates/time periods, the first occurrence in concatenation order is kept.
Examples
--------
import canopy as cp
# Uniting a historical run with a scenario run
# --------------------------------------------
aaet_hist = cp.Field.from_file('/path/to/historical/run/aaet.out')
aaet_ssp126 = cp.Field.from_file('/path/to/ssp126/run/aaet.out')
# In this case we might want to check that there are no missing gridcells
# between the 2 runs and that the time series connect with no gaps
aaet_full2 = cp.unite([aaet_hist, aaet_ssp126], chekcs = 'cg')
# Uniting pieces of a simulation broken down into smaller spatial domains
# -----------------------------------------------------------------------
anpp1 = cp.Field.from_file('/path/to/run1/anpp.out')
anpp2 = cp.Field.from_file('/path/to/run2/anpp.out')
anpp3 = cp.Field.from_file('/path/to/run3/anpp.out')
# In this case we might want to check that the time series of all the pieces are the same
# and that the fields do not overlap spatially
anpp = cp.unite([anpp1, anpp2, anpp3], checks='td')
"""
# Required checks
# ---------------
freq_match = [ field.time_freq == fields[0].time_freq for field in fields[1:] ]
if not all(freq_match):
raise ValueError("All fields must have the same time frequency.")
grid_type_match = [ field.grid.grid_type == fields[0].grid.grid_type for field in fields[1:] ]
if not all(grid_type_match):
raise ValueError("Field grids must all be of the same type.")
grid_compatible = [ field.grid.is_compatible(fields[0].grid) for field in fields[1:] ]
if not all(grid_compatible):
raise ValueError("Field grids must all be mutually compatible.")
common_layers = set(fields[0].layers)
for field in fields[1:]:
common_layers = common_layers & set(field.layers)
if len(common_layers) == 0:
raise ValueError("Fields don't have any common layers.")
# Optional checks
# ---------------
requested_checks = set(checks)
available_checks = set('tcgd')
if checks != "" and not requested_checks.issubset(available_checks):
raise ValueError(f"Unrecognized checks: {requested_checks - available_checks}")
if 't' in checks and 'c' in checks:
raise ValueError("Incompatible checks: 't' and 'c'.")
if 'g' in checks and 'd' in checks:
raise ValueError("Incompatible checks: 'g' and 'd'.")
if 't' in checks:
raise NotImplementedError
if 'c' in checks:
raise NotImplementedError
if 'g' in checks:
raise NotImplementedError
if 'd' in checks:
raise NotImplementedError
# Unite
# -----
# Combine all the grids
grid = sum((field.grid for field in fields), start=GridEmpty())
# This is to preserve the order of the 1st field (otherwise set will mess it up)
common_layers_list = [layer for layer in fields[0].layers if layer in common_layers]
# Concatente data along index
data = pd.concat((field.data[common_layers_list] for field in fields))
# Remove rows with duplicated indices
data = data.loc[~data.index.duplicated(keep='first'), :]
field = Field(data, grid, modified=True)
field.log(f"United fields: {[f.description for f in fields]}")
return field