Format Conversion

def read_edf(file_name: str, pn_dir: str = None, rd_time: bool = True,
             digital: bool = False, header_only: bool = False, 
             verbose: bool = False) -> Record:
    """
    Read EDF/EDF+ files and convert to WFDB Record format.
    
    Parameters:
    - file_name: str, EDF file name or path
    - pn_dir: str, directory containing the file
    - rd_time: bool, read time information from EDF+ 
    - digital: bool, return digital values instead of physical
    - header_only: bool, read header information only
    - verbose: bool, print detailed file information
    
    Returns:
    Record object containing the EDF data
    """

def wfdb_to_edf(record_name: str, pn_dir: str = None, sampfrom: int = 0,
                sampto: Union[int, str] = 'end', channels: List[int] = None,
                write_dir: str = '', edf_file: str = None, 
                file_type: str = 'EDF', digital: bool = False,
                header_only: bool = False) -> None:
    """
    Convert WFDB record to EDF format.
    
    Parameters:
    - record_name: str, WFDB record name
    - pn_dir: str, directory containing WFDB files
    - sampfrom: int, starting sample for conversion
    - sampto: int or 'end', ending sample for conversion
    - channels: list of int, specific channels to convert
    - write_dir: str, output directory for EDF file
    - edf_file: str, output EDF file name
    - file_type: str, 'EDF' or 'EDF+' format
    - digital: bool, write digital values instead of physical
    - header_only: bool, write header only
    """

def rdedfann(edf_file: str, ann_file: str = None, ann_ext: str = 'atr',
             pn_dir: str = None, encoding: str = 'latin-1') -> Annotation:
    """
    Read EDF+ annotations and convert to WFDB Annotation format.
    
    Parameters:
    - edf_file: str, EDF+ file containing annotations
    - ann_file: str, optional separate annotation file
    - ann_ext: str, annotation file extension
    - pn_dir: str, directory containing files
    - encoding: str, text encoding for annotation strings
    
    Returns:
    Annotation object containing the EDF+ annotations
    """

def wfdb_to_mat(record_name: str, pn_dir: str = None, sampfrom: int = 0,
                sampto: Union[int, str] = 'end', channels: List[int] = None,
                write_dir: str = '') -> None:
    """
    Convert WFDB record to MATLAB (.mat) format.
    
    Parameters:
    - record_name: str, WFDB record name
    - pn_dir: str, directory containing WFDB files
    - sampfrom: int, starting sample for conversion
    - sampto: int or 'end', ending sample for conversion
    - channels: list of int, specific channels to convert
    - write_dir: str, output directory for MAT file
    
    The output .mat file contains:
    - val: signal values matrix (samples x channels)
    - tm: time vector
    - Fs: sampling frequency
    - units: cell array of signal units
    - signame: cell array of signal names
    """

def csv_to_wfdb(file_name: str, fs: float, units: List[str], sig_name: List[str],
                record_name: str = None, start_time: str = None,
                use_checksum: bool = True, write_dir: str = '') -> None:
    """
    Convert CSV file to WFDB format.
    
    Parameters:
    - file_name: str, input CSV file name
    - fs: float, sampling frequency in Hz
    - units: list of str, signal units for each column
    - sig_name: list of str, signal names for each column
    - record_name: str, output record name (default: CSV filename stem)
    - start_time: str, record start time in HH:MM:SS format
    - use_checksum: bool, calculate and store file checksums
    - write_dir: str, output directory for WFDB files
    
    CSV format requirements:
    - First row: optional column headers
    - Subsequent rows: numeric signal values (one sample per row)
    """

def csv2ann(file_name: str, fs: float, record_name: str) -> None:
    """
    Convert CSV annotation file to WFDB annotation format.
    
    Parameters:
    - file_name: str, input CSV annotation file name
    - fs: float, sampling frequency for time conversion
    - record_name: str, associated record name for annotations
    
    CSV annotation format:
    - Column 1: time in seconds or sample numbers
    - Column 2: annotation symbols
    - Column 3: (optional) auxiliary notes
    """

def wfdb_to_wav(record_name: str, pn_dir: str = None, sampfrom: int = 0,
                sampto: Union[int, str] = 'end', channels: List[int] = None,
                write_dir: str = '', write_frequency: int = 8000) -> None:
    """
    Convert WFDB record to WAV audio format.
    
    Parameters:
    - record_name: str, WFDB record name
    - pn_dir: str, directory containing WFDB files
    - sampfrom: int, starting sample for conversion
    - sampto: int or 'end', ending sample for conversion
    - channels: list of int, specific channels to convert
    - write_dir: str, output directory for WAV files
    - write_frequency: int, output sampling frequency for WAV
    
    Note: Signals are automatically scaled and converted to appropriate
    bit depth for audio playback.
    """

