72 lines
2.7 KiB
Python
Executable File
72 lines
2.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Script d'extraction de données H5 calibrées (format 2026).
|
|
Lit calibrated_data/channel_X (valeurs physiques avec unités).
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import sys
|
|
import h5py
|
|
import numpy as np
|
|
|
|
def extract_window(file_path: str, channel: int, start_ts: int, duration_sec: int) -> dict:
|
|
try:
|
|
with h5py.File(file_path, 'r') as f:
|
|
# Métadonnées
|
|
meta = f['metadata']
|
|
sample_rate = meta.attrs['sample_rate_hz']
|
|
file_duration = meta.attrs['duration_sec']
|
|
total_samples = meta.attrs['n_samples']
|
|
|
|
# Dataset calibré
|
|
dataset = f[f'calibrated_data/channel_{channel}']
|
|
|
|
# Calcul indices (si start_ts = 0, on prend depuis le début)
|
|
start_idx = int(start_ts * sample_rate) if start_ts > 0 else 0
|
|
num_samples = int(duration_sec * sample_rate) if duration_sec > 0 else total_samples
|
|
end_idx = min(start_idx + num_samples, total_samples)
|
|
|
|
# Extraire
|
|
samples = dataset[start_idx:end_idx]
|
|
|
|
# Unité selon le canal
|
|
unit = 'm/s' if channel in [1, 2, 3] else 'Pa'
|
|
channel_name = f'Geophone {channel}' if channel in [1, 2, 3] else 'Hydrophone'
|
|
|
|
return {
|
|
"samples": samples.tolist(),
|
|
"start_idx": int(start_idx),
|
|
"end_idx": int(end_idx),
|
|
"total_samples": int(total_samples),
|
|
"sample_rate": int(sample_rate),
|
|
"duration_sec": float(file_duration),
|
|
"channel": channel,
|
|
"channel_name": channel_name,
|
|
"unit": unit,
|
|
"stats": {
|
|
"min": float(np.min(samples)),
|
|
"max": float(np.max(samples)),
|
|
"mean": float(np.mean(samples)),
|
|
"std": float(np.std(samples)),
|
|
"rms": float(np.sqrt(np.mean(samples**2)))
|
|
},
|
|
"source": "calibrated_h5_2026"
|
|
}
|
|
except Exception as e:
|
|
return {"error": str(e)}
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description='Extraction H5 calibré')
|
|
parser.add_argument('--file', required=True, help='Fichier H5')
|
|
parser.add_argument('--channel', type=int, required=True, help='Canal 1-4')
|
|
parser.add_argument('--start', type=int, default=0, help='Offset secondes (0=début)')
|
|
parser.add_argument('--duration', type=int, default=0, help='Durée secondes (0=tout)')
|
|
args = parser.parse_args()
|
|
|
|
result = extract_window(args.file, args.channel, args.start, args.duration)
|
|
print(json.dumps(result))
|
|
|
|
if __name__ == '__main__':
|
|
main()
|