read_svalbard_v1(
path: str,
stride: Optional[int] = None,
read_dmin_m: Optional[float] = None,
read_dmax_m: Optional[float] = None,
**kwargs
) -> DASDataset
Read a DAS acquisition from the Svalbard DAS4Whale dataset
(Bouffaut et al., 2022, Front. Mar. Sci.).
Data is stored as MATLAB HDF5 .mat files. Each file covers a
subset of channels along the 120 km Svalbard fiber optic cable
(Longyearbyen to open ocean through Isfjorden).
Units
Data is already in nanostrain — no scaling required.
Parameters
path : str
Full path to the .mat file.
stride : int, optional
Channel subsampling factor (tr[::stride, :]).
Returns
DASDataset
Source code in dasexplorer\core\readers_lib\processed.py
| def read_svalbard_v1(
path: str,
stride: Optional[int] = None,
read_dmin_m: Optional[float] = None,
read_dmax_m: Optional[float] = None,
**kwargs,
) -> DASDataset:
######################################################################
### OPTODAS ASN/ ALCATEL SUBMARINE NETWORK - (.mat) SVALBARD-2020
### https://doi.org/10.5281/zenodo.5823343
######################################################################
"""
Read a DAS acquisition from the Svalbard DAS4Whale dataset
(Bouffaut et al., 2022, Front. Mar. Sci.).
Data is stored as MATLAB HDF5 .mat files. Each file covers a
subset of channels along the 120 km Svalbard fiber optic cable
(Longyearbyen to open ocean through Isfjorden).
Units
-----
Data is already in nanostrain — no scaling required.
Parameters
----------
path : str
Full path to the .mat file.
stride : int, optional
Channel subsampling factor (tr[::stride, :]).
Returns
-------
DASDataset
"""
import scipy.io as sio
mat = sio.loadmat(path)
# Data: (n_channels, n_time), already in nanostrain
tr = np.asarray(mat["data"]).astype(np.float32)
# Sampling parameters
fs_hz = float(np.asarray(mat["info_sampling_frequency_Hz"]).ravel()[0])
dt_s = float(np.asarray(mat["info_sample_interval_s"]).ravel()[0])
dx_m = float(np.asarray(mat["info_SSI_m"]).ravel()[0])
gl_m = float(np.asarray(mat["info_GL_m"]).ravel()[0])
# Units
#raw_units = np.asarray(mat["info_units"]).ravel()
#units = str(raw_units[0]) if raw_units.size > 0 else "nanostrain"
units = "nanostrain"
# Channel distances from shore [m] — use x1_distance_from_shore_m
dist_m = np.asarray(mat["x1_distance_from_shore_m"]).ravel().astype(np.float64)
if dist_m.shape[0] != tr.shape[0]:
dist_m = np.arange(tr.shape[0]) * dx_m
# Time vector [s]
time_s = np.asarray(mat["x2_time_s"]).ravel().astype(np.float64)
if time_s.shape[0] != tr.shape[1]:
time_s = np.arange(tr.shape[1]) * dt_s
# UTC start time from info_timestamp (string: e.g. '20200627_052441')
start_dt = None
try:
raw_ts = np.asarray(mat["info_timestamp"]).ravel()
ts_str = str(raw_ts[0]).strip()
start_dt = datetime.datetime.strptime(ts_str, "%Y%m%d_%H%M%S").replace(
tzinfo=datetime.timezone.utc
)
except Exception:
# Fallback: parse from filename YYYYMMDD_HHMMSS_ch...
m = re.search(r"(\d{8})_(\d{6})", os.path.basename(path))
if m:
start_dt = datetime.datetime.strptime(
m.group(1) + m.group(2), "%Y%m%d%H%M%S"
).replace(tzinfo=datetime.timezone.utc)
downsample = None
if stride is not None and stride > 1:
tr = tr[::stride, :]
dist_m = dist_m[::stride]
downsample = stride
# Spatial crop applied after stride.
# channel_offset = original (stride=1) cable index of the first kept channel.
channel_offset = 0
if read_dmin_m is not None or read_dmax_m is not None:
dmin = read_dmin_m if read_dmin_m is not None else float(dist_m[0])
dmax = read_dmax_m if read_dmax_m is not None else float(dist_m[-1])
mask = (dist_m >= dmin) & (dist_m <= dmax)
first_idx = int(np.argmax(mask))
channel_offset = first_idx * int(downsample or 1)
tr = tr[mask, :]
dist_m = dist_m[mask]
return DASDataset(
tr=tr,
dist_m=dist_m,
time_s=time_s,
fs_hz=fs_hz,
start_datetime_utc=start_dt,
filename=os.path.basename(path),
reader="svalbard_v1",
downsample=downsample,
channel_offset=channel_offset,
metadata={
"dx_m": dx_m,
"gauge_length_m": gl_m,
},
units=units,
)
|