def read_wav(record_name: str, pn_dir: str = None, delete_file: bool = True,
             record_only: bool = False) -> Union[Record, Tuple[Record, str]]:
    """
    Read WAV file and convert to WFDB Record format.
    
    Parameters:
    - record_name: str, WAV file name (without .wav extension)
    - pn_dir: str, directory containing WAV file
    - delete_file: bool, delete temporary files after conversion
    - record_only: bool, return only Record object
    
    Returns:
    Record object, or (Record, file_path) tuple if record_only=False
    """

def rdtff(file_name: str, cut_end: bool = False) -> Record:
    """
    Read TFF (Text File Format) files and convert to WFDB Record.
    
    Parameters:
    - file_name: str, TFF file name or path
    - cut_end: bool, remove trailing zero values
    
    Returns:
    Record object containing the TFF data
    
    TFF format consists of:
    - Header lines with metadata
    - Data rows with tab-separated numeric values
    """

import wfdb
import os

# Read EDF file
edf_record = wfdb.io.convert.read_edf('sleep_study.edf', verbose=True)
print(f"EDF record: {edf_record.sig_name}")
print(f"Duration: {edf_record.sig_len / edf_record.fs:.1f} seconds")
print(f"Channels: {edf_record.n_sig}")

# Write as WFDB format
edf_record.wrsamp('sleep_study_wfdb')

# Convert WFDB back to EDF
wfdb.io.convert.wfdb_to_edf('sleep_study_wfdb', 
                           edf_file='converted_back.edf',
                           file_type='EDF+')

import wfdb
import scipy.io

# Convert WFDB to MATLAB format
wfdb.io.convert.wfdb_to_mat('100', pn_dir='mitdb', 
                           sampfrom=0, sampto=3600)  # First 10 seconds

# Load in Python using scipy
mat_data = scipy.io.loadmat('100.mat')
print("MATLAB file contents:")
for key, value in mat_data.items():
    if not key.startswith('__'):
        print(f"  {key}: {type(value)} {getattr(value, 'shape', '')}")

# Access signal data
signals = mat_data['val']        # Signal values
time_vec = mat_data['tm'].flatten()  # Time vector
fs = mat_data['Fs'][0, 0]        # Sampling frequency
sig_names = [name[0] for name in mat_data['signame'].flatten()]

import wfdb
import pandas as pd
import numpy as np

# Create sample CSV data
data = {
    'ECG_Lead_I': np.random.randn(1000),
    'ECG_Lead_II': np.random.randn(1000),
    'Respiration': np.random.randn(1000) * 0.1
}
df = pd.DataFrame(data)
df.to_csv('sample_signals.csv', index=False)

# Convert CSV to WFDB
wfdb.io.convert.csv_to_wfdb(
    'sample_signals.csv',
    fs=250,  # 250 Hz sampling
    units=['mV', 'mV', 'V'],
    sig_name=['ECG I', 'ECG II', 'Resp'],
    record_name='csv_converted'
)

# Read back as WFDB record
converted_record = wfdb.rdrecord('csv_converted')
print(f"Converted record: {converted_record.sig_name}")

# Create annotation CSV
ann_data = pd.DataFrame({
    'time': [1.0, 2.5, 4.2, 6.8],
    'symbol': ['N', 'V', 'N', 'N'],
    'note': ['Normal', 'PVC', 'Normal', 'Normal']
})
ann_data.to_csv('annotations.csv', index=False)

# Convert annotation CSV to WFDB
wfdb.io.convert.csv2ann('annotations.csv', fs=250, record_name='csv_converted')

import wfdb
import soundfile as sf

# Read physiological signal
record = wfdb.rdrecord('100', pn_dir='mitdb', channels=[0])

# Convert to WAV for audio analysis
wfdb.io.convert.wfdb_to_wav('100', pn_dir='mitdb', 
                           channels=[0], 
                           write_frequency=8000)  # 8 kHz audio

# Read WAV file back
wav_record = wfdb.io.convert.read_wav('100')
print(f"WAV conversion: {wav_record.fs} Hz, {wav_record.sig_len} samples")

# Alternative: use soundfile for direct WAV handling
if os.path.exists('100.wav'):
    audio_data, sample_rate = sf.read('100.wav')
    print(f"Audio data shape: {audio_data.shape}")
    print(f"Sample rate: {sample_rate} Hz")

import wfdb
import os
from pathlib import Path

def convert_directory_to_matlab(input_dir, output_dir):
    """Convert all WFDB records in directory to MATLAB format."""
    
    # Find all .hea files (WFDB headers)
    header_files = list(Path(input_dir).glob('*.hea'))
    
    os.makedirs(output_dir, exist_ok=True)
    
    for hea_file in header_files:
        record_name = hea_file.stem
        print(f"Converting {record_name}...")
        
        try:
            # Convert to MATLAB
            wfdb.io.convert.wfdb_to_mat(
                record_name, 
                pn_dir=input_dir,
                write_dir=output_dir
            )
            print(f"  ✓ {record_name}.mat created")
            
        except Exception as e:
            print(f"  ✗ Error converting {record_name}: {e}")

