MARIS Legacy

This handler reads the MARIS master database dump, a tab-separated text file that holds every legacy dataset, and writes one MARIS NetCDF4 file for each reference (ref_id). The dump already uses MARIS lookup-table ids, so this handler needs no nomenclature reconciliation, unlike provider handlers such as helcom or geotraces. Its pipeline renames columns, fixes types and encodes:

Rename the dump’s columns to MARIS standard names.
Cast STATION to a string.
Drop columns that are empty or hold only the ‘Not available’ id 0.
Map detection-limit symbols, such as <, to MARIS ids.
Parse dates and encode them as seconds since 1970-01-01.
Fix decimal commas in coordinates and drop out-of-range values.
Cast the sample ids to the types the NetCDF template needs.

encode then adds global attributes, such as the bounding box, time range and Zotero citation, and writes each reference to its own file.

import numpy as np

Configuration & file paths

status is the handler status that marisco-ingest lists. fname_in is the path to the MARIS dump on the maintainer’s machine. The dump isn’t in the repository, so the cells that read it are marked eval: false.

Exported source

status = 'Active'

fname_in = Path.home()/'pro/data/maris/2025-06-03 MARIS_QA_shapetype_id = 1.txt'

Load data

DataLoader reads the dump once. Called with a ref_id, it returns one DataFrame for each sample type of that reference. lut_smp_type maps the dump’s samptype values to MARIS group names, and the loader leaves out any sample type that isn’t in it.

source

DataLoader

def DataLoader(
    fname:str, # Path to the MARIS global dump CSV
    exclude_ref_id:list[int]=None, # ref_ids to skip (None = skip none)
):

Load MARIS dump data filtered by ref_id, returning one DataFrame per sample type group

Exported source

lut_smp_type = {  # chkstyle: ignore-node
    'Biota': 'BIOTA',
    'Seawater': 'SEAWATER',
    'Sediment': 'SEDIMENT',
    'Suspended matter': 'SUSPENDED_MATTER'}

The examples below use the OSPAR dataset, ref_id=191. HELCOM is ref_id=100.

dataloader = DataLoader(fname_in)
dfs = dataloader(ref_id=191)

for grp, grpdf in dfs.items():
    cols = ', '.join(grpdf.columns[:6])
    print(f"{grp:15s} ({len(grpdf):>5} rows, {len(grpdf.columns)} cols)  {cols} ...")

BIOTA           (14164 rows, 80 cols)  sample_id, area_id, areaname, samptype_id, samptype, ref_id ...
SEAWATER        (14848 rows, 80 cols)  sample_id, area_id, areaname, samptype_id, samptype, ref_id ...

print('Full list of seawater dataframe columns: \n', dfs['SEAWATER'].columns.to_list())

Full list of seawater dataframe columns: 
 ['sample_id', 'area_id', 'areaname', 'samptype_id', 'samptype', 'ref_id', 'displaytext', 'zoterourl', 'ref_note', 'datbase', 'lab_id', 'lab', 'latitude', 'longitude', 'begperiod', 'endperiod', 'samplingyear', 'totdepth', 'sampdepth', 'station', 'samplabcode', 'species_id', 'taxonname', 'taxonrank', 'biogroup', 'biogroup_id', 'taxondb', 'taxondbid', 'taxondburl', 'taxonrepname', 'bodypar_id', 'bodypar', 'sliceup', 'slicedown', 'sedtype_id', 'sedtype', 'sedrepname', 'nuclide_id', 'nusymbol', 'volume', 'salinity', 'temperatur', 'filtered', 'filtpore', 'samparea', 'drywt', 'wetwt', 'percentwt', 'sampmet_id', 'sampmet', 'prepmet_id', 'prepmet', 'drymet_id', 'drymet', 'counmet_id', 'counmet', 'decayedto', 'detection', 'activity', 'uncertaint', 'unit_id', 'unit', 'vartype', 'freq', 'rangelow', 'rangeupp', 'profile', 'transect_id', 'measure_note', 'shapetype_id', 'profile_id', 'sampnote', 'ref_fulltext', 'ref_yearpub', 'ref_sampleTypes', 'LongLat', 'shiftedcoordinates', 'shiftedlong', 'shiftedlat', 'id']

