Skip to content

Snapshot filter

SnapshotFilter

Bases: BaseModel

Source code in passengersim/config/snapshot_filter.py 
 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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
class SnapshotFilter(BaseModel, validate_assignment=True):
    type: Literal[
        "forecast",
        "leg_untruncation",
        "path_untruncation",
        "rm",
        "pro_bp",
        "forecast_adj",
        None,
    ] = None
    title: str = ""
    airline: str = ""
    trial: list[int] = []
    sample: list[int] = []
    dcp: list[int] = []
    orig: list[str] = []
    dest: list[str] = []
    flt_no: list[int] = []
    logger: str | None = None
    directory: pathlib.Path | None = None

    @field_validator("trial", "sample", "dcp", "orig", "dest", "flt_no", mode="before")
    def _allow_singletons(cls, v):
        """Allow a singleton value that is converted to a list of one item."""
        if not isinstance(v, list | tuple):
            v = [v]
        return v

    def filepath(self, sim, leg=None, path=None) -> pathlib.Path | None:
        if self.directory is None:
            return None
        pth = self.directory
        if leg is not None:
            pth = pth.joinpath(f"carrier-{leg.carrier}")
        pth = pth.joinpath(f"dpc-{sim.last_dcp}")
        if leg is not None:
            pth = pth.joinpath(f"orig-{leg.orig}")
        elif path is not None:
            pth = pth.joinpath(f"orig-{path.orig}")
        if leg is not None:
            pth = pth.joinpath(f"dest-{leg.dest}")
        elif path is not None:
            pth = pth.joinpath(f"dest-{path.dest}")
        if leg is not None:
            pth = pth.joinpath(f"fltno-{leg.flt_no}")
        elif path is not None:
            pth = pth.joinpath(f"fltno-{path.get_leg_fltno(0)}")
        if sim.num_trials > 1:
            pth = pth.joinpath(f"trial-{sim.trial}")
        pth = pth.joinpath(f"sample-{sim.sample}")
        pth.parent.mkdir(parents=True, exist_ok=True)
        return pth.with_suffix(".log")

    def run(
        self, sim, leg=None, path=None, carrier=None, why=False
    ) -> SnapshotInstruction:
        # Check the filter conditions
        info = ""

        if len(self.trial) > 0 and sim.trial not in self.trial and sim.num_trials > 1:
            return SnapshotInstruction(False, why=f"cause {sim.trial=}")
        info += f"  trial={sim.trial}"

        if len(self.sample) > 0 and sim.sample not in self.sample:
            return SnapshotInstruction(False, why=f"cause {sim.sample=}")
        info += f"  sample={sim.sample}"

        if len(self.dcp) > 0 and sim.last_dcp not in self.dcp:
            return SnapshotInstruction(False, why=f"cause {sim.last_dcp=}")
        info += f"  dcp={sim.last_dcp}"

        if leg is not None:
            if self.airline and leg.carrier != self.airline:
                return SnapshotInstruction(False, why=f"cause {leg.carrier=}")
            info += f"  carrier={leg.carrier}"

            if len(self.orig) > 0 and leg.orig not in self.orig:
                return SnapshotInstruction(False, why=f"cause {leg.orig=}")
            info += f"  orig={leg.orig}"

            if len(self.dest) > 0 and leg.dest not in self.dest:
                return SnapshotInstruction(False, why=f"cause {leg.dest=}")
            info += f"  dest={leg.dest}"

            if len(self.flt_no) > 0 and leg.flt_no not in self.flt_no:
                return SnapshotInstruction(False, why=f"cause {leg.flt_no=}")
            info += f"  flt_no={leg.flt_no}"

        if path is not None:
            if len(self.orig) > 0 and path.orig not in self.orig:
                return SnapshotInstruction(False, why=f"cause {path.orig=}")
            info += f"  orig={path.orig}"

            if len(self.dest) > 0 and path.dest not in self.dest:
                return SnapshotInstruction(False, why=f"cause {path.dest=}")
            info += f"  dest={path.dest}"

            if len(self.flt_no) > 0 and path.get_leg_fltno(0) not in self.flt_no:
                return SnapshotInstruction(False, why=f"cause {path.get_leg_fltno(0)=}")
            info += f"  flt_no={path.get_leg_fltno(0)}"

        if carrier is not None:
            if self.airline and carrier != self.airline:
                return SnapshotInstruction(False, why=f"cause {carrier=}")
            info += f"  carrier={carrier}"

        # Now do something
        snapshot_file = self.filepath(sim, leg, path)
        title = f"{self.title}:{info}\n{time.strftime('Snapshot created %Y-%m-%d %A %I:%M:%S %p')}\n"
        if len(self.title) > 0 and not snapshot_file:
            print(f"{self.title}:{info}", flush=True)

        self._last_run_info = info

        if self.type in ["leg_untruncation", "path_untruncation"]:
            return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
        elif self.type in ("forecast", "forecast_adj"):
            return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
        elif self.type == "rm":
            bucket_detail = leg.print_bucket_detail()
            snapshot_file = self.filepath(sim, leg, path)
            if snapshot_file:
                with snapshot_file.open(mode="a") as f:
                    f.write(title)
                    f.write(bucket_detail)
            else:
                print(bucket_detail)
            return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
        elif self.type == "pro_bp":
            return SnapshotInstruction(True, snapshot_file, why=title, filter=self)

        raise ValueError("unknown snapshot filter type")

