Callbacks

Callbacks used by handlers to transform raw provider dataframes through an ordered pipeline of transformations.

from nbdev.showdoc import show_doc

Foundation

The Transformer class coordinates an ordered pipeline of Callback objects applied to a dict of DataFrames (one per sample type group).

Each callback modifies its group’s DataFrame in sequence. This is how provider-specific conventions are gradually normalised to the MARIS schema.

---

source

Callback

def Callback():

Base class for callbacks

---

source

run_cbs

def run_cbs(
    cbs:list[Callback], # List of callbacks to run
    obj:Any, # Object to pass to the callbacks
):

Run the callbacks in the order they are specified

run_cbs runs the callbacks sorted by order, and adds each callback’s docstring to obj.logs. A callback without a docstring is not logged:

class DocCB(Callback):
    "Runs second"
    order = 1
    def __call__(self, obj): pass

class DocSecondCB(Callback):
    "Runs first"
    order = 0
    def __call__(self, obj): pass

class NoDocCB(Callback):
    def __call__(self, obj): pass

obj = type('Obj', (), {'logs': []})()
run_cbs([DocCB(), NoDocCB(), DocSecondCB()], obj)
test_eq(obj.logs, ['Runs first', 'Runs second'])

---

source

Transformer

def Transformer(
    data:dict[str, pd.DataFrame], # Groups to transform, keyed by group name
    cbs:list[Callback] | None=None, # List of callbacks to run
    inplace:bool=False, # Whether to modify the groups in place
):

Transform groups of dataframes according to the specified callbacks

A callback receives the Transformer and changes the groups in tfm.dfs. IncDepthCB adds 1 to depth in every group:

class IncDepthCB(Callback):
    "Add 1 to `depth` in every group"
    def __call__(self, tfm: Transformer):
        for df in tfm.dfs.values(): df['depth'] += 1

Transformer runs the callbacks on copies of the groups and returns the transformed groups:

dfs = {'biota': pd.DataFrame({'id': [0, 1, 2], 'depth': [2, 3, 4]}), 'seawater': pd.DataFrame({'id': [0, 1, 2], 'depth': [3, 4, 5]})}
tfm = Transformer(dfs, cbs=[IncDepthCB()])
dfs_test = tfm()

test_eq(dfs_test['biota']['depth'].to_list(), [3, 4, 5])
test_eq(dfs_test['seawater']['depth'].to_list(), [4, 5, 6])

Some providers deliver one flat table, and the group of each row is known only after the table has been parsed. Load such a table as a single group, for example 'RAW', transform it, then replace tfm.dfs with the real groups. Transformer gives no meaning to the 'RAW' key. NetCDFEncoder rejects any group name that is not in NC_GROUPS, so the split must happen before encoding. The geotraces handler uses this pattern. SplitByGroupCB splits the 'RAW' group by its GROUP column:

class SplitByGroupCB(Callback):
    "Replace the `'RAW'` group with one group per value of its `GROUP` column"
    def __call__(self, tfm: Transformer): tfm.dfs = {k: v.drop(columns='GROUP') for k, v in tfm.dfs.pop('RAW').groupby('GROUP')}

raw = pd.DataFrame({'GROUP': ['SEAWATER', 'BIOTA', 'SEAWATER'], 'depth': [2, 3, 4]})
res = Transformer({'RAW': raw}, cbs=[IncDepthCB(), SplitByGroupCB()])()
test_eq(sorted(res), ['BIOTA', 'SEAWATER'])
test_eq(res['SEAWATER']['depth'].to_list(), [3, 5])
res

---

source

Transformer.unique

def unique(
    col_name:str, # Column to read in every group
)->np.ndarray: # Sorted distinct values

Distinct non-NA values of col_name across all groups

unique collects the distinct values of a column from every group that has it:

test_eq(Transformer(dfs).unique('depth'), [2, 3, 4, 5])

---

source

PerGroupCB

def PerGroupCB(
    grps:list=None, # Groups to process; None = all groups in `tfm.dfs`
):

Calls each_grp for each group in tfm.dfs; set grps to restrict to specific groups

---

source

PerGroupCB.each_grp

def each_grp(
    grp:str, # Group key e.g. `'SEAWATER'`, `'BIOTA'`
    df:pd.DataFrame, # DataFrame for this group
    tfm, # Parent `Transformer`
):

Override to implement per-group transformation logic

Most callbacks change each group in the same way. For these, subclass PerGroupCB and override each_grp. PerGroupCB loops over the groups and skips any group missing from tfm.dfs. AddFlagCB adds a constant column to every group:

