0
# High-Level Functions
1
2
Convenience functions for complete file operations, data conversion, file manipulation, and batch processing tasks. The `pyedflib.highlevel` module provides simplified interfaces for common EDF/BDF operations without requiring detailed knowledge of the low-level API.
3
4
## Capabilities
5
6
### Complete File I/O
7
8
Read and write entire EDF files with single function calls, handling header parsing and data extraction automatically.
9
10
```python { .api }
11
def read_edf(edf_file: str, ch_nrs: Optional[List[int]] = None,
12
ch_names: Optional[List[str]] = None, digital: bool = False,
13
verbose: bool = False) -> Tuple[Union[np.ndarray, List[np.ndarray]], List[dict], dict]:
14
"""
15
Read complete EDF file with signal data and headers.
16
17
Parameters:
18
- edf_file: str, path to EDF file
19
- ch_nrs: List[int] or int or None, channel numbers to read (None = all)
20
- ch_names: List[str] or str or None, channel names to read (None = all)
21
- digital: bool, return digital values if True, physical values if False
22
- verbose: bool, print progress information
23
24
Returns:
25
Tuple containing:
26
- signals: numpy.ndarray or List[numpy.ndarray], signal data
27
- signal_headers: List[dict], signal header information
28
- header: dict, file header information
29
"""
30
31
def write_edf(edf_file: str, signals: Union[np.ndarray, List[np.ndarray]], signal_headers: List[Dict],
32
header: Optional[Dict] = None, digital: bool = False,
33
file_type: int = -1) -> bool:
34
"""
35
Write complete EDF file with signal data and headers.
36
37
Parameters:
38
- edf_file: str, output file path
39
- signals: numpy.ndarray, signal data (channels x samples or samples x channels)
40
- signal_headers: List[dict], signal header configurations
41
- header: dict or None, file header information
42
- digital: bool, whether signals contain digital values
43
- file_type: int, EDF file type (FILETYPE_EDF, FILETYPE_EDFPLUS, etc.)
44
- block_size: int or None, block size for writing
45
"""
46
47
def write_edf_quick(edf_file: str, signals: np.ndarray, sfreq: Union[float, int],
48
digital: bool = False) -> bool:
49
"""
50
Quick write EDF with minimal configuration.
51
52
Parameters:
53
- edf_file: str, output file path
54
- signals: numpy.ndarray, signal data
55
- sfreq: int or List[int], sampling frequency(ies)
56
- digital: bool, whether signals are digital values
57
"""
58
59
def read_edf_header(edf_file: str, read_annotations: bool = True) -> dict:
60
"""
61
Read only header information from EDF file.
62
63
Parameters:
64
- edf_file: str, path to EDF file
65
- read_annotations: bool, whether to read annotations
66
67
Returns:
68
dict: Complete file header with signal headers and annotations
69
"""
70
```
71
72
Usage examples:
73
74
```python
75
import pyedflib.highlevel as hl
76
import numpy as np
77
78
# Read complete file
79
signals, signal_headers, header = hl.read_edf('data.edf')
80
print(f"Read {len(signals)} channels, {signals[0].shape[0]} samples each")
81
82
# Read specific channels by number
83
signals, sig_hdrs, hdr = hl.read_edf('data.edf', ch_nrs=[0, 2, 4])
84
85
# Read specific channels by name
86
signals, sig_hdrs, hdr = hl.read_edf('data.edf', ch_names=['EEG Fp1', 'EEG C3'])
87
88
# Write complete file
89
sample_data = np.random.randn(4, 2560) # 4 channels, 2560 samples
90
sample_headers = [hl.make_signal_header(f'CH{i}', sample_frequency=256)
91
for i in range(4)]
92
hl.write_edf('output.edf', sample_data, sample_headers)
93
94
# Quick write with minimal setup
95
hl.write_edf_quick('quick.edf', sample_data, sfreq=256)
96
97
# Read only headers (fast)
98
header_info = hl.read_edf_header('data.edf')
99
signal_headers = header_info['SignalHeaders']
100
file_header = header_info
101
```
102
103
### Header Creation Utilities
104
105
Create properly formatted header dictionaries for file and signal configuration.
106
107
```python { .api }
108
def make_header(technician: str = '', recording_additional: str = '',
109
patientname: str = '', patient_additional: str = '',
110
patientcode: str = '', equipment: str = '', admincode: str = '',
111
sex: str = '', startdate: Optional[datetime] = None,
112
birthdate: str = '') -> Dict:
113
"""
114
Create file header dictionary with standard fields.
115
116
Parameters:
117
- technician: str, technician name
118
- recording_additional: str, additional recording information
119
- patientname: str, patient name
120
- patient_additional: str, additional patient information
121
- patientcode: str, patient identification code
122
- equipment: str, recording equipment description
123
- admincode: str, administration code
124
- sex: str, patient sex ('M', 'F', or '')
125
- startdate: datetime or None, recording start time (None = current time)
126
- birthdate: str, patient birthdate in YYYY-MM-DD format
127
128
Returns:
129
dict: File header dictionary
130
"""
131
132
def make_signal_header(label: str, dimension: str = 'uV', sample_frequency: int = 256,
133
physical_min: float = -200.0, physical_max: float = 200.0,
134
digital_min: int = -32768, digital_max: int = 32767,
135
transducer: str = '', prefilter: str = '') -> Dict:
136
"""
137
Create signal header dictionary with standard fields.
138
139
Parameters:
140
- label: str, signal label/name
141
- dimension: str, physical units (e.g., 'uV', 'mV', 'V')
142
- sample_frequency: int, sampling rate in Hz
143
- physical_min: float, minimum physical value
144
- physical_max: float, maximum physical value
145
- digital_min: int, minimum digital value
146
- digital_max: int, maximum digital value
147
- transducer: str, transducer description
148
- prefilter: str, prefilter description
149
150
Returns:
151
dict: Signal header dictionary
152
"""
153
154
def make_signal_headers(list_of_labels: List[str], dimension: str = 'uV',
155
sample_frequency: Optional[Union[int, float]] = 256,
156
physical_min: float = -200.0, physical_max: float = 200.0,
157
digital_min: Union[float, int] = -32768,
158
digital_max: Union[float, int] = 32767,
159
transducer: str = '', prefilter: str = '') -> List[Dict]:
160
"""
161
Create multiple signal headers from list of labels with common parameters.
162
163
Parameters:
164
- list_of_labels: List[str], signal labels/names
165
- dimension: str, physical units (e.g., 'uV', 'mV', 'V')
166
- sample_frequency: int or float or None, sampling rate in Hz
167
- physical_min: float, minimum physical value
168
- physical_max: float, maximum physical value
169
- digital_min: int or float, minimum digital value
170
- digital_max: int or float, maximum digital value
171
- transducer: str, transducer description
172
- prefilter: str, prefilter description
173
174
Returns:
175
List[dict]: Complete signal header dictionaries
176
"""
177
```
178
179
Usage examples:
180
181
```python
182
# Create complete file header
183
file_hdr = hl.make_header(
184
technician='Dr. Johnson',
185
patientname='Subject_001',
186
patientcode='S001',
187
equipment='EEG-64 System',
188
sex='F',
189
birthdate='1990-05-15'
190
)
191
192
# Create signal header
193
sig_hdr = hl.make_signal_header(
194
label='EEG Fp1',
195
dimension='uV',
196
sample_frequency=512,
197
physical_min=-500,
198
physical_max=500
199
)
200
201
# Create multiple headers from labels
202
labels = ['EEG Fp1', 'EEG Fp2', 'EEG C3', 'EEG C4']
203
signal_headers = hl.make_signal_headers(labels, sample_frequency=256, dimension='uV')
204
205
# Mix of labels and partial configs
206
mixed_headers = [
207
'EEG Fp1', # Will use defaults
208
{'label': 'ECG', 'dimension': 'mV', 'sample_frequency': 1000},
209
'EMG'
210
]
211
headers = hl.make_signal_headers(mixed_headers, sample_frequency=256)
212
```
213
214
### Data Conversion
215
216
Convert between digital and physical values using calibration parameters.
217
218
```python { .api }
219
def dig2phys(signal: Union[np.ndarray, int], dmin: int, dmax: int,
220
pmin: float, pmax: float) -> Union[np.ndarray, float]:
221
"""
222
Convert digital values to physical values.
223
224
Parameters:
225
- signal: numpy.ndarray or int, digital signal values
226
- dmin: int, digital minimum value
227
- dmax: int, digital maximum value
228
- pmin: float, physical minimum value
229
- pmax: float, physical maximum value
230
231
Returns:
232
numpy.ndarray or float: Physical values
233
"""
234
235
def phys2dig(signal: Union[np.ndarray, float], pmin: float, pmax: float,
236
dmin: int, dmax: int) -> Union[np.ndarray, int]:
237
"""
238
Convert physical values to digital values.
239
240
Parameters:
241
- signal: numpy.ndarray or float, physical signal values
242
- pmin: float, physical minimum value
243
- pmax: float, physical maximum value
244
- dmin: int, digital minimum value
245
- dmax: int, digital maximum value
246
247
Returns:
248
numpy.ndarray or int: Digital values
249
"""
250
```
251
252
Usage examples:
253
254
```python
255
# Convert digital signal to physical values (e.g., ADC counts to microvolts)
256
digital_data = np.array([1000, -500, 2000], dtype=np.int16)
257
physical_data = hl.dig2phys(digital_data, -32768, 32767, -500.0, 500.0)
258
259
# Convert physical signal to digital values
260
eeg_signal = np.array([100.5, -75.2, 200.1]) # microvolts
261
digital_values = hl.phys2dig(eeg_signal, -500.0, 500.0, -32768, 32767)
262
```
263
264
### File Manipulation
265
266
Modify existing EDF files by dropping channels, cropping time ranges, and renaming channels.
267
268
```python { .api }
269
def drop_channels(edf_source: str, edf_target: Optional[str] = None,
270
to_keep: Optional[Union[List[str], List[int]]] = None,
271
to_drop: Optional[Union[List[str], List[int]]] = None,
272
verbose: bool = False) -> str:
273
"""
274
Remove channels from EDF file.
275
276
Parameters:
277
- edf_source: str, source EDF file
278
- edf_target: str or None, output file path (None = auto-generated)
279
- to_keep: List[str or int] or None, channels to keep (overrides to_drop)
280
- to_drop: List[str or int] or None, channels to remove (by name or index)
281
- verbose: bool, print operation details
282
283
Returns:
284
str: Path to output file
285
"""
286
287
def crop_edf(edf_file: str, *, new_file: Optional[str] = None,
288
start: Optional[Union[datetime, int, float]] = None,
289
stop: Optional[Union[datetime, int, float]] = None,
290
start_format: str = "datetime", stop_format: str = "datetime",
291
verbose: bool = True) -> None:
292
"""
293
Crop EDF file to desired time range.
294
295
Parameters:
296
- edf_file: str, source EDF file
297
- new_file: str or None, output file path (None = auto-generated)
298
- start: datetime or int or float or None, new start time (datetime or seconds)
299
- stop: datetime or int or float or None, new stop time (datetime or seconds)
300
- start_format: str, format of start ("datetime" or "seconds")
301
- stop_format: str, format of stop ("datetime" or "seconds")
302
- verbose: bool, show progress
303
"""
304
305
def rename_channels(edf_file: str, mapping: Union[Dict[str, str], List[str]],
306
new_file: Optional[str] = None, verbose: bool = True) -> None:
307
"""
308
Rename channels in EDF file.
309
310
Parameters:
311
- edf_file: str, path to input EDF file
312
- mapping: dict or List[str], channel name mapping or new names list
313
- new_file: str or None, output file path (None = modify in-place)
314
- verbose: bool, print operation details
315
"""
316
317
def change_polarity(edf_file: str, channels: Union[List[str], str],
318
new_file: Optional[str] = None, verbose: bool = True) -> None:
319
"""
320
Change signal polarity (multiply by -1) for specified channels.
321
322
Parameters:
323
- edf_file: str, path to input EDF file
324
- channels: List[str] or str, channel names to invert
325
- new_file: str or None, output file path (None = modify in-place)
326
- verbose: bool, print operation details
327
"""
328
```
329
330
Usage examples:
331
332
```python
333
# Drop specific channels
334
hl.drop_channels('recording.edf', to_drop=['EMG1', 'EMG2', 'EOG'])
335
336
# Keep only EEG channels
337
hl.drop_channels('recording.edf', to_keep=['EEG Fp1', 'EEG Fp2', 'EEG C3', 'EEG C4'])
338
339
# Crop to specific time window (10-60 seconds)
340
hl.crop_edf('long_recording.edf', start_sec=10, end_sec=60, copy_file=True)
341
342
# Rename channels with mapping
343
channel_mapping = {
344
'EEG FP1-REF': 'EEG Fp1',
345
'EEG FP2-REF': 'EEG Fp2',
346
'EEG C3-REF': 'EEG C3'
347
}
348
hl.rename_channels('raw.edf', channel_mapping, new_file='renamed.edf')
349
350
# Rename with new names list
351
new_names = ['Ch1', 'Ch2', 'Ch3', 'Ch4']
352
hl.rename_channels('data.edf', new_names)
353
354
# Invert polarity of specific channels
355
hl.change_polarity('eeg.edf', channels=['EEG C3', 'EEG C4'])
356
```
357
358
### File Comparison and Validation
359
360
Compare EDF files and validate file integrity.
361
362
```python { .api }
363
def compare_edf(edf_file1: str, edf_file2: str, verbose: bool = True) -> bool:
364
"""
365
Compare two EDF files for differences in headers and data.
366
367
Parameters:
368
- edf_file1: str, path to first EDF file
369
- edf_file2: str, path to second EDF file
370
- verbose: bool, print detailed comparison results
371
372
Returns:
373
bool: True if files are identical, False otherwise
374
"""
375
```
376
377
Usage example:
378
379
```python
380
# Compare two files
381
are_identical = hl.compare_edf('original.edf', 'processed.edf')
382
if are_identical:
383
print("Files are identical")
384
else:
385
print("Files differ")
386
```
387
388
### Anonymization
389
390
Remove or modify patient-identifying information in EDF files for privacy compliance.
391
392
```python { .api }
393
def anonymize_edf(edf_file: str, new_file: Optional[str] = None,
394
to_remove: List[str] = ['patientname', 'birthdate'],
395
new_values: List[str] = ['xxx', ''],
396
verify: bool = False, verbose: bool = False) -> bool:
397
"""
398
Anonymize EDF file by replacing header fields.
399
400
Parameters:
401
- edf_file: str, source EDF file
402
- new_file: str or None, output file path (None = auto-generated)
403
- to_remove: List[str], header fields to replace
404
- new_values: List[str], replacement values
405
- verify: bool, compare files after anonymization
406
- verbose: bool, show progress
407
408
Returns:
409
bool: True if successful
410
"""
411
```
412
413
Usage example:
414
415
```python
416
# Remove patient identifying information
417
hl.anonymize_edf('patient_data.edf',
418
new_file='anonymous.edf',
419
to_remove=['patientname', 'patientcode', 'patient_additional'])
420
421
# Replace with anonymous values
422
new_values = {
423
'patientname': 'Anonymous',
424
'patientcode': 'ANON001',
425
'patient_additional': '',
426
'birthdate': '1900-01-01'
427
}
428
hl.anonymize_edf('identifiable.edf',
429
new_file='deidentified.edf',
430
new_values=new_values)
431
```
432
433
### Utility Functions
434
435
```python { .api }
436
def tqdm(iterable: Iterable, *args, **kwargs) -> Iterable:
437
"""
438
Progress bar wrapper for long operations.
439
440
Optional dependency - install with: pip install tqdm
441
Falls back to standard iterator if tqdm not available.
442
443
Parameters:
444
- iterable: Iterable, sequence to iterate over
445
- *args, **kwargs: Additional arguments passed to tqdm
446
447
Returns:
448
Iterable: Progress bar iterator or standard iterator
449
"""
450
```
451
452
## Complete High-Level Workflow Example
453
454
```python
455
import pyedflib.highlevel as hl
456
import numpy as np
457
from datetime import datetime
458
459
# Step 1: Read existing file
460
print("Reading EDF file...")
461
signals, signal_headers, header = hl.read_edf('raw_recording.edf', verbose=True)
462
463
print(f"Loaded {len(signals)} channels:")
464
for i, hdr in enumerate(signal_headers):
465
print(f" {i}: {hdr['label']} ({hdr['sample_frequency']} Hz)")
466
467
# Step 2: Process signals (example: apply high-pass filter simulation)
468
print("Processing signals...")
469
for i in range(len(signals)):
470
# Simple high-pass filter simulation (remove DC offset)
471
signals[i] = signals[i] - np.mean(signals[i])
472
473
# Step 3: Create new file with processed data
474
print("Creating processed file...")
475
new_header = hl.make_header(
476
technician='Automated Processing',
477
recording_additional='High-pass filtered',
478
patientname=header.get('patientname', 'Unknown'),
479
patientcode=header.get('patientcode', 'UNK'),
480
equipment='pyEDFlib processor'
481
)
482
483
# Update signal headers to reflect processing
484
for hdr in signal_headers:
485
hdr['prefilter'] = hdr.get('prefilter', '') + ' HP:DC_removed'
486
487
hl.write_edf('processed.edf', signals, signal_headers, new_header)
488
489
# Step 4: Crop to analysis window
490
print("Cropping to analysis window...")
491
hl.crop_edf('processed.edf', start_sec=30, end_sec=300) # 30-300 seconds
492
493
# Step 5: Remove artifact channels
494
print("Removing artifact channels...")
495
hl.drop_channels('processed.edf', to_drop=['EMG', 'EOG', 'ECG'])
496
497
# Step 6: Anonymize for sharing
498
print("Creating anonymized version...")
499
hl.anonymize_edf('processed.edf',
500
new_file='anonymous_processed.edf',
501
new_values={
502
'patientname': 'Subject_001',
503
'patientcode': 'S001',
504
'patient_additional': 'Age_group_20-30',
505
'birthdate': '1990-01-01'
506
})
507
508
# Step 7: Verify final result
509
print("Verifying final file...")
510
final_signals, final_headers, final_header = hl.read_edf_header('anonymous_processed.edf')
511
print(f"Final file has {len(final_headers)} channels")
512
print(f"Patient name: {final_header['patientname']}")
513
514
print("Processing complete!")
515
```