es_sfgtools.data_mgmt.directorymgmt.schemas module

class es_sfgtools.data_mgmt.directorymgmt.schemas.CampaignDir(*, location: Path | S3Path | None = None, raw: Path | S3Path | None = None, processed: Path | S3Path | None = None, intermediate: Path | S3Path | None = None, surveys: dict[str, SurveyDir] | None = {}, log_directory: Path | S3Path | None = None, qc: Path | S3Path | None = None, metadata_directory: Path | S3Path | None = None, campaign_metadata: Path | S3Path | None = None, rinex_metadata: Path | S3Path | None = None, svp_file: Path | S3Path | None = None, station: Path | S3Path, name: str)

Bases: _Base

Represents a campaign directory structure.

add_survey(name: str) → SurveyDir

Adds a new survey to the campaign.

Parameters:: name (str) – The name of the survey to add.
Returns:: The newly created or existing SurveyDir object.
Return type:: SurveyDir

build(): Creates the directory structure for the campaign.

campaign_metadata: Path | S3Path | None

intermediate: Path | S3Path | None

classmethod is_campaign_directory(path: Path | S3Path) → bool

Check if the given path is a valid campaign directory.

A valid campaign directory contains subdirectories for surveys.

Parameters:: path (Path | S3Path) – The path to check.
Returns:: True if the path is a valid campaign directory, False otherwise.
Return type:: bool

classmethod load_from_path(path: Path | S3Path) → CampaignDir

Load a CampaignDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the campaign directory.
Returns:: The loaded CampaignDir instance.
Return type:: CampaignDir

location: Path | S3Path | None

log_directory: Path | S3Path | None

metadata_directory: Path | S3Path | None

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'campaign_metadata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The campaign metadata file path'), 'intermediate': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Intermediate data directory path'), 'location': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The campaign directory path'), 'log_directory': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Logs directory path'), 'metadata_directory': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Metadata directory path'), 'name': FieldInfo(annotation=str, required=True), 'processed': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Processed data directory path'), 'qc': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Quality control directory path'), 'raw': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='Raw data directory path'), 'rinex_metadata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The RINEX metadata file path'), 'station': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The station directory path'), 'surveys': FieldInfo(annotation=Union[dict[str, SurveyDir], NoneType], required=False, default={}, description='Surveys in the campaign'), 'svp_file': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The sound velocity profile file path')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

processed: Path | S3Path | None

qc: Path | S3Path | None

raw: Path | S3Path | None

rinex_metadata: Path | S3Path | None

station: Path | S3Path

surveys: dict[str, SurveyDir] | None

svp_file: Path | S3Path | None

Bases: _Base

Represents a GARPOS survey directory structure.

build(): Creates the directory structure for the GARPOS survey.

default_obsfile: Path | None

default_settings: Path | None

find_filtered_shotdata() → Path | None

Find the filtered shotdata file in the parent survey directory.

Returns:: The path to the filtered shotdata file if found, else None.
Return type:: Optional[Path]

find_rectified_shotdata() → Path | None

Find the rectified shotdata file in the GARPOS directory.

Returns:: The path to the rectified shotdata file if found, else None.
Return type:: Optional[Path]

classmethod is_garpos_directory(path: Path | S3Path) → bool

Check if the given path is a valid GARPOS survey directory.

A valid GARPOS survey directory contains GARPOS default files.

Parameters:: path (Path | S3Path) – The path to check.
Returns:: True if the path is a valid GARPOS survey directory, False otherwise.
Return type:: bool

classmethod load_from_path(path: Path | S3Path) → GARPOSSurveyDir

Load a GARPOSSurveyDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the GARPOS survey directory.
Returns:: The loaded GARPOSSurveyDir instance.
Return type:: GARPOSSurveyDir

location: Path | None

log_directory: Path | None

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'default_obsfile': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The default observation file path'), 'default_settings': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The default GARPOS settings file path'), 'location': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The survey directory path'), 'log_directory': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The log directory path'), 'results_dir': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The results directory path'), 'shotdata_filtered': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The filtered shotdata file path'), 'shotdata_rectified': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The survey shotdata file path'), 'survey_dir': FieldInfo(annotation=Path, required=True, description='The parent survey directory path'), 'svp_file': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The sound velocity profile file path')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialise private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self – The BaseModel instance.
context – The context.

results_dir: Path | None

shotdata_filtered: Path | None

shotdata_rectified: Path | None

survey_dir: Path

svp_file: Path | None

class es_sfgtools.data_mgmt.directorymgmt.schemas.NetworkDir(*, stations: dict[str, StationDir] | None = {}, location: Path | S3Path | None = None, name: str, main_directory: Path | S3Path)

Bases: _Base

Represents a network directory structure.

add_station(name: str) → StationDir

Adds a new station to the network.

Parameters:: name (str) – The name of the station to add.
Returns:: The newly created or existing StationDir object.
Return type:: StationDir

build(): Creates the directory structure for the network.

classmethod is_network_directory(path: Path | S3Path) → bool

Check if the given path is a valid network directory.

A valid network directory contains subdirectories for stations.

Parameters:: path (Path | S3Path) – The path to check.
Returns:: True if the path is a valid network directory, False otherwise.
Return type:: bool

classmethod load_from_path(path: Path | S3Path) → NetworkDir

Load a NetworkDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the network directory.
Returns:: The loaded NetworkDir instance.
Return type:: NetworkDir

location: Path | S3Path | None

main_directory: Path | S3Path

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'location': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The network directory path'), 'main_directory': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The main directory path'), 'name': FieldInfo(annotation=str, required=True, description='The network name'), 'stations': FieldInfo(annotation=Union[dict[str, StationDir], NoneType], required=False, default={}, description='Stations in the network')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

