0
# I/O Operations
1

2
Core functionality for reading and writing WFDB records and annotations, including database access, data source management, and file utilities. This module provides the fundamental data structures and functions needed to work with physiological signal data in WFDB format.
3

4
## Capabilities
5

6
### Record Reading
7

8
Read WFDB record files with flexible options for sample ranges, channel selection, and data format.
9

10
```python { .api }
11
def rdheader(record_name: str, pn_dir: str = None, rd_segments: bool = False) -> Union[Record, MultiRecord]:
12
    """
13
    Read WFDB header file and return Record or MultiRecord object.
14
    
15
    Parameters:
16
    - record_name: str, record name without extension
17
    - pn_dir: str, optional PhysioNet database directory
18
    - rd_segments: bool, read segment headers for multi-segment records
19
    
20
    Returns:
21
    Record or MultiRecord object with metadata only
22
    """
23

24
def rdrecord(record_name: str, sampfrom: int = 0, sampto: Union[int, str] = None, 
25
             channels: List[int] = None, physical: bool = True, pn_dir: str = None,
26
             m2s: bool = True, smooth_frames: bool = True, ignore_skew: bool = False,
27
             return_res: int = 64, force_channels: bool = True, 
28
             channel_names: List[str] = None, warn_empty: bool = False) -> Union[Record, MultiRecord]:
29
    """
30
    Read WFDB record with signal data.
31
    
32
    Parameters:
33
    - record_name: str, record name without extension
34
    - sampfrom: int, starting sample number (0-indexed)
35
    - sampto: int or 'end', ending sample number
36
    - channels: list of int, channel indices to read
37
    - physical: bool, return physical units (True) or digital (False)
38
    - pn_dir: str, PhysioNet database directory
39
    - m2s: bool, convert MultiRecord to Record
40
    - smooth_frames: bool, smooth multi-sample per frame signals
41
    - ignore_skew: bool, ignore signal skew
42
    - return_res: int, return resolution (8, 16, 32, 64)
43
    - force_channels: bool, force channel matching for variable layout
44
    - channel_names: list of str, channel names to read
45
    - warn_empty: bool, warn if no signals returned
46
    
47
    Returns:
48
    Record or MultiRecord object with signal data
49
    """
50

51
def rdsamp(record_name: str, sampfrom: int = 0, sampto: Union[int, str] = None,
52
           channels: List[int] = None, pn_dir: str = None, 
53
           channel_names: List[str] = None, warn_empty: bool = False,
54
           return_res: int = 64) -> Tuple[np.ndarray, Dict[str, Any]]:
55
    """
56
    Read WFDB record signals and basic metadata.
57
    
58
    Parameters:
59
    - record_name: str, record name
60
    - sampfrom: int, starting sample number
61
    - sampto: int or 'end', ending sample number
62
    - channels: list of int, channel indices to read
63
    - pn_dir: str, PhysioNet database directory
64
    - channel_names: list of str, channel names to read
65
    - warn_empty: bool, warn if no signals returned
66
    - return_res: int, return resolution
67
    
68
    Returns:
69
    - signals: ndarray, physical signal array (MxN)
70
    - fields: dict, metadata including fs, units, sig_name, comments
71
    """
72
```
73

74
### Record Writing
75

76
Write single-segment WFDB records with comprehensive metadata support.
77

78
```python { .api }
79
def wrsamp(record_name: str, fs: float, units: List[str], sig_name: List[str],
80
           p_signal: np.ndarray = None, d_signal: np.ndarray = None,
81
           e_p_signal: Union[np.ndarray, List] = None, e_d_signal: Union[np.ndarray, List] = None,
82
           samps_per_frame: List[int] = None, fmt: List[str] = None,
83
           adc_gain: List[float] = None, baseline: List[int] = None,
84
           comments: List[str] = None, base_time: datetime.time = None,
85
           base_date: datetime.date = None, base_datetime: datetime.datetime = None,
86
           write_dir: str = "") -> None:
87
    """
88
    Write single-segment WFDB record.
89
    
90
    Parameters:
91
    - record_name: str, record name
92
    - fs: float, sampling frequency in Hz
93
    - units: list of str, signal units for each channel
94
    - sig_name: list of str, signal names for each channel
95
    - p_signal: ndarray, physical signals (MxN array)
96
    - d_signal: ndarray, digital signals (MxN array)
97
    - e_p_signal: ndarray or list, expanded physical signals
98
    - e_d_signal: ndarray or list, expanded digital signals
99
    - samps_per_frame: list of int, samples per frame for each channel
100
    - fmt: list of str, WFDB format for each channel
101
    - adc_gain: list of float, ADC gain values
102
    - baseline: list of int, digital baseline values
103
    - comments: list of str, header comments
104
    - base_time: datetime.time, record start time
105
    - base_date: datetime.date, record start date
106
    - base_datetime: datetime.datetime, combined date/time
107
    - write_dir: str, output directory
108
    """
109
```
110