class AddFlagCB(PerGroupCB):
    def __init__(self, col, val): store_attr()
    def each_grp(self, grp, df, tfm): df[self.col] = self.val

dfs = {'SEAWATER': pd.DataFrame({'depth': [1, 2]}), 'BIOTA': pd.DataFrame({'depth': [3, 4]})}

tfm = Transformer(dfs, cbs=[AddFlagCB(col='flag', val=1)])
dfs_result = tfm()
test_eq(dfs_result['SEAWATER']['flag'].to_list(), [1, 1])
test_eq(dfs_result['BIOTA']['flag'].to_list(), [1, 1])

Set grps, as a class attribute or a constructor argument, to process only some groups. Here only BIOTA gets the column:

class BiotaFlagCB(PerGroupCB):
    grps = ['BIOTA']
    def each_grp(self, grp, df, tfm): df['is_biota'] = True

tfm2 = Transformer(dfs, cbs=[BiotaFlagCB()])
dfs_result2 = tfm2()
test_eq('is_biota' in dfs_result2['BIOTA'].columns, True)
test_eq('is_biota' in dfs_result2['SEAWATER'].columns, False)

Cleaning & validation

Callbacks for cleaning and validating data: coordinate sanitisation (invalid lon/lat), and removal of rows with all-NA values in key columns.

---

source

SanitizeLonLatCB

def SanitizeLonLatCB(
    lon_col:str='LON', # Longitude column name
    lat_col:str='LAT', # Latitude column name
    verbose:bool=False, # Whether to print the number of invalid longitude & latitude values
):

Drop rows with invalid longitude & latitude values. Convert , separator to . separator

SanitizeLonLatCB reads a decimal comma as a decimal point. It drops rows at (0, 0), and rows with a longitude outside -180 to 180 or a latitude outside -90 to 90:

dfs = {'BIOTA': pd.DataFrame({'LON': ['45,2', 0, -190, 1.1, 2], 'LAT': ['43,1', 0, 1, 91, 3]})}
res = Transformer(dfs, cbs=[SanitizeLonLatCB()])()['BIOTA']
test_eq(res.values.tolist(), [[45.2, 43.1], [2.0, 3.0]])
res

	LON	LAT
0	45.2	43.1
4	2.0	3.0

---

source

RemoveAllNAValuesCB

def RemoveAllNAValuesCB(
    cols_to_check:dict[str, list] | list, # Dict or list of columns to check
    how:str='all', # How to handle NA values 'all' or 'any'
):

Remove rows with all NA values in specified columns

RemoveAllNAValuesCB drops the rows where every column in cols_to_check is NA. A list names the same columns for every group. Here only row c has neither a value nor an uncertainty:

dfs = {'SEAWATER': pd.DataFrame(dict(value=[1.0, np.nan, np.nan, 4.0], uncertainty=[0.1, 0.2, np.nan, 0.4], meta=['a', 'b', 'c', 'd']))}
res = Transformer(dfs, cbs=[RemoveAllNAValuesCB(['value', 'uncertainty'])])()['SEAWATER']
test_eq(res['meta'].to_list(), ['a', 'b', 'd'])
res

	value	uncertainty	meta
0	1.0	0.1	a
1	NaN	0.2	b
3	4.0	0.4	d

A dict names the columns to check in each group:

dfs = {'SEAWATER': pd.DataFrame({'value': [1.0, np.nan], 'uncertainty': [0.1, np.nan]}),
    'SEDIMENT': pd.DataFrame({'value': [np.nan, 2.0], 'uncertainty': [0.3, 0.2]})}
res = Transformer(dfs, cbs=[RemoveAllNAValuesCB({'SEAWATER': ['value', 'uncertainty'], 'SEDIMENT': ['value']})])()
test_eq({grp: len(df) for grp, df in res.items()}, {'SEAWATER': 1, 'SEDIMENT': 1})

Value mapping

---

source

RemapCB

def RemapCB(
    lut:dict | Callable, # Lookup: dict, or callable(dfs)->dict
    col_remap:str, # Destination column to create
    col_src:str, # Source column with provider values
    default_val:int=0, # Value assigned to unmapped source values
    grps:list[str]=None, # Groups to process (None = all)
):

Remap source values to MARIS standard identifiers using a lookup table

RemapCB maps each value of col_src through the lookup table and writes the result to col_remap. A value missing from the table gets default_val, which is 0 by default:

