Skip to content

Daily LEO RB Strategy

el_paso.saving_strategies.daily_leo_rb_strategy.DailyLEORBStrategy

Bases: MonthlyRBStrategy

Save PRBEM-standard LEO radiation-belt data into one NetCDF file per day.

This strategy extends MonthlyRBStrategy but splits the output into daily files instead of monthly ones, and fixes the output variable list and file format (NetCDF) for low-Earth-orbit radiation-belt missions.

Source code in el_paso/saving_strategies/daily_leo_rb_strategy.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
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
class DailyLEORBStrategy(ep.typing.MonthlyRBStrategy):
    """Save PRBEM-standard LEO radiation-belt data into one NetCDF file per day.

    This strategy extends `MonthlyRBStrategy` but splits the output into daily
    files instead of monthly ones, and fixes the output variable list and file
    format (NetCDF) for low-Earth-orbit radiation-belt missions.
    """

    def _get_output_file_entries(self) -> list[ep.typing.InternalName]:
        """Return the standard variable list plus user-defined custom variables."""
        return [
            "FEDU",
            "FEIU",
            "FEDO",
            "FPDU",
            "Epoch",
            "Alpha_Eq",
            "Alpha_Eq_range",
            "Energy_FEDU",
            "Energy_FEIU",
            "Energy_FEDO",
            "Energy_FPDU",
            "Alpha",
            "Alpha_range",
            "B_Calc",
            "B_Eq",
            "InvK",
            "InvMu",
            "Position",
            "PSD",
            "R_Eq",
            "MLT",
            "L_m",
            "L_star",
            "Alpha_LC",
            "Alpha_LC_Eq",
            "Position_geo_alt",
            "Position_geo_lat",
            "Position_geo_lon",
        ]

    def get_file_path(
        self, interval_start: datetime, interval_end: datetime, output_file: ep.typing.OutputFile,  # noqa: ARG002
    ) -> Path:
        """Generate the daily file path for the configured format."""
        file_name = f"{self.get_file_name_stem()}_{interval_start.strftime('%Y%m%d')}_{self.mag_field}.nc"

        return self.get_file_path_stem() / file_name

    def get_time_intervals_to_save(
        self, start_time: datetime | None, end_time: datetime | None
    ) -> list[ep.typing.TimeInterval]:
        """Split the requested time range into full daily intervals."""
        time_intervals: list[ep.typing.TimeInterval] = []

        if start_time is None or end_time is None:
            msg = "start_time and end_time must be provided for DailyWaveStrategy!"
            raise ValueError(msg)

        current_time = start_time.replace(hour=0, minute=0, second=0, microsecond=0)
        while current_time <= end_time:
            interval_start = current_time
            interval_end = current_time + timedelta(days=1, microseconds=-1)

            time_intervals.append((interval_start, interval_end))
            current_time += timedelta(days=1)

        return time_intervals

Methods:

el_paso.saving_strategies.daily_leo_rb_strategy.DailyLEORBStrategy.get_file_path 

get_file_path

Generate the daily file path for the configured format.

Source code in el_paso/saving_strategies/daily_leo_rb_strategy.py 
52
53
54
55
56
57
58
def get_file_path(
    self, interval_start: datetime, interval_end: datetime, output_file: ep.typing.OutputFile,  # noqa: ARG002
) -> Path:
    """Generate the daily file path for the configured format."""
    file_name = f"{self.get_file_name_stem()}_{interval_start.strftime('%Y%m%d')}_{self.mag_field}.nc"

    return self.get_file_path_stem() / file_name

el_paso.saving_strategies.daily_leo_rb_strategy.DailyLEORBStrategy.get_time_intervals_to_save 

get_time_intervals_to_save

Split the requested time range into full daily intervals.

Source code in el_paso/saving_strategies/daily_leo_rb_strategy.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
def get_time_intervals_to_save(
    self, start_time: datetime | None, end_time: datetime | None
) -> list[ep.typing.TimeInterval]:
    """Split the requested time range into full daily intervals."""
    time_intervals: list[ep.typing.TimeInterval] = []

    if start_time is None or end_time is None:
        msg = "start_time and end_time must be provided for DailyWaveStrategy!"
        raise ValueError(msg)

    current_time = start_time.replace(hour=0, minute=0, second=0, microsecond=0)
    while current_time <= end_time:
        interval_start = current_time
        interval_end = current_time + timedelta(days=1, microseconds=-1)

        time_intervals.append((interval_start, interval_end))
        current_time += timedelta(days=1)

    return time_intervals