111
### Annotation I/O
112

113
Read and write WFDB annotation files with support for various annotation types and metadata.
114

115
```python { .api }
116
def rdann(record_name: str, extension: str, sampfrom: int = 0, 
117
          sampto: Union[int, str] = 'end', shift_samps: bool = False,
118
          pn_dir: str = None, summarize_labels: bool = False,
119
          return_label_elements: List[str] = ['symbol'], 
120
          encoding: str = None) -> Annotation:
121
    """
122
    Read WFDB annotation file.
123
    
124
    Parameters:
125
    - record_name: str, record name without extension
126
    - extension: str, annotation file extension (e.g., 'atr', 'qrs')
127
    - sampfrom: int, starting sample number
128
    - sampto: int or 'end', ending sample number
129
    - shift_samps: bool, shift sample numbers by sampfrom
130
    - pn_dir: str, PhysioNet database directory
131
    - summarize_labels: bool, summarize contained labels
132
    - return_label_elements: list of str, label elements to return
133
    - encoding: str, file encoding
134
    
135
    Returns:
136
    Annotation object
137
    """
138

139
def wrann(record_name: str, extension: str, sample: Union[List[int], np.ndarray],
140
          symbol: List[str] = None, subtype: Union[List[int], np.ndarray] = None,
141
          chan: Union[List[int], np.ndarray] = None, 
142
          num: Union[List[int], np.ndarray] = None,
143
          aux_note: List[str] = None, fs: float = None, 
144
          write_dir: str = "") -> None:
145
    """
146
    Write WFDB annotation file.
147
    
148
    Parameters:
149
    - record_name: str, record name
150
    - extension: str, annotation file extension
151
    - sample: array-like, sample locations for annotations
152
    - symbol: list of str, annotation symbols
153
    - subtype: array-like, annotation subtypes
154
    - chan: array-like, signal channels for annotations
155
    - num: array-like, annotation numbers
156
    - aux_note: list of str, auxiliary notes
157
    - fs: float, sampling frequency
158
    - write_dir: str, output directory
159
    """
160

161
def show_ann_labels() -> None:
162
    """Display standard WFDB annotation labels and their meanings."""
163

164
def show_ann_classes() -> None:
165
    """Display standard WFDB annotation classes and categories."""
166

167
def mrgann(record_name: str, extension: str, ann_1: Annotation, ann_2: Annotation,
168
           fs: float, write_dir: str = "") -> None:
169
    """
170
    Merge two annotation objects and write to file.
171
    
172
    Parameters:
173
    - record_name: str, output record name
174
    - extension: str, output annotation extension
175
    - ann_1: Annotation, first annotation object
176
    - ann_2: Annotation, second annotation object  
177
    - fs: float, sampling frequency
178
    - write_dir: str, output directory
179
    """
180
```
181

182
### Database Access
183

184
Download and access PhysioNet databases and records.
185

186
```python { .api }
187
def dl_database(db_dir: str, dl_dir: str, records: Union[List[str], str] = "all",
188
                annotators: Union[List[str], str, None] = "all", 
189
                keep_subdirs: bool = True, overwrite: bool = False) -> None:
190
    """
191
    Download complete PhysioNet database.
192
    
193
    Parameters:
194
    - db_dir: str, PhysioNet database directory name
195
    - dl_dir: str, local download directory path
196
    - records: list of str or 'all', specific records to download
197
    - annotators: list of str, 'all', or None, annotation types to download
198
    - keep_subdirs: bool, preserve directory structure
199
    - overwrite: bool, overwrite existing files
200
    """
201

202
def dl_files(db: str, dl_dir: str, files: List[str], 
203
             keep_subdirs: bool = True, overwrite: bool = False) -> None:
204
    """
205
    Download specific files from PhysioNet database.
206
    
207
    Parameters:
208
    - db: str, database name
209
    - dl_dir: str, local download directory
210
    - files: list of str, specific files to download
211
    - keep_subdirs: bool, preserve directory structure
212
    - overwrite: bool, overwrite existing files
213
    """
214

215
def get_dbs() -> List[str]:
216
    """
217
    Get list of available PhysioNet databases.
218
    
219
    Returns:
220
    List of database names and descriptions
221
    """
222

223
def get_record_list(db_dir: str, records: Union[List[str], str] = "all") -> List[str]:
224
    """
225
    Get record list for a PhysioNet database.
226
    
227
    Parameters:
228
    - db_dir: str, database directory name
229
    - records: list of str or 'all', specific records or all
230
    
231
    Returns:
232
    List of record names in the database
233
    """
234

235
def set_db_index_url(db_index_url: str) -> None:
236
    """
237
    Set custom database index URL for PhysioNet access.
238
    
239
    Parameters:
240
    - db_index_url: str, URL to database index
241
    """
242
```
243