airline class-attribute instance-attribute 

airline: str = ''

dcp class-attribute instance-attribute 

dcp: list[int] = []

dest class-attribute instance-attribute 

dest: list[str] = []

directory class-attribute instance-attribute 

directory: pathlib.Path | None = None

flt_no class-attribute instance-attribute 

flt_no: list[int] = []

orig class-attribute instance-attribute 

orig: list[str] = []

sample class-attribute instance-attribute 

sample: list[int] = []

title class-attribute instance-attribute 

title: str = ''

trial class-attribute instance-attribute 

trial: list[int] = []

type class-attribute instance-attribute 

type: Literal[
    "forecast",
    "leg_untruncation",
    "path_untruncation",
    "rm",
    "pro_bp",
    "forecast_adj",
    None,
] = None

filepath 

filepath(sim, leg=None, path=None) -> pathlib.Path | None
Source code in passengersim/config/snapshot_filter.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def filepath(self, sim, leg=None, path=None) -> pathlib.Path | None:
    if self.directory is None:
        return None
    pth = self.directory
    if leg is not None:
        pth = pth.joinpath(f"carrier-{leg.carrier}")
    pth = pth.joinpath(f"dpc-{sim.last_dcp}")
    if leg is not None:
        pth = pth.joinpath(f"orig-{leg.orig}")
    elif path is not None:
        pth = pth.joinpath(f"orig-{path.orig}")
    if leg is not None:
        pth = pth.joinpath(f"dest-{leg.dest}")
    elif path is not None:
        pth = pth.joinpath(f"dest-{path.dest}")
    if leg is not None:
        pth = pth.joinpath(f"fltno-{leg.flt_no}")
    elif path is not None:
        pth = pth.joinpath(f"fltno-{path.get_leg_fltno(0)}")
    if sim.num_trials > 1:
        pth = pth.joinpath(f"trial-{sim.trial}")
    pth = pth.joinpath(f"sample-{sim.sample}")
    pth.parent.mkdir(parents=True, exist_ok=True)
    return pth.with_suffix(".log")

run 