Pipeline steps

Column renaming

The dump uses its own column names, such as area_id, activity and station. cois_renaming_rules maps them to MARIS standard names, and RenameColumnsCB keeps only the renamed columns. The mapping is maintained by hand, because the dump schema and the NetCDF schema are independent. NC_CSV in nbs/api/configs.ipynb lists the MARIS names.

Exported source

cois_renaming_rules = {  # chkstyle: ignore-node
    'sample_id': 'SMP_ID',
    'samplabcode': 'SMP_ID_PROVIDER',
    'latitude': 'LAT',
    'longitude': 'LON',
    'begperiod': 'TIME',
    'sampdepth': 'SMP_DEPTH',
    'totdepth': 'TOT_DEPTH',
    'station': 'STATION',
    'uncertaint': 'UNC',
    'unit_id': 'UNIT',
    'detection': 'DL',
    'area_id': 'AREA',
    'species_id': 'SPECIES',
    'biogroup_id': 'BIO_GROUP',
    'bodypar_id': 'BODY_PART',
    'sedtype_id': 'SED_TYPE',
    'volume': 'VOL',
    'salinity': 'SAL',
    'temperatur': 'TEMP',
    'sampmet_id': 'SAMP_MET',
    'prepmet_id': 'PREP_MET',
    'counmet_id': 'COUNT_MET',
    'activity': 'VALUE',
    'nuclide_id': 'NUCLIDE',
    'sliceup': 'TOP',
    'slicedown': 'BOTTOM'}

tfm = Transformer(dfs, cbs=[RenameColumnsCB(cois_renaming_rules)])

dfs_tfm = tfm()
print('Keys:', dfs_tfm.keys())
print('Columns:', dfs_tfm['SEAWATER'].columns)

Keys: dict_keys(['BIOTA', 'SEAWATER'])
Columns: Index(['SMP_ID', 'SMP_ID_PROVIDER', 'LAT', 'LON', 'TIME', 'SMP_DEPTH',
       'TOT_DEPTH', 'STATION', 'UNC', 'UNIT', 'DL', 'AREA', 'SPECIES',
       'BIO_GROUP', 'BODY_PART', 'SED_TYPE', 'VOL', 'SAL', 'TEMP', 'SAMP_MET',
       'PREP_MET', 'COUNT_MET', 'VALUE', 'NUCLIDE', 'TOP', 'BOTTOM'],
      dtype='str')

STATION as string type

The STATION column in the MARIS dump can hold mixed or numeric types, but the NetCDF template declares station as a variable-length string. CastStationToStringCB casts it to the string dtype and fills missing values with an empty string:

source

CastStationToStringCB

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

Convert STATION column to string type, filling any missing values with empty string

dfs_mock = {'SEAWATER': pd.DataFrame({'STATION': ['A1', None, 42]}), 'BIOTA': pd.DataFrame({'STATION': ['B2', None]})}
tfm = Transformer(dfs_mock, cbs=[CastStationToStringCB()])
tfm()

test_eq(tfm.dfs['SEAWATER']['STATION'].dtype.name, 'string')
test_eq(tfm.dfs['SEAWATER']['STATION'].isna().sum(), 0)
test_eq(tfm.dfs['SEAWATER']['STATION'].to_list(), ['A1', '', '42'])
test_eq(tfm.dfs['BIOTA']['STATION'].to_list(), ['B2', ''])

tfm = Transformer(dfs, cbs=[  # chkstyle: ignore-node
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB()
])

dfs_tfm = tfm()
print(dfs_tfm['SEAWATER']['STATION'].dtype)