244
### Data Source Management
245

246
Configure and manage data sources for accessing local and remote databases.
247

248
```python { .api }
249
def show_data_sources() -> None:
250
    """Display currently configured data sources."""
251

252
def add_data_source(ds: DataSource) -> None:
253
    """
254
    Add new data source configuration.
255
    
256
    Parameters:
257
    - ds: DataSource, data source object to add
258
    """
259

260
def remove_data_source(ds_name: str) -> None:
261
    """
262
    Remove data source by name.
263
    
264
    Parameters:
265
    - ds_name: str, name of data source to remove
266
    """
267

268
def reset_data_sources(keep_pn: bool = False) -> None:
269
    """
270
    Reset data sources to default configuration.
271
    
272
    Parameters:
273
    - keep_pn: bool, keep PhysioNet data source
274
    """
275
```
276

277
### Utility Functions
278

279
Additional utilities for record information and time conversion.
280

281
```python { .api }
282
def sampfreq(record_name: str, pn_dir: str = None) -> None:
283
    """
284
    Display sampling frequency for each signal in a record.
285
    Prints signal names and frequencies to stdout.
286
    
287
    Parameters:
288
    - record_name: str, record name
289
    - pn_dir: str, PhysioNet database directory
290
    
291
    Returns:
292
    None - prints output to stdout
293
    """
294

295
def signame(record_name: str, pn_dir: str = None, sig_nums: List[int] = []) -> None:
296
    """
297
    Display signal names for a record.
298
    Prints signal names to stdout.
299
    
300
    Parameters:
301
    - record_name: str, record name
302
    - pn_dir: str, PhysioNet database directory
303
    - sig_nums: list of int, specific signal numbers to display
304
    
305
    Returns:
306
    None - prints output to stdout
307
    """
308

309
def wfdbdesc(record_name: str, pn_dir: str = None) -> None:
310
    """
311
    Display detailed record information.
312
    Prints comprehensive record metadata to stdout.
313
    
314
    Parameters:
315
    - record_name: str, record name
316
    - pn_dir: str, PhysioNet database directory
317
    
318
    Returns:
319
    None - prints output to stdout
320
    """
321

322
def wfdbtime(record_name: str, input_times: List[Union[int, str]], 
323
             pn_dir: str = None) -> None:
324
    """
325
    Convert and display time formats for a record.
326
    Prints formatted time information to stdout.
327
    
328
    Parameters:
329
    - record_name: str, record name
330
    - input_times: list, time values to convert
331
    - pn_dir: str, PhysioNet database directory
332
    
333
    Returns:
334
    None - prints output to stdout
335
    """
336
```
337

338
## Types
339