lut_dict = {'Cs-137': 1, 'K-40': 2, 'Sr-90': 3}
dfs = {'SEAWATER': pd.DataFrame({'nuclide': ['Cs-137', 'K-40', 'Am-241']})}
res = Transformer(dfs, cbs=[RemapCB(lut=lut_dict, col_remap='NUCLIDE_ID', col_src='nuclide')])()['SEAWATER']
test_eq(res['NUCLIDE_ID'].to_list(), [1, 2, 0])
res

	nuclide	NUCLIDE_ID
0	Cs-137	1
1	K-40	2
2	Am-241	0

Set default_val to mark unmapped values differently:

res = Transformer(dfs, cbs=[RemapCB(lut=lut_dict, col_remap='NUCLIDE_ID', col_src='nuclide', default_val=-1)])()['SEAWATER']
test_eq(res['NUCLIDE_ID'].to_list(), [1, 2, -1])

grps limits the remapping to some groups:

dfs = {'SEAWATER': pd.DataFrame({'nuclide': ['Cs-137']}), 'BIOTA': pd.DataFrame({'nuclide': ['K-40']})}
result = Transformer(dfs, cbs=[RemapCB(lut=lut_dict, col_remap='NUCLIDE_ID', col_src='nuclide', grps=['BIOTA'])])()
test_eq('NUCLIDE_ID' in result['SEAWATER'].columns, False)
test_eq(result['BIOTA']['NUCLIDE_ID'].to_list(), [2])

lut can also be a callable that takes the groups and returns the lookup table. RemapCB calls it when the pipeline runs. marisco.match builds such callables:

from marisco.match import make_lut, make_lut_from

make_lut_from uses the provider’s own nomenclature. Here the provider codes h-3 and k-40 match the MARIS names h3 and k40, whose IDs are 1 and 4:

prov_nomencl = pd.DataFrame({'code': ['h-3', 'k-40'], 'maris_ref': ['h3', 'k40']})
nuclide_lut = make_lut_from(prov_nomencl, key_col='code', match_col='maris_ref', lut_key='NUCLIDE')
dfs = {'SEAWATER': pd.DataFrame({'NUCLIDE': ['h-3', 'k-40']})}
res = Transformer(dfs, cbs=[RemapCB(lut=nuclide_lut, col_remap='NUCLIDE_ID', col_src='NUCLIDE')])()['SEAWATER']
test_eq(res['NUCLIDE_ID'].to_list(), [1, 4])

make_lut matches the source values directly against the MARIS names:

dfs = {'SEAWATER': pd.DataFrame({'NUCLIDE': ['h3', 'k40']})}
res = Transformer(dfs, cbs=[RemapCB(lut=make_lut('NUCLIDE'), col_remap='NUCLIDE_ID', col_src='NUCLIDE')])()['SEAWATER']
test_eq(res['NUCLIDE_ID'].to_list(), [1, 4])

---

source

LowerStripNameCB

def LowerStripNameCB(
    col_src:str, # Source column name e.g. 'Nuclide'
    col_dst:str=None, # Destination column name
    fn_transform:Callable=<lambda>, # Transformation function
    grps:list=None, # Groups to process; None = all groups in `tfm.dfs`
):

Convert values to lowercase and strip any trailing spaces

LowerStripNameCB lowercases the values of col_src, strips surrounding spaces, and writes the result to col_dst:

dfs = {'seawater': pd.DataFrame({'Nuclide': ['CS137', '226RA']})}
tfm = Transformer(dfs, cbs=[LowerStripNameCB(col_src='Nuclide', col_dst='NUCLIDE')])
test_eq(tfm()['seawater']['NUCLIDE'].to_list(), ['cs137', '226ra'])

Without col_dst, it changes col_src in place:

tfm = Transformer(dfs, cbs=[LowerStripNameCB(col_src='Nuclide')])
test_eq(tfm()['seawater']['Nuclide'].to_list(), ['cs137', '226ra'])

NA values stay NA. Use grps to skip the groups that lack the source column:

dfs = {'BIOTA': pd.DataFrame({'species': [' Fucus vesiculosus ', None]}), 'SEAWATER': pd.DataFrame({'Nuclide': ['CS137']})}
tfm = Transformer(dfs, cbs=[LowerStripNameCB(col_src='species', col_dst='SPECIES', grps=['BIOTA'])])
tfm()
test_eq(tfm.dfs['BIOTA']['SPECIES'].to_list()[0], 'fucus vesiculosus')
test_eq(tfm.dfs['BIOTA']['SPECIES'].isna().to_list(), [False, True])
test_eq('SPECIES' in tfm.dfs['SEAWATER'].columns, False)

Schema alignment

---

source

AddSampleTypeIdColumnCB

