epos-ai/epos-ai-picking-tools
3
0
Fork 1
Code Issues Pull Requests Actions Packages Projects Releases 1 Activity

ISEPOS-2350 improve validation of input stream

Browse Source
This commit is contained in:
Hubert Siejkowski 2025-04-10 14:54:49 +02:00
parent 9b69f42dc2
commit 75618c9a0d
1 changed files with 13 additions and 4 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

17
src/epos_ai_picking_tools/model_runner.py
View File

@ -73,16 +73,25 @@ class ModelRunner:
fpath = self.output_dir / f"{stream_path.stem}_annotations.mseed" fpath = self.output_dir / f"{stream_path.stem}_annotations.mseed"
ann.write(fpath, format="MSEED") ann.write(fpath, format="MSEED")
@staticmethod def validate_stream(self, stream):
def validate_stream(stream): model = self.model
groups = defaultdict(list) groups = defaultdict(list)
samples = defaultdict(list)
for trace in stream: for trace in stream:
groups[trace.stats.station].append(trace.stats.channel[-1]) groups[trace.stats.station].append(trace.stats.channel[-1])
samples[trace.stats.station].append(
float(trace.stats.npts / trace.stats.sampling_rate)
)
number_of_channels = list(map(len, groups.values())) number_of_channels = list(map(len, groups.values()))
lenght_of_traces = list(map(max , samples.values()))
if max(number_of_channels) < 3: if max(number_of_channels) < model.in_channels:
exit_error("Not enough traces in the stream") exit_error("Not enough channels in the stream.")
minimal_trace_length = model.in_samples / model.sampling_rate
if max(lenght_of_traces) < minimal_trace_length:
exit_error(f"All traces are shorter than required {int(minimal_trace_length):d} seconds")
def find_picks(self, stream_file_name, save_annotations=True): def find_picks(self, stream_file_name, save_annotations=True):
stream_path = pathlib.Path(stream_file_name) stream_path = pathlib.Path(stream_file_name)