# Example usage
# convert_directory_to_matlab('./wfdb_data', './matlab_data')

def batch_edf_to_wfdb(edf_directory, output_directory):
    """Convert multiple EDF files to WFDB format."""
    
    edf_files = list(Path(edf_directory).glob('*.edf'))
    os.makedirs(output_directory, exist_ok=True)
    
    for edf_file in edf_files:
        try:
            # Read EDF
            record = wfdb.io.convert.read_edf(str(edf_file))
            
            # Write as WFDB
            output_name = edf_file.stem
            record.wrsamp(output_name, write_dir=output_directory)
            
            print(f"Converted {edf_file.name} -> {output_name}")
            
        except Exception as e:
            print(f"Error converting {edf_file.name}: {e}")

# Example usage  
# batch_edf_to_wfdb('./edf_files', './wfdb_output')

import wfdb
import numpy as np

def validate_conversion_accuracy(original_record_name, pn_dir=None):
    """Test conversion accuracy by round-trip conversion."""
    
    # Read original WFDB record
    original = wfdb.rdrecord(original_record_name, pn_dir=pn_dir)
    
    # Convert to MATLAB and back
    wfdb.io.convert.wfdb_to_mat(original_record_name, pn_dir=pn_dir)
    
    # Note: Direct MAT to WFDB conversion not available
    # This demonstrates concept - actual implementation would require
    # manual MAT file reading and WFDB writing
    
    # Convert to CSV and back
    # First, save as CSV manually for this test
    import pandas as pd
    df = pd.DataFrame(original.p_signal, columns=original.sig_name)
    df.to_csv('temp_conversion.csv', index=False)
    
    # Convert CSV back to WFDB
    wfdb.io.convert.csv_to_wfdb(
        'temp_conversion.csv',
        fs=original.fs,
        units=original.units,
        sig_name=original.sig_name,
        record_name='temp_converted'
    )
    
    # Read converted record
    converted = wfdb.rdrecord('temp_converted')
    
    # Compare signals
    signal_diff = np.abs(original.p_signal - converted.p_signal)
    max_error = np.max(signal_diff)
    mean_error = np.mean(signal_diff)
    
    print(f"Conversion validation for {original_record_name}:")
    print(f"  Max error: {max_error:.6f}")
    print(f"  Mean error: {mean_error:.6f}")
    print(f"  Original shape: {original.p_signal.shape}")
    print(f"  Converted shape: {converted.p_signal.shape}")
    
    # Cleanup
    os.remove('temp_conversion.csv')
    for ext in ['.hea', '.dat']:
        if os.path.exists(f'temp_converted{ext}'):
            os.remove(f'temp_converted{ext}')
    
    return max_error < 1e-10  # Tolerance for floating point precision

# Example usage
# success = validate_conversion_accuracy('100', pn_dir='mitdb')

import wfdb
import json

def create_conversion_metadata(record_name, pn_dir=None):
    """Create metadata file for converted records."""
    
    # Read original record
    record = wfdb.rdrecord(record_name, pn_dir=pn_dir)
    
    # Create metadata dictionary
    metadata = {
        'original_format': 'WFDB',
        'record_name': record.record_name,
        'sampling_frequency': record.fs,
        'signal_length': record.sig_len,
        'n_signals': record.n_sig,
        'signal_names': record.sig_name,
        'units': record.units,
        'comments': record.comments,
        'conversion_timestamp': pd.Timestamp.now().isoformat(),
        'available_formats': []
    }
    
    # Convert to multiple formats
    formats_to_convert = ['matlab', 'csv', 'wav']
    
    for fmt in formats_to_convert:
        try:
            if fmt == 'matlab':
                wfdb.io.convert.wfdb_to_mat(record_name, pn_dir=pn_dir)
                metadata['available_formats'].append('matlab')
                
            elif fmt == 'csv':
                # Save signals as CSV
                df = pd.DataFrame(record.p_signal, columns=record.sig_name)
                df.to_csv(f'{record_name}_signals.csv', index=False)
                metadata['available_formats'].append('csv')
                
            elif fmt == 'wav':
                wfdb.io.convert.wfdb_to_wav(record_name, pn_dir=pn_dir)
                metadata['available_formats'].append('wav')
                
        except Exception as e:
            print(f"Could not convert to {fmt}: {e}")
    
    # Save metadata
    with open(f'{record_name}_metadata.json', 'w') as f:
        json.dump(metadata, f, indent=2)
    
    return metadata

# Example usage
# metadata = create_conversion_metadata('100', pn_dir='mitdb')
# print(f"Created conversions: {metadata['available_formats']}")