Skip to content

Density NetCDF Strategy

el_paso.saving_strategies.density_netcdf_strategy.DensityNetCDFStrategy

Bases: MonthlyRBStrategy

Saving strategy for writing plasma density and related data to monthly NetCDF files.

This strategy extends MonthlyRBStrategy but implements saving to the NetCDF format (.nc), primarily targeting the time-series of density, position, and coordinate variables (e.g., L-star, MLT).

The variables included and their dependencies are configured based on whether the data is associated with the "RBSP" satellites or "Other".

Attributes:

Name Type Description
output_files list[OutputFile]

List of file configurations to be produced.
file_path Path

Base path for output files (inherited).
dependency_dict dict[str, list[str]]

Defines the NetCDF dimension names (e.g., 'time', 'xGEO_components') that each variable depends on.
Source code in el_paso/saving_strategies/density_netcdf_strategy.py 
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class DensityNetCDFStrategy(MonthlyRBStrategy):
    """Saving strategy for writing plasma density and related data to monthly NetCDF files.

    This strategy extends `MonthlyRBStrategy` but implements saving to the NetCDF
    format (`.nc`), primarily targeting the time-series of density, position, and
    coordinate variables (e.g., L-star, MLT).

    The variables included and their dependencies are configured based on whether
    the data is associated with the **"RBSP"** satellites or **"Other"**.

    Attributes:
        output_files (list[OutputFile]): List of file configurations to be produced.
        file_path (Path): Base path for output files (inherited).
        dependency_dict (dict[str, list[str]]): Defines the NetCDF dimension names
            (e.g., 'time', 'xGEO_components') that each variable depends on.
    """

    output_files: list[OutputFile]

    file_path: Path
    dependency_dict: dict[str, list[str]]

    def __init__(
        self,
        base_data_path: str | Path,
        mission: str,
        instrument: str,
        mag_field: MagneticFieldLiteral,
        satellite: Literal["RBSP", "Other"] = "Other",
        data_standard: Optional[DataStandard[StandardName]] = None,
    ) -> None:
        """Initializes the monthly NetCDF saving strategy.

        Args:
            base_data_path (str | Path): The base directory where the output NetCDF files will be saved.
            mission (str): The mission name, used in file path and name generation.
            instrument (str): The instrument name, used in file path and name generation.
            mag_field (MagneticFieldLiteral):
                A string specifying the magnetic field model used.
            satellite (Literal["RBSP", "Other"], optional):
                            Specifies the satellite associated with the data. This determines which set of
                            density-related variables are included in the output file. Defaults to "Other".
            data_standard (DataStandard | None, optional):
                An optional `DataStandard` instance to use for standardizing variables.
                If `None`, `ep.data_standards.PRBEMStandard` is used by default.
        """
        self.mag_field = mag_field
        self.data_standard = data_standard or ep.data_standards.PRBEMStandard()

        super().__init__(
            base_data_path=base_data_path,
            satellite=satellite,
            mission=mission,
            instrument=instrument,
            mag_field=self.mag_field,
            file_format="nc",
            data_standard=data_standard,
        )

        output_file_entries = [
            "time",
            "xGEO",
            "MLT",
            "R_eq",
            "Lstar",
            "xGEO_eq",
        ]

        self.dependency_dict = {
            "time": ["time"],
            "xGEO": ["time", "xGEO_components"],
            "MLT": ["time"],
            "R_eq": ["time"],
            "xGEO_eq": ["time", "xGEO_components"],
            "Lstar": ["time"],
        }

        if satellite == "Other":
            output_file_entries += ["density_local", "density_eq"]
            self.dependency_dict |= {"density_local": ["time"], "density_eq": ["time"]}

        elif satellite == "RBSP":
            output_file_entries += [
                "density_emfisis_local",
                "density_efw_local",
                "density_hiss_derived_local",
                "density_emfisis_eq",
                "density_efw_eq",
                "density_hiss_derived_eq",
            ]

            self.dependency_dict |= {
                "density_emfisis_local": ["time"],
                "density_efw_local": ["time"],
                "density_hiss_derived_local": ["time"],
                "density_emfisis_eq": ["time"],
                "density_efw_eq": ["time"],
                "density_hiss_derived_eq": ["time"],
            }

        else:
            msg = "Enountered invalid satellite! Valid names are: 'RBSP', 'Other'."
            raise ValueError(msg)

        self.output_files = [
            OutputFile("full", output_file_entries, save_incomplete=True),
        ]

    def _calculate_dimensions(self, data_dict: dict[str, np.ndarray]) -> dict[str, int]:
        """Calculate density NetCDF dimension sizes from the data dictionary."""
        dimensions = {"time": data_dict["time"].shape[0]}

        has_local_position = "xGEO" in data_dict and data_dict["xGEO"].size > 0
        has_equatorial_position = "xGEO_eq" in data_dict and data_dict["xGEO_eq"].size > 0
        if has_local_position or has_equatorial_position:
            dimensions["xGEO_components"] = 3

        return dimensions

    def standardize_variable(
        self, variable: ep.Variable, name_in_file: str, *, first_call_of_interval: bool
    ) -> ep.Variable:
        """Standardizes a variable based on the configured `DataStandard`.

        This method delegates the standardization process to a `DataStandard` instance,
        ensuring that the variable's units and dimensions are consistent with the
        defined standard.

        Args:
            variable (ep.Variable): The variable instance to be standardized.
            name_in_file (str): The name of the variable as it will appear in the file.
            first_call_of_interval (bool): Flag to indicate if it is the first call of a time interval

        Returns:
            ep.Variable: The standardized variable.
        """
        return self.data_standard.standardize_variable(
            name_in_file, variable, reset_consistency_check=first_call_of_interval
        )