def AddSampleTypeIdColumnCB(
    lut:dict={'SEAWATER': 1, 'BIOTA': 2, 'SEDIMENT': 3, 'SUSPENDED_MATTER': 4}, # Lookup table for sample type
    col_name:str='SAMPLE_TYPE', # Column name to store the sample type id
):

Add a column with the sample type as defined in the CDL

AddSampleTypeIdColumnCB adds the SAMPLE_TYPE column, holding each group’s sample type ID from SMP_TYPE_LUT:

dfs = {smp_type: pd.DataFrame({'col_test': [0, 1, 2]}) for smp_type in SMP_TYPE_LUT}
dfs_test = Transformer(dfs, cbs=[AddSampleTypeIdColumnCB()])()
test_eq({grp: df['SAMPLE_TYPE'].unique().item() for grp, df in dfs_test.items()}, SMP_TYPE_LUT)

---

source

RenameColumnsCB

def RenameColumnsCB(
    renaming_rules:dict, # Renaming rules {old_name: new_name}
):

Rename variables to MARIS standard names, keeping only renamed columns

RenameColumnsCB renames columns to their MARIS names and drops every column not in renaming_rules. Here column c is dropped:

dfs = {'SEAWATER': pd.DataFrame(dict(a=[1, 2], b=[3, 4], c=[5, 6]))}
res = Transformer(dfs, cbs=[RenameColumnsCB({'a': 'ALPHA', 'b': 'BETA'})])()['SEAWATER']
test_eq(list(res.columns), ['ALPHA', 'BETA'])
res

	ALPHA	BETA
0	1	3
1	2	4

Wide-to-long reshaping

MeltWideNuclidesCB converts a provider’s wide format, with one column per nuclide, to the MARIS long format, with one row per measurement. Its spec argument is a list of dicts, one for each nuclide. Each dict gives the source column names and the MARIS IDs to set. To support a new nuclide, add a dict to spec.

---

source

MeltWideNuclidesCB

def MeltWideNuclidesCB(
    spec:list, # List of dicts with keys: val, unc, nuclide, unit, lab
    grp:str='SEAWATER', # Group in tfm.dfs to reshape
):

Reshape wide nuclide columns to long format using a named-dict spec

Each spec entry names a value column and its uncertainty column, and gives the MARIS IDs to set. Rows with no value are dropped. cs137_bq_l has one value and k40_bq_l has two, so the result has three rows:

melt_spec = [  # chkstyle: ignore-node
    {'val': 'cs137_bq_l', 'unc': 'unc_cs137_bq_l', 'nuclide': 1, 'unit': 3, 'lab': 0},
    {'val': 'k40_bq_l',   'unc': 'unc_k40_bq_l',   'nuclide': 4, 'unit': 3, 'lab': 0},
]
dfs = {'SEAWATER': pd.DataFrame({  # chkstyle: ignore-node
    'LAT': [1.0, 2.0], 'LON': [10.0, 20.0],
    'cs137_bq_l':     [0.5, np.nan], 'unc_cs137_bq_l': [0.1, np.nan],
    'k40_bq_l':       [100.0, 200.0], 'unc_k40_bq_l':  [5.0, 10.0],
})}
res = Transformer(dfs, cbs=[MeltWideNuclidesCB(spec=melt_spec)])()['SEAWATER']
test_eq(res['NUCLIDE'].to_list(), [1, 4, 4])
test_eq(res['VALUE'].to_list(), [0.5, 100.0, 200.0])
res

	LAT	LON	cs137_bq_l	unc_cs137_bq_l	k40_bq_l	unc_k40_bq_l	NUCLIDE	VALUE	UNC	UNIT	LAB
0	1.0	10.0	0.5	0.1	100.0	5.0	1	0.5	0.1	3	0
1	1.0	10.0	0.5	0.1	100.0	5.0	4	100.0	5.0	3	0
2	2.0	20.0	NaN	NaN	200.0	10.0	4	200.0	10.0	3	0

MeltWideNuclidesCB does nothing when the data has no grp group:

res = Transformer({'BIOTA': pd.DataFrame({'x': [1]})}, cbs=[MeltWideNuclidesCB(spec=melt_spec)])()
test_eq(res['BIOTA'].to_dict('list'), {'x': [1]})

Sample IDs

AddSampleIDCB numbers the rows of each group in SMP_ID, starting at 1. Run it after any callback that adds or removes rows, such as MeltWideNuclidesCB. It can also convert a provider ID column to str, which NetCDF needs for variable-length strings.

---

source

AddSampleIDCB

def AddSampleIDCB(
    col_provider:str=None, # Provider ID column to cast to str; None = skip
):