run(
    sim, leg=None, path=None, carrier=None, why=False
) -> SnapshotInstruction
Source code in passengersim/config/snapshot_filter.py 
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
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
def run(
    self, sim, leg=None, path=None, carrier=None, why=False
) -> SnapshotInstruction:
    # Check the filter conditions
    info = ""

    if len(self.trial) > 0 and sim.trial not in self.trial and sim.num_trials > 1:
        return SnapshotInstruction(False, why=f"cause {sim.trial=}")
    info += f"  trial={sim.trial}"

    if len(self.sample) > 0 and sim.sample not in self.sample:
        return SnapshotInstruction(False, why=f"cause {sim.sample=}")
    info += f"  sample={sim.sample}"

    if len(self.dcp) > 0 and sim.last_dcp not in self.dcp:
        return SnapshotInstruction(False, why=f"cause {sim.last_dcp=}")
    info += f"  dcp={sim.last_dcp}"

    if leg is not None:
        if self.airline and leg.carrier != self.airline:
            return SnapshotInstruction(False, why=f"cause {leg.carrier=}")
        info += f"  carrier={leg.carrier}"

        if len(self.orig) > 0 and leg.orig not in self.orig:
            return SnapshotInstruction(False, why=f"cause {leg.orig=}")
        info += f"  orig={leg.orig}"

        if len(self.dest) > 0 and leg.dest not in self.dest:
            return SnapshotInstruction(False, why=f"cause {leg.dest=}")
        info += f"  dest={leg.dest}"

        if len(self.flt_no) > 0 and leg.flt_no not in self.flt_no:
            return SnapshotInstruction(False, why=f"cause {leg.flt_no=}")
        info += f"  flt_no={leg.flt_no}"

    if path is not None:
        if len(self.orig) > 0 and path.orig not in self.orig:
            return SnapshotInstruction(False, why=f"cause {path.orig=}")
        info += f"  orig={path.orig}"

        if len(self.dest) > 0 and path.dest not in self.dest:
            return SnapshotInstruction(False, why=f"cause {path.dest=}")
        info += f"  dest={path.dest}"

        if len(self.flt_no) > 0 and path.get_leg_fltno(0) not in self.flt_no:
            return SnapshotInstruction(False, why=f"cause {path.get_leg_fltno(0)=}")
        info += f"  flt_no={path.get_leg_fltno(0)}"

    if carrier is not None:
        if self.airline and carrier != self.airline:
            return SnapshotInstruction(False, why=f"cause {carrier=}")
        info += f"  carrier={carrier}"

    # Now do something
    snapshot_file = self.filepath(sim, leg, path)
    title = f"{self.title}:{info}\n{time.strftime('Snapshot created %Y-%m-%d %A %I:%M:%S %p')}\n"
    if len(self.title) > 0 and not snapshot_file:
        print(f"{self.title}:{info}", flush=True)

    self._last_run_info = info

    if self.type in ["leg_untruncation", "path_untruncation"]:
        return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
    elif self.type in ("forecast", "forecast_adj"):
        return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
    elif self.type == "rm":
        bucket_detail = leg.print_bucket_detail()
        snapshot_file = self.filepath(sim, leg, path)
        if snapshot_file:
            with snapshot_file.open(mode="a") as f:
                f.write(title)
                f.write(bucket_detail)
        else:
            print(bucket_detail)
        return SnapshotInstruction(True, snapshot_file, why=title, filter=self)
    elif self.type == "pro_bp":
        return SnapshotInstruction(True, snapshot_file, why=title, filter=self)

    raise ValueError("unknown snapshot filter type")

SnapshotInstruction

