sidescanpy.core.schema

Schema validation and xr.Accessor for SideScanPy standard HDF5 format

This module defines the .ss.h5 (standardized SideScan HDF5) contract that all datasets must conform to. It provides validation via xr.Accessor and utilities for tracking processing state.

2026 rshomberg@uri.edu

Contents

REQUIRED_DIMS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_COORDS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_VARS	set() -> new empty set object set(iterable) -> new set object
OPTIONAL_PER_PING_VARS
POSITION_SOURCE_ATTR	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_SOURCE_REASON_ATTR	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_SHIP	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_TOWFISH	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_MIXED	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_UNKNOWN	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
POSITION_SOURCES	Built-in immutable sequence.
combine_position_sources	One POSITION_SOURCES value for a set of per-ping sources.
REQUIRED_ATTRS	set() -> new empty set object set(iterable) -> new set object
KNOWN_PROCESSING_LEVELS	Values sidescanpy.core.history.processing_level() can return.
LEGACY_PROCESSING_LEVEL_VALUES	Values the legacy processing_level attribute may hold on older files.
HDF5_GROUP_AUX_STREAMS	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_GROUP_NAVIGATION	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_GROUP_ATTITUDE	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_GROUP_DVL	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_GROUP_SENSORS	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_GROUP_FREQUENCIES	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_FREQ_PREFIX	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
SS_FORMAT_NAME	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
SS_FORMAT_VERSION	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_CHUNK_PING	int([x]) -> integer int(x, base=10) -> integer
HDF5_COMPRESSION	str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str
HDF5_COMPRESSION_LEVEL	int([x]) -> integer int(x, base=10) -> integer
SideScanAccessor	xarray.Dataset accessor for sidescan specific validation and utilities
REQUIRED_NAV_DIMS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_NAV_COORDS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_NAV_VARS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_NAV_ATTRS	set() -> new empty set object set(iterable) -> new set object
NavigationRecordAccessor	xarray.Dataset accessor for navigation auxiliary streams.
REQUIRED_ATT_DIMS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_ATT_COORDS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_ATT_VARS	set() -> new empty set object set(iterable) -> new set object
OPTIONAL_ATT_VARS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_ATT_ATTRS	set() -> new empty set object set(iterable) -> new set object
AttitudeRecordAccessor	Auxiliary stream for attitude data (roll, pitch, heave).
REQUIRED_DVL_DIMS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_DVL_COORDS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_DVL_VARS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_DVL_ATTRS	set() -> new empty set object set(iterable) -> new set object
DVLRecordAccessor	Auxiliary stream for DVL altimeter data.
REQUIRED_GEN_DIMS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_GEN_COORDS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_GEN_VARS	set() -> new empty set object set(iterable) -> new set object
REQUIRED_GEN_ATTRS	set() -> new empty set object set(iterable) -> new set object
SensorRecordAccessor	Generic auxiliary stream for arbitrary time-series sensor data.
make_empty_navigation_record	Zero-obs valid navigation_record.
make_empty_attitude_record	Zero-obs valid attitude_record.
make_empty_dvl_record	Zero-obs valid dvl_record.
make_empty_sensor_record	Zero-obs valid sensor_record.