string

Drop all-empty columns

Some columns in the MARIS dump are entirely empty or contain only ‘Not available’ markers (id=0 in MARIS lookup tables). DropNAColumnsCB removes these columns from every group before further processing. It keeps the columns in keep, by default SMP_ID_PROVIDER. When a dataset has no provider sample ids, CastSampleIDCB fills that column with empty strings:

source

DropNAColumnsCB

def DropNAColumnsCB(
    na_value:int=0, # MARIS NA id to drop
    keep:tuple=('SMP_ID_PROVIDER',), # Columns to keep even when empty
):

Drop columns containing only NaN or ‘Not available’ (id=0 in MARIS lookup tables), except those in keep

df = pd.DataFrame(dict(STATION=['A1', 'B2'], SMP_ID_PROVIDER=[np.nan]*2, EMPTY=[np.nan]*2, MARIS_NA=[0, 0], VALUE=[1.0, 2.0]))
tfm = Transformer({'SEAWATER': df}, cbs=[DropNAColumnsCB()])
tfm()

test_eq(list(tfm.dfs['SEAWATER'].columns), ['STATION', 'SMP_ID_PROVIDER', 'VALUE'])

tfm = Transformer(dfs, cbs=[  # chkstyle: ignore-node
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB(),
    DropNAColumnsCB()
])

dfs_tfm = tfm()
print('Columns:', list(dfs_tfm['SEAWATER'].columns))

Columns: ['SMP_ID', 'SMP_ID_PROVIDER', 'LAT', 'LON', 'TIME', 'SMP_DEPTH', 'STATION', 'UNC', 'UNIT', 'DL', 'AREA', 'VALUE', 'NUCLIDE']

Remap detection limit values

The detection column holds detection-limit symbols as strings, such as <, = and ND. The MARIS NetCDF format stores them as integer ids from the dbo_detectlimit lookup table. RemapCB maps each symbol to its id with the table from get_lut. A value that isn’t in the table becomes 0 (Not Available).

FEEDBACK TO MARIS DATA TEAM

Future MARIS dump exports should provide the detection lut integer id directly rather than its symbolic representation, removing the need for this remapping step.

Exported source

lut_dl = get_lut('DL', key='name', value='id')

lut_dl

{'Not applicable': -1, 'Not Available': 0, '=': 1, '<': 2, 'ND': 3, 'DE': 4}

dfs_mock = {'SEAWATER': pd.DataFrame({'DL': ['=', '<', 'ND', 'DE', None]}), 'BIOTA': pd.DataFrame({'DL': ['=', 'ND', None]})}
tfm = Transformer(dfs_mock, cbs=[RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0)])
tfm()

test_eq(list(tfm.dfs['SEAWATER']['DL']), [1, 2, 3, 4, 0])

tfm = Transformer(dfs, cbs=[  # chkstyle: ignore-node
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB(),
    DropNAColumnsCB(),
    RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0)
])

dfs_tfm = tfm()
print('DL values present:', sorted(dfs_tfm['BIOTA']['DL'].unique()))

DL values present: [np.int64(1), np.int64(2)]

Parse and encode time

MARIS NetCDF files store time as an integer number of seconds since 1970-01-01 00:00:00.0, as nbs/files/cdl/maris.cdl defines. ParseTimeCB parses the date strings in TIME, and EncodeTimeCB converts them to seconds. A missing date is dropped:

tfm = Transformer({'SEAWATER': pd.DataFrame({'TIME': ['1990-01-01', None]})}, cbs=[ParseTimeCB(), EncodeTimeCB()])
tfm()

test_eq(list(tfm.dfs['SEAWATER']['TIME']), [631152000])

Sanitize coordinates

Raw coordinates in the MARIS dump may use commas as decimal separators instead of periods, or fall outside valid lat/lon ranges. SanitizeLonLatCB converts , to . and drops rows with out-of-range values.