Source code in passengersim/config/snapshot_filter.py 
10
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
class SnapshotInstruction:
    def __init__(
        self,
        trigger: bool = False,
        filepath: pathlib.Path | None = None,
        why: str | None = None,
        filter: SnapshotFilter | None = None,
        mode: Literal["w", "a"] = "w",
    ):
        self.trigger = bool(trigger)
        """Has this snapshot been triggered."""
        self.why = why
        """Explanation of why snapshot is (or is not) triggered."""
        self.filepath = filepath
        """Where to save snapshot content."""
        self.filter = filter
        """A reference to the filter that spawned this instruction."""
        self.mode = mode
        """Write mode for new content, `w` overwrites existing file, `a` appends."""

    def __bool__(self) -> bool:
        return self.trigger

    def write(self, content: str = ""):
        """Write snapshot content to a file, or just print it"""
        if not content:
            return
        if self.filepath:
            with self.filepath.open(mode=self.mode) as f:
                f.write(self.why)
                f.write("\n")
                if isinstance(content, bytes):
                    content = content.decode("utf-8")
                elif not isinstance(content, str):
                    content = str(content)
                f.write(content)
                if content[-1] != "\n":
                    f.write("\n")
        else:
            if self.why:
                print(self.why)
            print(content)

    def write_more(self, content: str = ""):
        """Write additional snapshot content to a file, or just print it"""
        if not content:
            return
        if self.filepath:
            with self.filepath.open(mode="a") as f:
                if isinstance(content, bytes):
                    content = content.decode("utf-8")
                elif not isinstance(content, str):
                    content = str(content)
                f.write(content)
                if content[-1] != "\n":
                    f.write("\n")
        else:
            print(content)

filepath instance-attribute 

filepath = filepath

Where to save snapshot content.

filter instance-attribute 

filter = filter

A reference to the filter that spawned this instruction.

mode instance-attribute 

mode = mode

Write mode for new content, w overwrites existing file, a appends.

trigger instance-attribute 

trigger = bool(trigger)

Has this snapshot been triggered.

why instance-attribute 

why = why

Explanation of why snapshot is (or is not) triggered.

__bool__ 

__bool__() -> bool
Source code in passengersim/config/snapshot_filter.py 
30
31
def __bool__(self) -> bool:
    return self.trigger

__init__ 

__init__(
    trigger: bool = False,
    filepath: pathlib.Path | None = None,
    why: str | None = None,
    filter: SnapshotFilter | None = None,
    mode: Literal["w", "a"] = "w",
)
Source code in passengersim/config/snapshot_filter.py 
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
def __init__(
    self,
    trigger: bool = False,
    filepath: pathlib.Path | None = None,
    why: str | None = None,
    filter: SnapshotFilter | None = None,
    mode: Literal["w", "a"] = "w",
):
    self.trigger = bool(trigger)
    """Has this snapshot been triggered."""
    self.why = why
    """Explanation of why snapshot is (or is not) triggered."""
    self.filepath = filepath
    """Where to save snapshot content."""
    self.filter = filter
    """A reference to the filter that spawned this instruction."""
    self.mode = mode
    """Write mode for new content, `w` overwrites existing file, `a` appends."""

write 

write(content: str = '')

Write snapshot content to a file, or just print it

Source code in passengersim/config/snapshot_filter.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
def write(self, content: str = ""):
    """Write snapshot content to a file, or just print it"""
    if not content:
        return
    if self.filepath:
        with self.filepath.open(mode=self.mode) as f:
            f.write(self.why)
            f.write("\n")
            if isinstance(content, bytes):
                content = content.decode("utf-8")
            elif not isinstance(content, str):
                content = str(content)
            f.write(content)
            if content[-1] != "\n":
                f.write("\n")
    else:
        if self.why:
            print(self.why)
        print(content)

write_more 

write_more(content: str = '')

Write additional snapshot content to a file, or just print it

Source code in passengersim/config/snapshot_filter.py 
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
def write_more(self, content: str = ""):
    """Write additional snapshot content to a file, or just print it"""
    if not content:
        return
    if self.filepath:
        with self.filepath.open(mode="a") as f:
            if isinstance(content, bytes):
                content = content.decode("utf-8")
            elif not isinstance(content, str):
                content = str(content)
            f.write(content)
            if content[-1] != "\n":
                f.write("\n")
    else:
        print(content)