MARIS Legacy

This handler ingests the MARIS master database dump (a tab-separated .text with all legacy datasets) and encodes each unique reference (ref_id) into a self-contained MARIS NetCDF4 file. Unlike provider-specific handlers (HELCOM, GEOTRACES, etc.), this one operates on data that is already aligned to the MARIS schema, so no nomenclature reconciliation is needed; the pipeline focuses on column selection, type casting, and encoding.

The pipeline processes each ref_id through these main stages:

Column selection and renaming: subset to columns of interest and rename to MARIS standard names
Type casting: coercing STATION to string for VLEN NetCDF variables
Drop NA columns: remove columns that are all NaN or all “Not available” (id=0)
Detection limit remapping: map DL symbols (<, =, etc.) to MARIS enum IDs
Time parsing and encoding: parse date strings, encode as seconds since epoch
Coordinate sanitization: validate lat/lon ranges and fix separator issues
Unique index: add a sequential row ID per group
NetCDF encoding: all groups assembled with global attributes (bbox, time range, Zotero citation, processing logs)

Configuration & file paths

fname_in: path to the folder containing the MARIS dump data in CSV format.
dir_dest: path to the folder where the NetCDF output will be saved.

Exported source

#fname_in = Path().home() / 'pro/data/maris/2025-06-03 MARIS_QA_shapetype_id = 1.txt'
fname_in = '../../_data/2025-06-03 MARIS_QA_shapetype_id = 1.txt'
dir_dest = '../../_data/output/dump'

df = pd.read_csv(fname_in, sep='\t', encoding='utf-8', low_memory=False)

Utils

Below a utility class to load a specific MARIS dump dataset optionally filtered through its ref_id.

source

DataLoader

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

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

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.

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.

Load data