df = pd.DataFrame(dict(LAT=[57.25, 91.0, '57,250'], LON=[12.08, 181.0, '12,083']))
tfm = Transformer({'SEAWATER': df}, cbs=[SanitizeLonLatCB()])
tfm()

test_eq(list(tfm.dfs['SEAWATER']['LAT']), [57.25, 57.25])
test_eq(list(tfm.dfs['SEAWATER']['LON']), [12.08, 12.083])

Add sample ids

The dump provides sample_id, an internal MARIS sequential id, and samplabcode, the provider’s original sample identifier. The renaming step maps them to SMP_ID and SMP_ID_PROVIDER. CastSampleIDCB then casts SMP_ID to an integer and SMP_ID_PROVIDER to a variable-length string, and fills missing provider ids with an empty string:

source

CastSampleIDCB

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

Cast SMP_ID to int and SMP_ID_PROVIDER to string (renamed from samplabcode in the pipeline)

df = pd.DataFrame(dict(SMP_ID=[1, 2], SMP_ID_PROVIDER=[None, 'RC1']))
tfm = Transformer({'SEAWATER': df}, cbs=[CastSampleIDCB()])
tfm()

test_eq(tfm.dfs['SEAWATER']['SMP_ID'].dtype, int)
test_eq(tfm.dfs['SEAWATER']['SMP_ID_PROVIDER'].to_list(), ['', 'RC1'])

tfm = Transformer(dfs, cbs=[  # chkstyle: ignore-node
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB(),
    DropNAColumnsCB(),
    RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0),
    ParseTimeCB(),
    EncodeTimeCB(),
    SanitizeLonLatCB(),
    CastSampleIDCB()
])
dfs_tfm = tfm()
print(dfs_tfm['SEAWATER'][['SMP_ID', 'SMP_ID_PROVIDER']].head(3).to_string(index=False))

 SMP_ID SMP_ID_PROVIDER
 332133                
 332133                
 332133

Encode to NetCDF

get_cbs returns the pipeline’s callbacks, in order. encode and the examples below use it, so the pipeline is defined once.

source

get_cbs

def get_cbs()->list:

Callbacks, in pipeline order, that turn MARIS dump data into MARIS-standard DataFrames

Exported source

def get_cbs() -> list:
    "Callbacks, in pipeline order, that turn MARIS dump data into MARIS-standard DataFrames"
    return [  # chkstyle: ignore-node
        RenameColumnsCB(cois_renaming_rules),
        CastStationToStringCB(),
        DropNAColumnsCB(),
        RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0),
        ParseTimeCB(),
        EncodeTimeCB(),
        SanitizeLonLatCB(),
        CastSampleIDCB()]

The Transformer adds each callback’s docstring to tfm.logs. encode writes them to the global attribute publisher_postprocess_logs:

tfm = Transformer(dfs, cbs=get_cbs())
tfm()
tfm.logs

['Rename variables to MARIS standard names, keeping only renamed columns',
 'Convert STATION column to string type, filling any missing values with empty string',
 "Drop columns containing only NaN or 'Not available' (id=0 in MARIS lookup tables), except those in `keep`",
 "Remap values from 'DL' to 'DL' for groups: all.",
 'Parse time column from ISO8601 string to datetime',
 'Encode time as seconds since epoch',
 'Drop rows with invalid longitude & latitude values. Convert `,` separator to `.` separator',
 'Cast SMP_ID to int and SMP_ID_PROVIDER to string (renamed from samplabcode in the pipeline)']

kw lists the keywords that encode writes to every file:

source

get_zotero_key

def get_zotero_key(
    dfs:dict, # Dict of {group_name: DataFrame} per sample type
)->str: # Zotero key extracted from URL

Extract Zotero bibliography key from the MARIS dump DataFrame

Exported source

