es_sfgtools.data_mgmt.directorymgmt.handler module

This module defines a set of Pydantic models for managing a directory structure for geodesy data processing.

The models define a hierarchical structure of directories and files, starting from a main directory, and branching into networks, stations, campaigns, and various data and results directories.

class DirectoryHandler:

The main class that manages the entire directory structure.

class NetworkDir:

Represents a network of stations.

class StationDir:

Represents a single station with multiple campaigns.

class CampaignDir:

Represents a data collection campaign.

class TileDBDir:

Manages TileDB arrays for a station.

class GARPOSCampaignDir:

Manages GARPOS-specific data and results.

class GARPOSSurveyDir:

Represents a single GARPOS survey.

class es_sfgtools.data_mgmt.directorymgmt.handler.DirectoryHandler(*, filepath: Path | None = None, asset_catalog_db_path: Path | None = None, networks: dict[str, NetworkDir] | None = {}, pride_directory: Path | None = None, location: Path | S3Path)

Bases: _Base

The main class for managing the directory structure.

add_network(name: str) → NetworkDir

Adds a new network to the directory structure.

Parameters:

name (str) – The name of the network to add.

Returns:

The newly created or existing NetworkDir object.

Return type:

NetworkDir

asset_catalog_db_path: Path | None
build()

Creates the main directory structure.

build_station_directory(network_name: str, station_name: str = None, campaign_name: str = None, survey_name: str = None) → tuple[NetworkDir, StationDir, CampaignDir, SurveyDir] | None

Builds a station directory, and optionally a campaign directory.

Parameters:

  • network_name (str) – The name of the network.

  • station_name (str, optional) – The name of the station, by default None.

  • campaign_name (str, optional) – The name of the campaign, by default None.

  • survey_name (str, optional) – The name of the survey, by default None.

Returns:

A tuple containing the created directory objects, or None if the directory was not built successfully.

Return type:

Optional[tuple[NetworkDir, StationDir, CampaignDir, SurveyDir]]

filepath: Path | None
classmethod load_from_path(path: str | Path | S3Path) → DirectoryHandler

Searches the local or S3 path for existing data.

Parameters:

path (str | Path | S3Path) – The local or S3 directory path.

Returns:

A DirectoryHandler object.

Return type:

DirectoryHandler

classmethod load_from_s3(bucket_path: str) → DirectoryHandler

Searches the s3 bucket for existing data.

Parameters:

bucket_path (str) – The S3 bucket path (e.g., “s3://my-bucket/path”).

Returns:

A DirectoryHandler object.

Return type:

DirectoryHandler

location: 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 = {'asset_catalog_db_path': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='Path to the asset catalog database'), 'filepath': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='Path to the directory structure JSON file'), 'location': FieldInfo(annotation=Union[Path, S3Path], required=True, description='The main directory path'), 'networks': FieldInfo(annotation=Union[dict[str, NetworkDir], NoneType], required=False, default={}), 'pride_directory': FieldInfo(annotation=Union[Path, NoneType], required=False, default=None, description='The PRIDE PPPAR binary 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.

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.

networks: dict[str, NetworkDir] | None
point_to_s3(bucket_path: str | S3Path) → DirectoryHandler

Points the directory handler to an S3 bucket using cloudpathlib.

  1. Create model copy of current directory handler.

    >> new_handler = self.model_copy()

  2. Update location to CloudPath of S3 bucket.

    >> new_handler.location = “/Volumes/ThisVolume/Project/SeafloorGeodesy/SFGMain” >> local_location = new_handler.location >> new_handler.location = cloudpathlib.CloudPath(“s3://my-bucket/path”)

  3. Change all path attributes in the model to CloudPath objects, replacing local_location with new_handler.location.

Parameters:

bucket_path (str) – The S3 bucket path (e.g., “s3://my-bucket/path”).

pride_directory: Path | None
save()

Saves the directory structure to a JSON file.

es_sfgtools.data_mgmt.directorymgmt.handler.get_s3_client() → S3Client | None

Gets an S3 client using cloudpathlib with support for AWS profiles.

es_sfgtools.data_mgmt.directorymgmt.handler.load_credentials() → tuple[str | None, str | None, str | None, str | None]

Loads credentials for accessing remote resources.

Returns:

(aws_access_key, aws_secret_key, aws_session_token, aws_profile)

Return type:

tuple