from canopy.core.field import Field
from typing import List
import pandas as pd
from canopy.core.grid import GridEmpty
[docs]
def join(fields: List[Field], checks: str = '') -> Field:
"""
Join two or more fields with non-overlapping layers
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)
- Fields must have no overlapping layers
Entries with non-overlapping spatio-temporal coordinates are completed with NaNs.
Parameters
----------
fields : List[Field]
A list of fields to join
checks : str
A string specifying what checks to run. Each character represents a check:
't': check that fields have identical time series
'g': check that all gridcells overlap between fields
'i': check that indices are identical (supersedes both checks above)
Returns
-------
A unique field object with the layers of all passed fields
"""
# Required checks
# ---------------
freq_match = all( field.time_freq == fields[0].time_freq for field in fields[1:] )
if not freq_match:
raise ValueError("All fields must have the same time frequency.")
grid_type_match = all( field.grid.grid_type == fields[0].grid.grid_type for field in fields[1:] )
if not grid_type_match:
raise ValueError("Field grids must all be of the same type.")
grid_compatible = all( field.grid.is_compatible(fields[0].grid) for field in fields[1:] )
if not grid_compatible:
raise ValueError("Field grids must all be mutually compatible.")
all_layers = []
for field in fields:
all_layers.extend(field.layers)
if len(all_layers) != len(set(all_layers)):
raise ValueError("Fields have common layers.")
# Optional checks
# ---------------
if 't' in checks:
raise NotImplementedError
if 'c' in checks:
raise NotImplementedError
if 'i' in checks:
raise NotImplementedError
# Combine all the grids
grid = sum((field.grid for field in fields), start=GridEmpty())
# Join
data = pd.concat([field.data for field in fields], axis=1)
field = Field(data, grid, modified=True)
field.log(f"Joined fields: {[f.description for f in fields]}")
return field