Here below a quick overview of the MARIS dump data structure. For example, OSPAR data has ref_id=191, HELCOM has 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           (  696 rows, 80 cols)  sample_id, area_id, areaname, samptype_id, samptype, ref_id ...
SEAWATER        (  754 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 MARIS DB dump uses its own column names (e.g. area_id, activity, station). Here we map them to the MARIS standard names used throughout the pipeline. This mapping is hand-maintained because the dump schema and the NetCDF schema are independent; see nbs/api/configs.ipynb for the canonical NC_CSV that the NetCDF schema defines.

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 may arrive with mixed or numeric types, but the NetCDF template defines it as a VLEN string variable. CastStationToStringCB coerces it to string[python] 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

tfm = Transformer({
    'SEAWATER': pd.DataFrame({'STATION': ['A1', None, 42]}),
    'BIOTA': pd.DataFrame({'STATION': ['B2', None]})
}, 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=[
    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, keeping the output compact.

source

DropNAColumnsCB

def DropNAColumnsCB(
    na_value:int=0, # MARIS NA id to drop (default 0)
):

Drop variable containing only NaN or ‘Not available’ (id=0 in MARIS lookup tables).

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

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

tfm = Transformer(dfs, cbs=[
    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 stores detection limit symbols as strings (<, =, ND, etc.), but the MARIS NetCDF format encodes these as integer identifiers from the dbo_detectlimit lookup table. RemapCB maps each symbol to its integer id using get_lut, with unmapped values defaulting to 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=[
    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

In the MARIS NetCDF format, time is stored as an integer representing the number of seconds since a reference date (1970-01-01 00:00:00.0, as defined in nbs/api/files/cdl/maris.cdl). ParseTimeCB converts the TIME column from the original date string format, and EncodeTimeCB converts it to integer seconds.

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.

tfm = Transformer({
    'SEAWATER': pd.DataFrame({
        'LAT': [57.25, 91.0, '57,250'],
        'LON': [12.08, 181.0, '12,083']
    })
}, 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 MARIS dump provides sample_id (an internal MARIS sequential id) and samplabcode (the provider’s original sample identifier). The renaming step maps these to the MARIS standard names SMP_ID and SMP_ID_PROVIDER. AddSampleIDCB then casts SMP_ID to integer and SMP_ID_PROVIDER to a variable-length string, filling missing values with an empty string.

source

AddSampleIDCB

def AddSampleIDCB(
    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).

tfm = Transformer({
    'SEAWATER': pd.DataFrame({'SMP_ID': [1, 2], 'SMP_ID_PROVIDER': [None, 'RC1']})
}, cbs=[AddSampleIDCB()])
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=[
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB(),
    DropNAColumnsCB(),
    RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0),
    ParseTimeCB(),
    EncodeTimeCB(),
    SanitizeLonLatCB(),
    AddSampleIDCB()
])
dfs_tfm = tfm()
print(dfs_tfm['SEAWATER'][['SMP_ID', 'SMP_ID_PROVIDER']].head(3).to_string(index=False))

 SMP_ID SMP_ID_PROVIDER
 327125         1995013
 325627                
 330184        20160417

Encode to NetCDF

tfm = Transformer(dfs, cbs=[
    RenameColumnsCB(cois_renaming_rules),
    CastStationToStringCB(),
    DropNAColumnsCB(),
    RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0),
    ParseTimeCB(),
    EncodeTimeCB(),
    SanitizeLonLatCB(),
    AddSampleIDCB()
])

dfs_tfm = 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 variable containing only NaN or 'Not available' (id=0 in MARIS lookup tables).",
 "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 = ['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)']

def get_attrs(tfm, zotero_key, kw=kw):
    "Retrieve global attributes from MARIS dump."
    return GlobAttrsFeeder(tfm.dfs, cbs=[
        BboxCB(),
        DepthRangeCB(),
        TimeRangeCB(),
        ZoteroCB(zotero_key),
        KeyValuePairCB('keywords', ', '.join(kw)),
        KeyValuePairCB('publisher_postprocess_logs', ', '.join(tfm.logs))
        ])()

get_attrs(tfm, zotero_key='3W354SQG', kw=kw)

{'geospatial_lat_min': '36.4166666666667',
 'geospatial_lat_max': '80.56',
 'geospatial_lon_min': '-34.0',
 'geospatial_lon_max': '43.01',
 'geospatial_bounds': 'POLYGON ((-34 36.4166666666667, 43.01 36.4166666666667, 43.01 80.56, -34 80.56, -34 36.4166666666667))',
 'geospatial_vertical_max': '289.0',
 'geospatial_vertical_min': '0.0',
 'time_coverage_start': '1995-01-08T00:00:00',
 'time_coverage_end': '2019-12-05T00: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 variable containing only NaN or 'Not available' (id=0 in MARIS lookup tables)., 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

The encode function ties the full pipeline together: it loads each unique ref_id from the MARIS dump, runs the standard transformation pipeline, assembles global attributes (bbox, time range, Zotero citation, processing logs), and writes each reference as a separate NetCDF file.

def encode(
    fname_in: str, # Path to the MARIS dump data in CSV format
    dir_dest: str, # Path to the folder where the NetCDF output will be saved
    **kwargs # Additional keyword arguments
    ):
    "Encode MARIS dump to NetCDF."
    dataloader = DataLoader(fname_in)
    ref_ids = kwargs.get('ref_ids')
    if ref_ids is None:
        ref_ids = 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} -> {dir_dest}/{get_fname(dfs)}')
        tfm = Transformer(dfs, cbs=[
            RenameColumnsCB(cois_renaming_rules),
            CastStationToStringCB(),
            DropNAColumnsCB(),
            RemapCB(lut=lut_dl, col_src='DL', col_remap='DL', default_val=0),
            ParseTimeCB(),
            EncodeTimeCB(),
            SanitizeLonLatCB(),
            AddSampleIDCB(),
        ])
        
        tfm()
        encoder = NetCDFEncoder(tfm.dfs, 
                                dest_fname=Path(dir_dest) / get_fname(dfs), 
                                global_attrs=get_attrs(tfm, zotero_key=get_zotero_key(dfs), kw=kw),
                                verbose=kwargs.get('verbose', False)
                                )
        encoder.encode()

Single dataset

ref_id = 106
encode(
    fname_in,
    dir_dest,
    verbose=False, 
    ref_ids=[ref_id])

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

All datasets

encode(
    fname_in, 
    dir_dest, 
    ref_ids=None,
    verbose=False)