stations: dict[str, StationDir] | None

class es_sfgtools.data_mgmt.directorymgmt.schemas.StationDir(*, campaigns: dict[str, CampaignDir] | None = {}, location: Path | S3Path | None = None, tiledb_directory: TileDBDir | None = None, metadata_directory: Path | S3Path | None = None, site_metadata: Path | S3Path | None = None, name: str, network: Path | S3Path)

Bases: _Base

Represents a station directory structure.

add_campaign(name: str) → CampaignDir

Adds a new campaign to the station.

Parameters:: name (str) – The name of the campaign to add.
Returns:: The newly created or existing CampaignDir object.
Return type:: CampaignDir

build(): Creates the directory structure for the station.

campaigns: dict[str, CampaignDir] | None

classmethod is_station_directory(path: Path | S3Path) → bool

Check if the given path is a valid station directory.

A valid station directory contains subdirectories for campaigns.

Parameters:: path (Path | S3Path) – The path to check.
Returns:: True if the path is a valid station directory, False otherwise.
Return type:: bool

classmethod load_from_path(path: Path | S3Path) → StationDir

Load a StationDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the station directory.
Returns:: The loaded StationDir instance.
Return type:: StationDir

location: Path | S3Path | None

metadata_directory: Path | S3Path | None

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'campaigns': FieldInfo(annotation=Union[dict[str, CampaignDir], NoneType], required=False, default={}, description='Campaigns in the station'), 'location': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The station directory path'), 'metadata_directory': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The metadata directory path'), 'name': FieldInfo(annotation=str, required=True, description='The station name'), 'network': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The network directory path'), 'site_metadata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The site metadata file path'), 'tiledb_directory': FieldInfo(annotation=Union[TileDBDir, NoneType], required=False, default=None, description='The TileDB directory path')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

network: Path | S3Path

site_metadata: Path | S3Path | None

tiledb_directory: TileDBDir | None

Bases: _Base

Represents a survey directory structure.

build(): Creates the directory structure for the survey.

campaign: Path | S3Path

garpos: GARPOSSurveyDir | None

imupositiondata: Path | S3Path | None

classmethod is_survey_directory(path: Path | S3Path) → bool

Check if the given path is a valid survey directory.

A valid survey directory contains a GARPOS subdirectory.

Parameters:: path (Path | S3Path) – The path to check.
Returns:: True if the path is a valid survey directory, False otherwise.
Return type:: bool

kinpositiondata: Path | S3Path | None

classmethod load_from_path(path: Path | S3Path) → SurveyDir

Load a SurveyDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the survey directory.
Returns:: The loaded SurveyDir instance.
Return type:: SurveyDir

location: Path | S3Path | None

metadata: Path | S3Path | None

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'campaign': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The campaign directory path'), 'garpos': FieldInfo(annotation=Union[GARPOSSurveyDir, NoneType], required=False, default=None, description='GARPOS data directory path'), 'imupositiondata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The IMU position file path'), 'kinpositiondata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The kinematic position file path'), 'location': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The survey directory path'), 'metadata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The survey metadata file path'), 'name': FieldInfo(annotation=str, required=True, description='The survey name'), 'shotdata': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The shotdata file path'), 'shotdata_filtered': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The filtered shotdata file path')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

name: str

shotdata: Path | S3Path | None

shotdata_filtered: Path | S3Path | None

Bases: _Base

Represents a directory structure for TileDB arrays.

acoustic_data: Path | S3Path | None

build(): Creates the directory structure for the TileDB arrays.

gnss_obs_data: Path | S3Path | None

gnss_obs_data_secondary: Path | S3Path | None

imu_position_data: Path | S3Path | None

kin_position_data: Path | S3Path | None

classmethod load_from_path(path: Path | S3Path) → TileDBDir

Load a TileDBDir instance from an existing directory path.

Parameters:: path (Path | S3Path) – The path to the TileDB directory.
Returns:: The loaded TileDBDir instance.
Return type:: TileDBDir

location: Path | S3Path | None

model_computed_fields = {}: A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config = {'arbitrary_types_allowed': True, 'json_encoders': {<class 'cloudpathlib.s3.s3path.S3Path'>: <function _Base.<lambda>>, <class 'datetime.datetime'>: <function _Base.<lambda>>, <class 'pathlib.Path'>: <function _Base.<lambda>>}}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields = {'acoustic_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The acoustic TileDB path'), 'gnss_obs_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The GNSS observation TileDB path'), 'gnss_obs_data_secondary': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The secondary GNSS observation TileDB path'), 'imu_position_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The IMU position TileDB path'), 'kin_position_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The kinematic position TileDB path'), 'location': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The TileDB directory path'), 'qc_gnss_obs_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The QC GNSS observation TileDB path'), 'qc_kin_position_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The QC kinematic position TileDB path'), 'qc_shot_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The QC shotdata TileDB path'), 'qc_shot_data_pre': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The QC preprocessed shotdata TileDB path'), 'shot_data': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The shotdata TileDB path'), 'shot_data_pre': FieldInfo(annotation=Union[Path, S3Path, NoneType], required=False, default=None, description='The preprocessed shotdata TileDB path'), 'station': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The station directory path')}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo].

This replaces Model.__fields__ from Pydantic V1.

qc_gnss_obs_data: Path | S3Path | None

qc_kin_position_data: Path | S3Path | None

qc_shot_data: Path | S3Path | None

qc_shot_data_pre: Path | S3Path | None

shot_data: Path | S3Path | None

shot_data_pre: Path | S3Path | None

station: Path | S3Path

to_s3() → None