Coverage for src / tracekit / exporters / matlab_export.py: 72%
119 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-11 23:04 +0000
1"""MATLAB export functionality.
3This module provides trace export to MATLAB .mat format with metadata.
6Example:
7 >>> from tracekit.exporters.matlab_export import export_mat
8 >>> export_mat(trace, "waveform.mat")
9 >>> export_mat({"ch1": ch1, "ch2": ch2}, "channels.mat")
11References:
12 MATLAB MAT-File Format (https://www.mathworks.com/help/pdf_doc/matlab/matfile_format.pdf)
13"""
15from __future__ import annotations
17from datetime import datetime
18from pathlib import Path
19from typing import Any
21import numpy as np
23try:
24 import scipy.io as sio
26 HAS_SCIPY = True
27except ImportError:
28 HAS_SCIPY = False
30try:
31 import h5py
33 HAS_H5PY = True
34except ImportError:
35 HAS_H5PY = False
37from tracekit.core.types import DigitalTrace, WaveformTrace
40def export_mat(
41 data: WaveformTrace | DigitalTrace | dict[str, Any],
42 path: str | Path,
43 *,
44 version: str = "5",
45 compression: bool = True,
46 include_metadata: bool = True,
47) -> None:
48 """Export data to MATLAB .mat format.
50 Args:
51 data: Data to export. Can be:
52 - Single WaveformTrace or DigitalTrace
53 - Dictionary mapping names to traces or data
54 path: Output file path.
55 version: MAT-file version ("5", "7.3"). Version 5 is more compatible.
56 Version 7.3 requires h5py and uses HDF5 backend for large files.
57 compression: Enable compression.
58 include_metadata: Include trace metadata in output.
60 Raises:
61 ImportError: If scipy is not installed, or h5py for version 7.3.
63 Raises:
64 TypeError: If data type is not supported.
66 Example:
67 >>> export_mat(trace, "waveform.mat")
68 >>> export_mat({"ch1": ch1, "ch2": ch2}, "channels.mat")
69 >>> export_mat(measurements, "results.mat", version="5")
71 Note:
72 Version 5 is the default and most compatible format, readable by
73 scipy.io.loadmat and MATLAB.
75 Version 7.3 uses HDF5 backend and supports:
76 - Files > 2 GB
77 - Compression
78 - But requires h5py and cannot be read by scipy.io.loadmat
80 References:
81 EXP-008
82 """
83 if not HAS_SCIPY:
84 raise ImportError("scipy is required for MATLAB export. Install with: pip install scipy")
86 path = Path(path)
88 # Prepare data dictionary for MATLAB
89 mat_dict: dict[str, Any] = {}
91 if isinstance(data, WaveformTrace | DigitalTrace):
92 # Single trace - use standard variable names
93 _add_trace_to_dict(mat_dict, "trace", data, include_metadata)
94 elif isinstance(data, dict): 94 ↛ 102line 94 didn't jump to line 102 because the condition on line 94 was always true
95 for name, value in data.items():
96 if isinstance(value, WaveformTrace | DigitalTrace):
97 _add_trace_to_dict(mat_dict, name, value, include_metadata)
98 else:
99 # Convert numpy arrays and other types
100 mat_dict[_sanitize_varname(name)] = _convert_value(value)
101 else:
102 raise TypeError(f"Unsupported data type: {type(data)}")
104 # Add export metadata
105 # Note: MATLAB field names cannot start with underscore
106 if include_metadata: 106 ↛ 114line 106 didn't jump to line 114 because the condition on line 106 was always true
107 mat_dict["tracekit_export"] = {
108 "version": "1.0",
109 "exported_at": datetime.now().isoformat(),
110 "format": "tracekit_matlab",
111 }
113 # Save to .mat file
114 if version == "7.3":
115 # Use HDF5 backend (requires h5py)
116 if not HAS_H5PY: 116 ↛ 117line 116 didn't jump to line 117 because the condition on line 116 was never true
117 raise ImportError(
118 "h5py is required for MATLAB v7.3 export. "
119 "Install with: pip install h5py, or use version='5'"
120 )
121 _save_hdf5_mat(path, mat_dict, compression)
122 else:
123 # Version 5 format (default, most compatible)
124 sio.savemat(
125 str(path),
126 mat_dict,
127 do_compression=compression,
128 oned_as="column",
129 )
132def _save_hdf5_mat(path: Path, mat_dict: dict[str, Any], compression: bool) -> None:
133 """Save MATLAB 7.3 format file using h5py (HDF5 backend).
135 Args:
136 path: Output file path.
137 mat_dict: Dictionary of MATLAB variables.
138 compression: Enable compression.
139 """
140 compression_opts = "gzip" if compression else None
142 with h5py.File(path, "w") as f:
143 # Set MATLAB 7.3 header attributes
144 f.attrs["MATLAB_class"] = np.bytes_("struct")
146 for key, value in mat_dict.items():
147 _write_hdf5_value(f, key, value, compression_opts)
150def _write_hdf5_value(
151 parent: h5py.File | h5py.Group, key: str, value: Any, compression: str | None
152) -> None:
153 """Write a value to HDF5 file in MATLAB 7.3 compatible format.
155 Args:
156 parent: HDF5 file or group object.
157 key: Variable name.
158 value: Value to write.
159 compression: Compression algorithm.
160 """
161 if isinstance(value, np.ndarray):
162 # Create dataset for arrays
163 if compression and value.size > 100: 163 ↛ 166line 163 didn't jump to line 166 because the condition on line 163 was always true
164 parent.create_dataset(key, data=value, compression=compression)
165 else:
166 parent.create_dataset(key, data=value)
167 # Set MATLAB class attribute
168 parent[key].attrs["MATLAB_class"] = np.bytes_("double")
169 elif isinstance(value, dict):
170 # Create group for structs/dicts
171 grp = parent.create_group(key)
172 grp.attrs["MATLAB_class"] = np.bytes_("struct")
173 for k, v in value.items():
174 _write_hdf5_value(grp, k, v, compression)
175 elif isinstance(value, str):
176 # String as uint16 array (MATLAB format)
177 dt = h5py.string_dtype(encoding="utf-8")
178 parent.create_dataset(key, data=value, dtype=dt)
179 parent[key].attrs["MATLAB_class"] = np.bytes_("char")
180 elif isinstance(value, int | float): 180 ↛ 184line 180 didn't jump to line 184 because the condition on line 180 was always true
181 # Scalar as 1x1 array
182 parent.create_dataset(key, data=np.array([[value]]))
183 parent[key].attrs["MATLAB_class"] = np.bytes_("double")
184 elif isinstance(value, bool):
185 parent.create_dataset(key, data=np.array([[value]], dtype=np.uint8))
186 parent[key].attrs["MATLAB_class"] = np.bytes_("logical")
187 elif isinstance(value, list):
188 # Convert list to array
189 arr = np.array(value)
190 if compression and arr.size > 100:
191 parent.create_dataset(key, data=arr, compression=compression)
192 else:
193 parent.create_dataset(key, data=arr)
194 parent[key].attrs["MATLAB_class"] = np.bytes_("double")
197def _add_trace_to_dict(
198 mat_dict: dict[str, Any],
199 name: str,
200 trace: WaveformTrace | DigitalTrace,
201 include_metadata: bool,
202) -> None:
203 """Add trace to MATLAB dictionary with metadata.
205 Args:
206 mat_dict: MATLAB variable dictionary.
207 name: Variable name for trace.
208 trace: Trace to add.
209 include_metadata: Include metadata fields.
210 """
211 name = _sanitize_varname(name)
213 # Add waveform data
214 if isinstance(trace, WaveformTrace):
215 mat_dict[f"{name}_data"] = trace.data
216 mat_dict[f"{name}_time"] = trace.time_vector
217 else: # DigitalTrace
218 mat_dict[f"{name}_data"] = trace.data.astype(np.uint8)
219 mat_dict[f"{name}_time"] = trace.time_vector
221 # Add metadata as struct
222 if include_metadata: 222 ↛ exitline 222 didn't return from function '_add_trace_to_dict' because the condition on line 222 was always true
223 meta = trace.metadata
224 metadata_struct: dict[str, Any] = {
225 "sample_rate": meta.sample_rate,
226 "time_base": meta.time_base,
227 "num_samples": len(trace.data),
228 "duration": trace.duration,
229 }
231 if meta.vertical_scale is not None:
232 metadata_struct["vertical_scale"] = meta.vertical_scale
233 if meta.vertical_offset is not None:
234 metadata_struct["vertical_offset"] = meta.vertical_offset
235 if meta.acquisition_time is not None:
236 metadata_struct["acquisition_time"] = meta.acquisition_time.isoformat()
237 if meta.source_file is not None:
238 metadata_struct["source_file"] = str(meta.source_file)
239 if meta.channel_name is not None:
240 metadata_struct["channel_name"] = meta.channel_name
241 if meta.trigger_info: 241 ↛ 242line 241 didn't jump to line 242 because the condition on line 241 was never true
242 metadata_struct["trigger_info"] = meta.trigger_info
244 metadata_struct["trace_type"] = (
245 "waveform" if isinstance(trace, WaveformTrace) else "digital"
246 )
248 mat_dict[f"{name}_metadata"] = metadata_struct
251def _sanitize_varname(name: str) -> str:
252 """Sanitize variable name for MATLAB compatibility.
254 Args:
255 name: Variable name to sanitize.
257 Returns:
258 Sanitized variable name compatible with MATLAB.
260 Note:
261 MATLAB variable names must:
262 - Start with a letter
263 - Contain only letters, digits, and underscores
264 - Be <= 63 characters
265 """
266 import re
268 # Replace invalid characters with underscores
269 name = re.sub(r"[^a-zA-Z0-9_]", "_", name)
271 # Ensure starts with letter
272 if name and not name[0].isalpha(): 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 name = "var_" + name
275 # Truncate to 63 characters
276 if len(name) > 63: 276 ↛ 277line 276 didn't jump to line 277 because the condition on line 276 was never true
277 name = name[:63]
279 return name if name else "var"
282def _convert_value(value: Any) -> Any:
283 """Convert Python value to MATLAB-compatible format.
285 Args:
286 value: Python value to convert.
288 Returns:
289 MATLAB-compatible representation of the value.
290 """
291 if isinstance(value, np.ndarray):
292 return value
293 if isinstance(value, list): 293 ↛ 294line 293 didn't jump to line 294 because the condition on line 293 was never true
294 return np.array(value)
295 if isinstance(value, dict): 295 ↛ 297line 295 didn't jump to line 297 because the condition on line 295 was never true
296 # Convert nested dict
297 return {_sanitize_varname(k): _convert_value(v) for k, v in value.items()}
298 if isinstance(value, str | int | float | bool): 298 ↛ 300line 298 didn't jump to line 300 because the condition on line 298 was always true
299 return value
300 if isinstance(value, complex):
301 return value
302 if isinstance(value, datetime):
303 return value.isoformat()
304 # For other types, try converting to string
305 return str(value)
308def export_multi_trace_mat(
309 traces: list[WaveformTrace | DigitalTrace],
310 path: str | Path,
311 *,
312 names: list[str] | None = None,
313 version: str = "5",
314 include_metadata: bool = True,
315) -> None:
316 """Export multiple traces to single MATLAB file.
318 Args:
319 traces: List of traces to export.
320 path: Output file path.
321 names: Variable names for each trace. If not provided, uses trace_1, trace_2, etc.
322 version: MAT-file version ("5", "7.3").
323 include_metadata: Include trace metadata.
325 Raises:
326 ValueError: If number of names does not match number of traces.
328 Example:
329 >>> export_multi_trace_mat(
330 ... [ch1, ch2, ch3],
331 ... "channels.mat",
332 ... names=["ch1", "ch2", "ch3"]
333 ... )
335 References:
336 EXP-008
337 """
338 if names is None:
339 names = [f"trace_{i + 1}" for i in range(len(traces))]
341 if len(names) != len(traces):
342 raise ValueError("Number of names must match number of traces")
344 # Create dictionary mapping names to traces
345 trace_dict = dict(zip(names, traces, strict=True))
347 # Export using main function
348 export_mat(trace_dict, path, version=version, include_metadata=include_metadata)
351__all__ = [
352 "export_mat",
353 "export_multi_trace_mat",
354]