340
```python { .api }
341
class Record:
342
    """Single-segment WFDB record representation."""
343
    record_name: str          # Record name without extension
344
    n_sig: int               # Number of signals
345
    fs: float                # Sampling frequency in Hz  
346
    sig_len: int             # Signal length in samples
347
    p_signal: np.ndarray     # Physical signal values (MxN array)
348
    d_signal: np.ndarray     # Digital signal values (MxN array)
349
    e_p_signal: Union[np.ndarray, List]  # Expanded physical signals
350
    e_d_signal: Union[np.ndarray, List]  # Expanded digital signals
351
    sig_name: List[str]      # Signal names for each channel
352
    units: List[str]         # Units for each channel
353
    file_name: List[str]     # Data file names
354
    fmt: List[str]           # WFDB format for each channel
355
    adc_gain: List[float]    # ADC gain values
356
    baseline: List[int]      # Digital baseline values
357
    comments: List[str]      # Header comments
358
    base_time: datetime.time # Record start time
359
    base_date: datetime.date # Record start date
360
    base_datetime: datetime.datetime  # Combined date/time
361
    
362
    def get_frame_number(self, time_value: Union[int, float, str]) -> int: ...
363
    def get_elapsed_time(self, time_value: Union[int, float, str]) -> float: ...
364
    def get_absolute_time(self, time_value: Union[int, float, str]) -> datetime.datetime: ...
365
    def wrsamp(self, expanded: bool = False, write_dir: str = "") -> None: ...
366
    def to_dataframe(self) -> pd.DataFrame: ...
367
    def check_field(self, field: str, required_channels: Union[str, List[int]] = "all") -> bool: ...
368

369
class MultiRecord:
370
    """Multi-segment WFDB record representation."""
371
    segments: List[Union[Record, None]]  # List of Record objects or None for empty segments
372
    layout: str                         # "fixed" or "variable" layout
373
    seg_name: List[str]                 # Segment names
374
    seg_len: List[int]                  # Length of each segment
375
    sig_segments: List[List[int]]       # Signal segments
376
    
377
    def multi_to_single(self, physical: bool, return_res: int = 64, 
378
                       expanded: bool = False) -> Record: ...
379
    def wrsamp(self, write_dir: str = "") -> None: ...
380

381
class Annotation:
382
    """WFDB annotation representation."""
383
    record_name: str         # Associated record name
384
    extension: str           # Annotation file extension
385
    sample: np.ndarray       # Annotation sample locations
386
    symbol: List[str]        # Annotation symbols
387
    subtype: np.ndarray      # Annotation subtypes
388
    chan: np.ndarray         # Signal channels for annotations
389
    num: np.ndarray          # Annotation numbers
390
    aux_note: List[str]      # Auxiliary notes
391
    fs: float               # Sampling frequency
392
    label_store: np.ndarray  # Integer label encoding
393
    description: List[str]   # Label descriptions
394
    custom_labels: pd.DataFrame     # Custom annotation labels
395
    contained_labels: pd.DataFrame  # Labels present in annotation
396
    
397
    def apply_range(self, sampfrom: int = 0, sampto: int = None) -> Annotation: ...
398
    def wrann(self, write_fs: bool = False, write_dir: str = "") -> None: ...
399

400
class DataSource:
401
    """Data source configuration."""
402
    name: str                # Data source name
403
    ds_type: DataSourceType  # LOCAL or HTTP
404
    uri: str                # Data source URI
405

406
DataSourceType = Literal["LOCAL", "HTTP"]
407

408
SIGNAL_CLASSES: pd.DataFrame  # Signal classification information
409
```
410

411
## Usage Examples
412

413
### Basic Record Operations
414

415
```python
416
import wfdb
417

418
# Read header only
419
record = wfdb.rdheader('100', pn_dir='mitdb')
420
print(f"Record has {record.n_sig} signals at {record.fs} Hz")
421

422
# Read specific channels and time range
423
record = wfdb.rdrecord('100', pn_dir='mitdb', 
424
                       sampfrom=0, sampto=3600,  # First 10 seconds
425
                       channels=[0, 1])           # First two channels
426

427
# Read signals without full record object
428
signals, fields = wfdb.rdsamp('100', pn_dir='mitdb', channels=[0])
429
print(f"Signal shape: {signals.shape}")
430
print(f"Sampling frequency: {fields['fs']} Hz")
431
```
432

433
### Annotation Handling
434

435
```python
436
import wfdb
437

438
# Read annotations
439
ann = wfdb.rdann('100', 'atr', pn_dir='mitdb')
440
print(f"Found {len(ann.sample)} annotations")
441
print(f"First 5 symbols: {ann.symbol[:5]}")
442

443
# Filter annotations by time range
444
ann_subset = ann.apply_range(sampfrom=1000, sampto=5000)
445

446
# Write annotations
447
wfdb.wrann('output', 'atr', 
448
           sample=[100, 200, 300],
449
           symbol=['N', 'V', 'N'])
450
```
451

452
### Database Access
453

454
```python
455
import wfdb
456

457
# List available databases
458
dbs = wfdb.get_dbs()
459
print(f"Found {len(dbs)} databases")
460

461
# Download specific record
462
wfdb.dl_files('mitdb', './data', ['100.hea', '100.dat', '100.atr'])
463

464
# Download entire database  
465
wfdb.dl_database('mitdb', './mitdb_data', records=['100', '101', '102'])
466
```