Assign 1-based sequential SMP_ID; optionally cast a provider ID column to str for NetCDF VLEN compatibility

With no arguments, AddSampleIDCB only adds SMP_ID:

dfs = {'SEAWATER': pd.DataFrame({'val': [10, 20, 30]})}
res = Transformer(dfs, cbs=[AddSampleIDCB()])()['SEAWATER']
test_eq(res['SMP_ID'].to_list(), [1, 2, 3])
res

	val	SMP_ID
0	10	1
1	20	2
2	30	3

col_provider names the provider ID column to convert. A float column whose values are all whole numbers loses the .0, so 289.0 becomes '289'. Other values convert as they are. A group without the column only gets SMP_ID:

dfs = {'SEAWATER': pd.DataFrame({'PROVIDER_ID': [289.0, 1209.0]}), 'BIOTA': pd.DataFrame({'PROVIDER_ID': [289.5]}),
    'SEDIMENT': pd.DataFrame({'val': [1.0]})}
res = Transformer(dfs, cbs=[AddSampleIDCB(col_provider='PROVIDER_ID')])()
test_eq(res['SEAWATER']['PROVIDER_ID'].to_list(), ['289', '1209'])
test_eq(res['BIOTA']['PROVIDER_ID'].to_list(), ['289.5'])
test_eq(list(res['SEDIMENT'].columns), ['val', 'SMP_ID'])

---

source

UniqueIndexCB

def UniqueIndexCB(
    index_name:str='ID'
):

Set unique index for each group

UniqueIndexCB adds an ID column that numbers the rows of each group from 0:

dfs = {'SEAWATER': pd.DataFrame({'val': [10, 20, 30]}), 'SEDIMENT': pd.DataFrame({'val': [100, 200]})}
res = Transformer(dfs, cbs=[UniqueIndexCB()])()
test_eq(res['SEAWATER']['ID'].to_list(), [0, 1, 2])
test_eq(res['SEDIMENT']['ID'].to_list(), [0, 1])

Time

Callbacks for parsing, encoding, and decoding time columns to/from NetCDF-compatible numeric values.

---

source

ParseTimeCB

def ParseTimeCB(
    time_col_name:str='TIME'
):

Parse time column from ISO8601 string to datetime

ParseTimeCB parses the ISO 8601 strings in the time column to datetimes:

dfs_test = {'SEAWATER': pd.DataFrame({'TIME': ['2023-01-01T00:00:00', '2023-06-15T12:30:00']}),
    'BIOTA': pd.DataFrame({'TIME': ['2010-03-22T08:00:00']})}
tfm = Transformer(dfs_test, cbs=[ParseTimeCB()])
dfs_result = tfm()
test_eq(dfs_result['SEAWATER']['TIME'].dtype.kind, 'M')
test_eq(dfs_result['BIOTA']['TIME'].dtype.kind, 'M')

---

source

EncodeTimeCB

def EncodeTimeCB(
    col_time:str='TIME', # Time column name
    verbose:bool=False, # Print warning about missing time values
    fn_units:Callable=get_time_units, # Function returning the time units
):

Encode time as seconds since epoch

EncodeTimeCB converts each time to a number in the unit that fn_units returns. With seconds since 1970-01-01, midnight on 1 January 2023 is 1672531200. Rows without a time are dropped:

units = 'seconds since 1970-01-01 00:00:00.0'
dfs_test = {'SEAWATER': pd.DataFrame({'TIME': pd.to_datetime(['2023-01-01', '2023-01-02', None]), 'value': [1, 2, 3]})}
res = Transformer(dfs_test, cbs=[EncodeTimeCB(fn_units=lambda: units)])()['SEAWATER']
test_eq(res['TIME'].to_list(), [1672531200, 1672617600])
res

	TIME	value
0	1672531200	1
1	1672617600	2

---

source

DecodeTimeCB

def DecodeTimeCB(
    col_time:str='TIME', # Time column name
    verbose:bool=False, # Print warning about missing time values
    fn_units:Callable=get_time_units, # Function returning the time units
):

Decode time from seconds since epoch to datetime format

DecodeTimeCB converts the numbers back to datetimes:

dfs_test = {'SEAWATER': pd.DataFrame({'TIME': [1672531200, 1672617600], 'value': [1, 2]})}
res = Transformer(dfs_test, cbs=[DecodeTimeCB(fn_units=lambda: units)])()['SEAWATER']
test_eq(res['TIME'].to_list(), pd.to_datetime(['2023-01-01', '2023-01-02']).to_list())
res

	TIME	value
0	2023-01-01	1
1	2023-01-02	2