Methods:

el_paso.saving_strategies.density_netcdf_strategy.DensityNetCDFStrategy.__init__ 

__init__

Initializes the monthly NetCDF saving strategy.

Parameters:

Name Type Description Default
base_data_path str | Path

The base directory where the output NetCDF files will be saved.

 required
mission str

The mission name, used in file path and name generation.

 required
instrument str

The instrument name, used in file path and name generation.

 required
mag_field MagneticFieldLiteral

A string specifying the magnetic field model used.

 required
satellite Literal['RBSP', 'Other'] 
        Specifies the satellite associated with the data. This determines which set of
        density-related variables are included in the output file. Defaults to "Other".
'Other'
data_standard DataStandard | None

An optional DataStandard instance to use for standardizing variables. If None, ep.data_standards.PRBEMStandard is used by default.

 None
Source code in el_paso/saving_strategies/density_netcdf_strategy.py 
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
def __init__(
    self,
    base_data_path: str | Path,
    mission: str,
    instrument: str,
    mag_field: MagneticFieldLiteral,
    satellite: Literal["RBSP", "Other"] = "Other",
    data_standard: Optional[DataStandard[StandardName]] = None,
) -> None:
    """Initializes the monthly NetCDF saving strategy.

    Args:
        base_data_path (str | Path): The base directory where the output NetCDF files will be saved.
        mission (str): The mission name, used in file path and name generation.
        instrument (str): The instrument name, used in file path and name generation.
        mag_field (MagneticFieldLiteral):
            A string specifying the magnetic field model used.
        satellite (Literal["RBSP", "Other"], optional):
                        Specifies the satellite associated with the data. This determines which set of
                        density-related variables are included in the output file. Defaults to "Other".
        data_standard (DataStandard | None, optional):
            An optional `DataStandard` instance to use for standardizing variables.
            If `None`, `ep.data_standards.PRBEMStandard` is used by default.
    """
    self.mag_field = mag_field
    self.data_standard = data_standard or ep.data_standards.PRBEMStandard()

    super().__init__(
        base_data_path=base_data_path,
        satellite=satellite,
        mission=mission,
        instrument=instrument,
        mag_field=self.mag_field,
        file_format="nc",
        data_standard=data_standard,
    )

    output_file_entries = [
        "time",
        "xGEO",
        "MLT",
        "R_eq",
        "Lstar",
        "xGEO_eq",
    ]

    self.dependency_dict = {
        "time": ["time"],
        "xGEO": ["time", "xGEO_components"],
        "MLT": ["time"],
        "R_eq": ["time"],
        "xGEO_eq": ["time", "xGEO_components"],
        "Lstar": ["time"],
    }

    if satellite == "Other":
        output_file_entries += ["density_local", "density_eq"]
        self.dependency_dict |= {"density_local": ["time"], "density_eq": ["time"]}

    elif satellite == "RBSP":
        output_file_entries += [
            "density_emfisis_local",
            "density_efw_local",
            "density_hiss_derived_local",
            "density_emfisis_eq",
            "density_efw_eq",
            "density_hiss_derived_eq",
        ]

        self.dependency_dict |= {
            "density_emfisis_local": ["time"],
            "density_efw_local": ["time"],
            "density_hiss_derived_local": ["time"],
            "density_emfisis_eq": ["time"],
            "density_efw_eq": ["time"],
            "density_hiss_derived_eq": ["time"],
        }

    else:
        msg = "Enountered invalid satellite! Valid names are: 'RBSP', 'Other'."
        raise ValueError(msg)

    self.output_files = [
        OutputFile("full", output_file_entries, save_incomplete=True),
    ]