kw = [  # chkstyle: ignore-node
    'oceanography', 'Earth Science > Oceans > Ocean Chemistry > Radionuclides',
    'Earth Science > Human Dimensions > Environmental Impacts > Nuclear Radiation Exposure',
    'Earth Science > Oceans > Ocean Chemistry > Ocean Tracers, Earth Science > Oceans > Marine Sediments',
    'Earth Science > Oceans > Ocean Chemistry, Earth Science > Oceans > Sea Ice > Isotopes',
    'Earth Science > Oceans > Water Quality > Ocean Contaminants',
    'Earth Science > Biological Classification > Animals/Vertebrates > Fish',
    'Earth Science > Biosphere > Ecosystems > Marine Ecosystems',
    'Earth Science > Biological Classification > Animals/Invertebrates > Mollusks',
    'Earth Science > Biological Classification > Animals/Invertebrates > Arthropods > Crustaceans',
    'Earth Science > Biological Classification > Plants > Macroalgae (Seaweeds)']

source

get_fname

def get_fname(
    dfs:dict, # Dict of {group_name: DataFrame} per sample type
)->str: # NetCDF filename like "12345.nc"

Construct NetCDF filename from the ref_id in the data

get_fname and get_zotero_key read a reference’s output file name and Zotero key from the ref_id and zoterourl columns. These are the values for OSPAR:

dfs_ospar = {'SEAWATER': pd.DataFrame({'ref_id': [191], 'zoterourl': ['https://www.zotero.org/groups/2432820/maris/items/LQRA4MMK']})}
test_eq(get_fname(dfs_ospar), '191.nc')
test_eq(get_zotero_key(dfs_ospar), 'LQRA4MMK')

source

get_attrs

def get_attrs(
    tfm:marisco.callbacks.Transformer, # Transformer after running the pipeline
    zotero_key:str, # Zotero key of the reference
    kw:list=['oceanography', 'Earth Science > Oceans > Ocean Chemistry > Radionuclides', 'Earth Science > Human Dimensions > Environmental Impacts > Nuclear Radiation Exposure', 'Earth Science > Oceans > Ocean Chemistry > Ocean Tracers, Earth Science > Oceans > Marine Sediments', 'Earth Science > Oceans > Ocean Chemistry, Earth Science > Oceans > Sea Ice > Isotopes', 'Earth Science > Oceans > Water Quality > Ocean Contaminants', 'Earth Science > Biological Classification > Animals/Vertebrates > Fish', 'Earth Science > Biosphere > Ecosystems > Marine Ecosystems', 'Earth Science > Biological Classification > Animals/Invertebrates > Mollusks', 'Earth Science > Biological Classification > Animals/Invertebrates > Arthropods > Crustaceans', 'Earth Science > Biological Classification > Plants > Macroalgae (Seaweeds)'], # Keywords
)->dict: # Global attributes

Retrieve global attributes from MARIS dump

get_attrs builds the global attributes of one file: bounding box, depth and time ranges, the Zotero citation, keywords and processing logs. ZoteroCB calls the Zotero API with the key in ZOTERO_API_KEY:

get_attrs(tfm, zotero_key=get_zotero_key(dfs))