sidescanpy.core.schema.REQUIRED_DIMS = {'channel', 'ping', 'sample'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_COORDS = {'channel_id', 'ping_number', 'time'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_VARS

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.OPTIONAL_PER_PING_VARS

sidescanpy.core.schema.POSITION_SOURCE_ATTR = 'position_source'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_SOURCE_REASON_ATTR = 'position_source_reason'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_SHIP = 'ship'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_TOWFISH = 'towfish'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_MIXED = 'mixed'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_UNKNOWN = 'unknown'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.POSITION_SOURCES

Built-in immutable sequence.

If no argument is given, the constructor returns an empty tuple. If iterable is specified the tuple is initialized from iterable’s items.

If the argument is a tuple, the return value is the same object.

sidescanpy.core.schema.combine_position_sources(per_ping)

One POSITION_SOURCES value for a set of per-ping sources.

All pings the same → that value; empty → unknown; otherwise mixed.

sidescanpy.core.schema.REQUIRED_ATTRS

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.KNOWN_PROCESSING_LEVELS = ('raw', 'tvg', 'geometric', 'range_correction', 'georef')

Values sidescanpy.core.history.processing_level() can return.

sidescanpy.core.schema.LEGACY_PROCESSING_LEVEL_VALUES = {'radiometric', 'raw'}

Values the legacy processing_level attribute may hold on older files. The attribute is no longer written; the level is derived from history.

sidescanpy.core.schema.HDF5_GROUP_AUX_STREAMS = 'aux_streams'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_GROUP_NAVIGATION = 'aux_streams/navigation'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_GROUP_ATTITUDE = 'aux_streams/attitude'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_GROUP_DVL = 'aux_streams/dvl'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_GROUP_SENSORS = 'aux_streams/sensors'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_GROUP_FREQUENCIES = 'frequencies'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_FREQ_PREFIX = 'hz_'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.SS_FORMAT_NAME = 'sidescanpy-ss'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.SS_FORMAT_VERSION = '0.1'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_CHUNK_PING = 256

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

sidescanpy.core.schema.HDF5_COMPRESSION = 'gzip'

str(object=’’) -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

Create a new string object from the given object. If encoding or errors is specified, then the object must expose a data buffer that will be decoded using the given encoding and error handler. Otherwise, returns the result of object.__str__() (if defined) or repr(object). encoding defaults to sys.getdefaultencoding(). errors defaults to ‘strict’.

sidescanpy.core.schema.HDF5_COMPRESSION_LEVEL = 4

int([x]) -> integer int(x, base=10) -> integer

Convert a number or string to an integer, or return 0 if no arguments are given. If x is a number, return x.__int__(). For floating-point numbers, this truncates towards zero.

If x is not a number or if base is given, then x must be a string, bytes, or bytearray instance representing an integer literal in the given base. The literal can be preceded by ‘+’ or ‘-’ and be surrounded by whitespace. The base defaults to 10. Valid bases are 0 and 2-36. Base 0 means to interpret the base from the string as an integer literal. >>> int(‘0b100’, base=0) 4

class sidescanpy.core.schema.SideScanAccessor(ds)

Bases: object

xarray.Dataset accessor for sidescan specific validation and utilities

Provides methods to validate datasets against the SideScanPy schema, track processing state, and compute derived quantities like slant range.

Accessed via ds.sidescan on any xarray.Dataset

Examples

>>> import xarray as xr
>>> ds = xr.Dataset(...)
>>> ds.sidescan.validate()  # Raises ValueError if invalid
>>> ds.sidescan.is_valid()  # Returns bool
>>> ds.sidescan.processing_level  # Returns str

validate()

Validate dataset against SideScanPy schema.

Checks (in order):

1. All required dimensions present and channel size == 2

2. All required coordinates present

3. All required variables present

4. All required attributes present

5. A legacy processing_level attribute, if present, holds a legacy value (raw/radiometric)

Raises:

ValueError

On first schema violation with specific, actionable message.

Examples

>>> ds.sidescan.validate()  # Raises if invalid, returns None if valid

is_valid()

Check if dataset is valid without raising.

Returns:

bool

True if dataset passes all validations, False otherwise.

Examples

>>> if ds.sidescan.is_valid():
...     process(ds)

property processing_level

Current processing level, derived from processing_history.

See sidescanpy.core.history.processing_level().

property processing_history

Parse and return processing history.

Parses JSON from ds.attrs['processing_history'] into a list of dictionaries. Each entry has keys: - ‘step’: str, name of processing step - ‘timestamp_utc’: str, ISO 8601 timestamp - ‘params’: dict, parameters used in step - ‘function’: str, fully qualified function name - ‘version’: str, semantic version of function

Returns:

list of dict

Processing history entries, or empty list if missing/malformed.

Examples

>>> history = ds.sidescan.processing_history
>>> for entry in history:
...     print(f"{entry['step']} at {entry['timestamp_utc']}")

append_to_history(step_name, function_name, params, version)

Append an entry to processing history.

Creates a new history entry with current timestamp and updates ds.attrs['processing_history'] as JSON string. Preserves all prior entries (append-only).

Parameters:

step_namestr

Name of processing step (e.g., ‘tvg_correction’).

function_namestr

Fully qualified function name (e.g., ‘sidescanpy.processing.radiometric.apply_tvg’).

paramsdict

Parameters used in this step. Can be any JSON-serializable dict.

versionstr

Semantic version of the function (e.g., ‘0.2.0’).

Examples

>>> ds.sidescan.append_to_history(
...     step_name='tvg_correction',
...     function_name='sidescanpy.processing.radiometric.apply_tvg',
...     params={'slope_db_per_m': 30.0, 'offset_db': 20.0},
...     version='0.2.0',
... )

sidescanpy.core.schema.REQUIRED_NAV_DIMS = {'obs'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_NAV_COORDS = {'time'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_NAV_VARS = {'latitude', 'longitude'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_NAV_ATTRS = {'source'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

class sidescanpy.core.schema.NavigationRecordAccessor(ds)

Bases: object

xarray.Dataset accessor for navigation auxiliary streams.

A navigation record is a time series of position fixes, typically extracted from NMEA messages (RMC, GLL, GGA) or fallback-extracted from per-ping embedded fields. Stored as separate aux stream so it can be re-interpolated onto ping times later.

Required dims: {obs} Required coords: {time} (datetime64[ns]) Required vars: {latitude, longitude} Optional vars: heading_deg, speed_knots, course_deg Required attrs: {source}

Examples

>>> nav.navigation_record.validate()
>>> nav.navigation_record.is_empty()

is_empty()

Return True if the dataset has zero observations.

sidescanpy.core.schema.REQUIRED_ATT_DIMS = {'obs'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_ATT_COORDS = {'time'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_ATT_VARS = {'heave_m', 'pitch_deg', 'roll_deg'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.OPTIONAL_ATT_VARS = {'heading_deg', 'temperature_c', 'yaw_deg'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_ATT_ATTRS = {'source'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

class sidescanpy.core.schema.AttitudeRecordAccessor(ds)

Bases: object

Auxiliary stream for attitude data (roll, pitch, heave).

Typically sourced from PitchRollDataMessage (JSF type 2020) or XTF ping headers.

Required dims: {obs} Required coords: {time} (datetime64[ns], same timebase as ping times) Required vars: {roll_deg, pitch_deg, heave_m} Optional vars: heading_deg (compass bearing, degrees clockwise from north), yaw_deg, temperature_c Required attrs: {source}

A NaN value marks a reading the source flagged invalid or did not carry.

sidescanpy.core.schema.REQUIRED_DVL_DIMS = {'obs'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_DVL_COORDS = {'time'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_DVL_VARS = {'altitude_m'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_DVL_ATTRS = {'source'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

class sidescanpy.core.schema.DVLRecordAccessor(ds)

Bases: object

Auxiliary stream for DVL altimeter data.

Sourced from DopplerVelocityLogDataMessage (JSF type 2080). Optional vars include bottom_vel_x/y/z, water_vel_x/y/z, sound_speed_ms, temperature_c.

sidescanpy.core.schema.REQUIRED_GEN_DIMS = {'obs'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_GEN_COORDS = {'time'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_GEN_VARS = {}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

sidescanpy.core.schema.REQUIRED_GEN_ATTRS = {'sensor_type', 'source'}

set() -> new empty set object set(iterable) -> new set object

Build an unordered collection of unique elements.

class sidescanpy.core.schema.SensorRecordAccessor(ds)

Bases: object

Generic auxiliary stream for arbitrary time-series sensor data.

Use for streams that don’t fit the navigation/attitude/dvl categories, e.g. pressure-derived depth, water temperature, conductivity.

Required dims: {obs} Required coords: {time} Required vars: (none — depends on sensor) Required attrs: {source, sensor_type}

sidescanpy.core.schema.make_empty_navigation_record(source='empty')

Zero-obs valid navigation_record.

sidescanpy.core.schema.make_empty_attitude_record(source='empty')

Zero-obs valid attitude_record.

sidescanpy.core.schema.make_empty_dvl_record(source='empty')

Zero-obs valid dvl_record.

sidescanpy.core.schema.make_empty_sensor_record(sensor_type, source='empty', *, var_name='depth_m')

Zero-obs valid sensor_record.