{'geospatial_lat_min': '36.1816666666667',
 'geospatial_lat_max': '81.2680555555555',
 'geospatial_lon_min': '-56.15',
 'geospatial_lon_max': '49.4322222222222',
 'geospatial_bounds': 'POLYGON ((-56.15 36.1816666666667, 49.4322222222222 36.1816666666667, 49.4322222222222 81.2680555555555, -56.15 81.2680555555555, -56.15 36.1816666666667))',
 'geospatial_vertical_max': '1850.0',
 'geospatial_vertical_min': '0.0',
 'time_coverage_start': '1995-01-01T00:00:00',
 'time_coverage_end': '2019-12-31T00:00:00',
 'id': '3W354SQG',
 'title': 'Radioactivity Monitoring of the Irish Marine Environment 1991 and 1992',
 'summary': '',
 'creator_name': '[{"creatorType": "author", "firstName": "A.", "lastName": "McGarry"}, {"creatorType": "author", "firstName": "S.", "lastName": "Lyons"}, {"creatorType": "author", "firstName": "C.", "lastName": "McEnri"}, {"creatorType": "author", "firstName": "T.", "lastName": "Ryan"}, {"creatorType": "author", "firstName": "M.", "lastName": "O\'Colmain"}, {"creatorType": "author", "firstName": "J.D.", "lastName": "Cunningham"}]',
 'keywords': 'oceanography, Earth Science > Oceans > Ocean Chemistry> Radionuclides, Earth Science > Human Dimensions > Environmental Impacts > Nuclear Radiation Exposure, Earth Science > Oceans > Ocean Chemistry > Ocean Tracers, Earth Science > Oceans > Marine Sediments, Earth Science > Oceans > Ocean Chemistry, Earth Science > Oceans > Sea Ice > Isotopes, Earth Science > Oceans > Water Quality > Ocean Contaminants, Earth Science > Biological Classification > Animals/Vertebrates > Fish, Earth Science > Biosphere > Ecosystems > Marine Ecosystems, Earth Science > Biological Classification > Animals/Invertebrates > Mollusks, Earth Science > Biological Classification > Animals/Invertebrates > Arthropods > Crustaceans, Earth Science > Biological Classification > Plants > Macroalgae (Seaweeds)',
 'publisher_postprocess_logs': "Rename variables to MARIS standard names, keeping only renamed columns, Convert STATION column to string type, filling any missing values with empty string, Drop columns containing only NaN or 'Not available' (id=0 in MARIS lookup tables), except those in `keep`, Remap values from 'DL' to 'DL' for groups: all., Parse time column from ISO8601 string to datetime, Encode time as seconds since epoch, Drop rows with invalid longitude & latitude values. Convert `,` separator to `.` separator, Cast SMP_ID to int and SMP_ID_PROVIDER to string (renamed from samplabcode in the pipeline)"}

Encoding

encode runs the pipeline on each reference in the dump, or on the references in ref_ids, and writes each one to dest as <ref_id>.nc. marisco-ingest maris_legacy calls it.

source

encode

def encode(
    dest:str, # Path to the folder where the NetCDF output will be saved
    src:str=None, # Path to the MARIS dump data in CSV format
    **kwargs
):

MARIS legacy database dump

Exported source

def encode(
    dest: str, # Path to the folder where the NetCDF output will be saved
    src: str = None, # Path to the MARIS dump data in CSV format
    **kwargs # Additional keyword arguments
):
    "MARIS legacy database dump"
    assert src, "maris_legacy requires `src`, the path to the MARIS dump CSV"
    dataloader = DataLoader(src)
    ref_ids = parse_ref_ids(kwargs.get('ref_ids'), valid=dataloader.df.ref_id.unique())
    print('Encoding ...')
    for i, ref_id in enumerate(ref_ids):
        dfs = dataloader(ref_id=ref_id)
        print(f'{i+1}/{len(ref_ids)}: ref_id={ref_id} -> {dest}/{get_fname(dfs)}')
        tfm = Transformer(dfs, cbs=get_cbs())
        tfm()
        encoder = NetCDFEncoder(tfm.dfs, dest_fname=Path(dest)/get_fname(dfs),
            global_attrs=get_attrs(tfm, zotero_key=get_zotero_key(dfs)), verbose=kwargs.get('verbose', False))
        encoder.encode()

Encode one reference, here ref_id=106:

dir_dest = '../../_data/output/dump'
encode(src=fname_in, dest=dir_dest, ref_ids=[106])

Encoding ...
1/1: ref_id=106 -> ../../_data/output/dump/106.nc

Encode every reference in the dump:

encode(src=fname_in, dest=dir_dest)