1 2# -*- coding: utf-8 -*- 3# Copyright (c) 2019 - 2020 Simon Kern 4# Copyright (c) 2015 - 2020 Holger Nahrstaedt 5# Copyright (c) 2011, 2015, Chris Lee-Messer 6# Copyright (c) 2016-2017 The pyedflib Developers 7# <https://github.com/holgern/pyedflib> 8# See LICENSE for license details. 9""" 10Created on Tue Jan 7 12:13:47 2020 11 12This file contains high-level functions to work with pyedflib. 13 14Includes 15 - Reading and writing EDFs 16 - Anonymizing EDFs 17 - Comparing EDFs 18 - Renaming Channels from EDF files 19 - Dropping Channels from EDF files 20 21@author: skjerns 22""" 23 24import os 25import numpy as np 26import warnings 27import pyedflib 28from copy import deepcopy 29from datetime import datetime 30# from . import EdfWriter 31# from . import EdfReader 32 33 34def _get_sample_frequency(signal_header): 35 # Temporary conditional assignment while we deprecate 'sample_rate' as a channel attribute 36 # in favor of 'sample_frequency', supporting the use of either to give 37 # users time to switch to the new interface. 38 return (signal_header['sample_rate'] 39 if signal_header.get('sample_frequency') is None 40 else signal_header['sample_frequency']) 41 42 43def tqdm(iteratable, *args, **kwargs): 44 """ 45 These is an optional dependecies that shows a progress bar for some 46 of the functions, e.g. loading. 47 48 install this dependency with `pip install tqdm` 49 50 if not installed this is just a pass through iterator. 51 """ 52 try: 53 from tqdm import tqdm as iterator 54 return iterator(iteratable, *args, **kwargs) 55 except: 56 return iteratable 57 58 59def _parse_date(string): 60 """ 61 A simple dateparser that detects common date formats 62 63 Parameters 64 ---------- 65 string : str 66 a date string in format as denoted below. 67 68 Returns 69 ------- 70 datetime.datetime 71 datetime object of a time. 72 73 """ 74 # some common formats. 75 formats = ['%Y-%m-%d', '%d-%m-%Y', '%d.%m.%Y', '%Y.%m.%d', '%d %b %Y', 76 '%Y/%m/%d', '%d/%m/%Y'] 77 for f in formats: 78 try: 79 return datetime.strptime(string, f) 80 except: 81 pass 82 try: 83 import dateparser 84 return dateparser.parse(string) 85 except: 86 print('dateparser is not installed. to convert strings to dates'\ 87 'install via `pip install dateparser`.') 88 raise ValueError('birthdate must be datetime object or of format'\ 89 ' `%d-%m-%Y`, eg. `24-01-2020`') 90 91def dig2phys(signal, dmin, dmax, pmin, pmax): 92 """ 93 converts digital edf values to physical values 94 95 Parameters 96 ---------- 97 signal : np.ndarray or int 98 A numpy array with int values (digital values) or an int. 99 dmin : int 100 digital minimum value of the edf file (eg -2048). 101 dmax : int 102 digital maximum value of the edf file (eg 2048). 103 pmin : float 104 physical maximum value of the edf file (eg -200.0). 105 pmax : float 106 physical maximum value of the edf file (eg 200.0). 107 108 Returns 109 ------- 110 physical : np.ndarray or float 111 converted physical values 112 113 """ 114 m = (pmax-pmin) / (dmax-dmin) 115 b = pmax / m - dmax 116 physical = m * (signal + b) 117 return physical 118 119 120def phys2dig(signal, dmin, dmax, pmin, pmax): 121 """ 122 converts physical values to digital values 123 124 Parameters 125 ---------- 126 signal : np.ndarray or int 127 A numpy array with int values (digital values) or an int. 128 dmin : int 129 digital minimum value of the edf file (eg -2048). 130 dmax : int 131 digital maximum value of the edf file (eg 2048). 132 pmin : float 133 physical maximum value of the edf file (eg -200.0). 134 pmax : float 135 physical maximum value of the edf file (eg 200.0). 136 137 Returns 138 ------- 139 digital : np.ndarray or int 140 converted digital values 141 142 """ 143 m = (pmax-pmin) / (dmax-dmin) 144 b = pmax / m - dmax 145 digital = signal/m - b 146 return digital 147 148 149 150def make_header(technician='', recording_additional='', patientname='', 151 patient_additional='', patientcode= '', equipment= '', 152 admincode= '', gender= '', startdate=None, birthdate= ''): 153 """ 154 A convenience function to create an EDF header (a dictionary) that 155 can be used by pyedflib to update the main header of the EDF 156 157 Parameters 158 ---------- 159 technician : str, optional 160 name of the technician. The default is ''. 161 recording_additional : str, optional 162 comments etc. The default is ''. 163 patientname : str, optional 164 the name of the patient. The default is ''. 165 patient_additional : TYPE, optional 166 more info about the patient. The default is ''. 167 patientcode : str, optional 168 alphanumeric code. The default is ''. 169 equipment : str, optional 170 which system was used. The default is ''. 171 admincode : str, optional 172 code of the admin. The default is ''. 173 gender : str, optional 174 gender of patient. The default is ''. 175 startdate : datetime.datetime, optional 176 startdate of recording. The default is None. 177 birthdate : str/datetime.datetime, optional 178 date of birth of the patient. The default is ''. 179 180 Returns 181 ------- 182 header : dict 183 a dictionary with the values given filled in. 184 185 """ 186 187 if not birthdate=='' and isinstance(birthdate, str): 188 birthdate = _parse_date(birthdate) 189 if startdate is None: 190 now = datetime.now() 191 startdate = datetime(now.year, now.month, now.day, 192 now.hour, now.minute, now.second) 193 del now 194 if isinstance(birthdate, datetime): 195 birthdate = birthdate.strftime('%d %b %Y').lower() 196 local = locals() 197 header = {} 198 for var in local: 199 if isinstance(local[var], datetime): 200 header[var] = local[var] 201 else: 202 header[var] = str(local[var]) 203 return header 204 205 206def make_signal_header(label, dimension='uV', sample_rate=256, sample_frequency=None, 207 physical_min=-200, physical_max=200, digital_min=-32768, 208 digital_max=32767, transducer='', prefiler=''): 209 """ 210 A convenience function that creates a signal header for a given signal. 211 This can be used to create a list of signal headers that is used by 212 pyedflib to create an edf. With this, different sampling frequencies 213 can be indicated. 214 215 Parameters 216 ---------- 217 label : str 218 the name of the channel. 219 dimension : str, optional 220 dimension, eg mV. The default is 'uV'. 221 sample_rate : int, optional 222 sampling frequency. The default is 256. Deprecated: use 'sample_frequency' instead. 223 sample_frequency : int, optional 224 sampling frequency. The default is 256. 225 physical_min : float, optional 226 minimum value in dimension. The default is -200. 227 physical_max : float, optional 228 maximum value in dimension. The default is 200. 229 digital_min : int, optional 230 digital minimum of the ADC. The default is -32768. 231 digital_max : int, optional 232 digital maximum of the ADC. The default is 32767. 233 transducer : str, optional 234 electrode type that was used. The default is ''. 235 prefiler : str, optional 236 filtering and sampling method. The default is ''. 237 238 Returns 239 ------- 240 signal_header : dict 241 a signal header that can be used to save a channel to an EDF. 242 243 """ 244 245 signal_header = {'label': label, 246 'dimension': dimension, 247 'sample_rate': sample_rate, 248 'sample_frequency': sample_frequency, 249 'physical_min': physical_min, 250 'physical_max': physical_max, 251 'digital_min': digital_min, 252 'digital_max': digital_max, 253 'transducer': transducer, 254 'prefilter': prefiler} 255 return signal_header 256 257 258def make_signal_headers(list_of_labels, dimension='uV', sample_rate=256, 259 sample_frequency=None, physical_min=-200.0, physical_max=200.0, 260 digital_min=-32768, digital_max=32767, 261 transducer='', prefiler=''): 262 """ 263 A function that creates signal headers for a given list of channel labels. 264 This can only be used if each channel has the same sampling frequency 265 266 Parameters 267 ---------- 268 list_of_labels : list of str 269 A list with labels for each channel. 270 dimension : str, optional 271 dimension, eg mV. The default is 'uV'. 272 sample_rate : int, optional 273 sampling frequency. The default is 256. Deprecated: use 'sample_frequency' instead. 274 sample_frequency : int, optional 275 sampling frequency. The default is 256. 276 physical_min : float, optional 277 minimum value in dimension. The default is -200. 278 physical_max : float, optional 279 maximum value in dimension. The default is 200. 280 digital_min : int, optional 281 digital minimum of the ADC. The default is -32768. 282 digital_max : int, optional 283 digital maximum of the ADC. The default is 32767. 284 transducer : str, optional 285 electrode type that was used. The default is ''. 286 prefiler : str, optional 287 filtering and sampling method. The default is ''. 288 289 Returns 290 ------- 291 signal_headers : list of dict 292 returns n signal headers as a list to save several signal headers. 293 294 """ 295 signal_headers = [] 296 for label in list_of_labels: 297 header = make_signal_header(label, dimension=dimension, sample_rate=sample_rate, 298 sample_frequency=sample_frequency, 299 physical_min=physical_min, physical_max=physical_max, 300 digital_min=digital_min, digital_max=digital_max, 301 transducer=transducer, prefiler=prefiler) 302 signal_headers.append(header) 303 return signal_headers 304 305 306def read_edf(edf_file, ch_nrs=None, ch_names=None, digital=False, verbose=False): 307 """ 308 Convenience function for reading EDF+/BDF data with pyedflib. 309 310 Will load the edf and return the signals, the headers of the signals 311 and the header of the EDF. If all signals have the same sample frequency 312 will return a numpy array, else a list with the individual signals 313 314 315 Parameters 316 ---------- 317 edf_file : str 318 link to an edf file. 319 ch_nrs : list of int, optional 320 The indices of the channels to read. The default is None. 321 ch_names : list of str, optional 322 The names of channels to read. The default is None. 323 digital : bool, optional 324 will return the signals as digital values (ADC). The default is False. 325 verbose : bool, optional 326 Print progress bar while loading or not. The default is False. 327 328 Returns 329 ------- 330 signals : np.ndarray or list 331 the signals of the chosen channels contained in the EDF. 332 signal_headers : list 333 one signal header for each channel in the EDF. 334 header : dict 335 the main header of the EDF file containing meta information. 336 337 """ 338 assert (ch_nrs is None) or (ch_names is None), \ 339 'names xor numbers should be supplied' 340 if ch_nrs is not None and not isinstance(ch_nrs, list): ch_nrs = [ch_nrs] 341 if ch_names is not None and \ 342 not isinstance(ch_names, list): ch_names = [ch_names] 343 344 with pyedflib.EdfReader(edf_file) as f: 345 # see which channels we want to load 346 available_chs = [ch.upper() for ch in f.getSignalLabels()] 347 n_chrs = f.signals_in_file 348 349 # find out which number corresponds to which channel 350 if ch_names is not None: 351 ch_nrs = [] 352 for ch in ch_names: 353 if not ch.upper() in available_chs: 354 warnings.warn('{} is not in source file (contains {})'\ 355 .format(ch, available_chs)) 356 print('will be ignored.') 357 else: 358 ch_nrs.append(available_chs.index(ch.upper())) 359 360 # if there ch_nrs is not given, load all channels 361 362 if ch_nrs is None: # no numbers means we load all 363 ch_nrs = range(n_chrs) 364 365 # convert negative numbers into positives 366 ch_nrs = [n_chrs+ch if ch<0 else ch for ch in ch_nrs] 367 368 # load headers, signal information and 369 header = f.getHeader() 370 signal_headers = [f.getSignalHeaders()[c] for c in ch_nrs] 371 372 # add annotations to header 373 annotations = f.readAnnotations() 374 annotations = [[s, d, a] for s,d,a in zip(*annotations)] 375 header['annotations'] = annotations 376 377 378 signals = [] 379 for i,c in enumerate(tqdm(ch_nrs, desc='Reading Channels', 380 disable=not verbose)): 381 signal = f.readSignal(c, digital=digital) 382 signals.append(signal) 383 384 # we can only return a np.array if all signals have the same samplefreq 385 sfreqs = [_get_sample_frequency(shead) for shead in signal_headers] 386 all_sfreq_same = sfreqs[1:]==sfreqs[:-1] 387 if all_sfreq_same: 388 dtype = np.int32 if digital else float 389 signals = np.array(signals, dtype=dtype) 390 391 assert len(signals)==len(signal_headers), 'Something went wrong, lengths'\ 392 ' of headers is not length of signals' 393 del f 394 return signals, signal_headers, header 395 396 397def write_edf(edf_file, signals, signal_headers, header=None, digital=False, 398 file_type=-1, block_size=1): 399 """ 400 Write signals to an edf_file. Header can be generated on the fly with 401 generic values. EDF+/BDF+ is selected based on the filename extension, 402 but can be overwritten by setting file_type to pyedflib.FILETYPE_XXX 403 404 Parameters 405 ---------- 406 edf_file : np.ndarray or list 407 where to save the EDF file 408 signals : list 409 The signals as a list of arrays or a ndarray. 410 411 signal_headers : list of dict 412 a list with one signal header(dict) for each signal. 413 See pyedflib.EdfWriter.setSignalHeader.. 414 header : dict 415 a main header (dict) for the EDF file, see 416 pyedflib.EdfWriter.setHeader for details. 417 If no header present, will create an empty header 418 digital : bool, optional 419 whether the signals are in digital format (ADC). The default is False. 420 file_type: int, optional 421 choose file_type for saving. 422 EDF = 0, EDF+ = 1, BDF = 2, BDF+ = 3, automatic from extension = -1 423 block_size : int 424 set the block size for writing. Should be divisor of signal length 425 in seconds. Higher values mean faster writing speed, but if it 426 is not a divisor of the signal duration, it will append zeros. 427 Can be any value between 1=><=60, -1 will auto-infer the fastest value. 428 429 Returns 430 ------- 431 bool 432 True if successful, False if failed. 433 """ 434 assert header is None or isinstance(header, dict), \ 435 'header must be dictioniary or None' 436 assert isinstance(signal_headers, list), \ 437 'signal headers must be list' 438 assert len(signal_headers)==len(signals), \ 439 'signals and signal_headers must be same length' 440 assert file_type in [-1, 0, 1, 2, 3], \ 441 'file_type must be in range -1, 3' 442 assert block_size<=60 and block_size>=-1 and block_size!=0, \ 443 'blocksize must be smaller or equal to 60' 444 445 # copy objects to prevent accidential changes to mutable objects 446 header = deepcopy(header) 447 signal_headers = deepcopy(signal_headers) 448 449 if file_type==-1: 450 ext = os.path.splitext(edf_file)[-1] 451 if ext.lower() == '.edf': 452 file_type = pyedflib.FILETYPE_EDFPLUS 453 elif ext.lower() == '.bdf': 454 file_type = pyedflib.FILETYPE_BDFPLUS 455 else: 456 raise ValueError('Unknown extension {}'.format(ext)) 457 458 n_channels = len(signals) 459 460 # if there is no header, we create one with dummy values 461 if header is None: 462 header = {} 463 default_header = make_header() 464 default_header.update(header) 465 header = default_header 466 467 # block_size sets the size of each writing block and should be a divisor 468 # of the length of the signal. If it is not, the remainder of the file 469 # will be filled with zeros. 470 signal_duration = len(signals[0]) // _get_sample_frequency(signal_headers[0]) 471 if block_size == -1: 472 block_size = max([d for d in range(1, 61) if signal_duration % d == 0]) 473 elif signal_duration % block_size != 0: 474 warnings.warn('Signal length is not dividable by block_size. '+ 475 'The file will have a zeros appended.') 476 477 # check dmin, dmax and pmin, pmax dont exceed signal min/max 478 for sig, shead in zip(signals, signal_headers): 479 dmin, dmax = shead['digital_min'], shead['digital_max'] 480 pmin, pmax = shead['physical_min'], shead['physical_max'] 481 label = shead['label'] 482 if digital: # exception as it will lead to clipping 483 assert dmin<=sig.min(), \ 484 'digital_min is {}, but signal_min is {}' \ 485 'for channel {}'.format(dmin, sig.min(), label) 486 assert dmax>=sig.max(), \ 487 'digital_min is {}, but signal_min is {}' \ 488 'for channel {}'.format(dmax, sig.max(), label) 489 assert pmin != pmax, \ 490 'physical_min {} should be different from physical_max {}'.format(pmin,pmax) 491 else: # only warning, as this will not lead to clipping 492 assert pmin<=sig.min(), \ 493 'phys_min is {}, but signal_min is {} ' \ 494 'for channel {}'.format(pmin, sig.min(), label) 495 assert pmax>=sig.max(), \ 496 'phys_max is {}, but signal_max is {} ' \ 497 'for channel {}'.format(pmax, sig.max(), label) 498 499 500 frequency_key = 'sample_rate' if shead.get('sample_frequency') is None else 'sample_frequency' 501 shead[frequency_key] *= block_size 502 503 # get annotations, in format [[timepoint, duration, description], [...]] 504 annotations = header.get('annotations', []) 505 506 with pyedflib.EdfWriter(edf_file, n_channels=n_channels, file_type=file_type) as f: 507 f.setDatarecordDuration(int(100000 * block_size)) 508 f.setSignalHeaders(signal_headers) 509 f.setHeader(header) 510 f.writeSamples(signals, digital=digital) 511 for annotation in annotations: 512 f.writeAnnotation(*annotation) 513 del f 514 515 return os.path.isfile(edf_file) 516 517 518def write_edf_quick(edf_file, signals, sfreq, digital=False): 519 """ 520 wrapper for write_pyedf without creating headers. 521 Use this if you don't care about headers or channel names and just 522 want to dump some signals with the same sampling freq. to an edf 523 524 Parameters 525 ---------- 526 edf_file : str 527 where to store the data/edf. 528 signals : np.ndarray 529 The signals you want to store as numpy array. 530 sfreq : int 531 the sampling frequency of the signals. 532 digital : bool, optional 533 if the data is present digitally (int) or as mV/uV.The default is False. 534 535 Returns 536 ------- 537 bool 538 True if successful, else False or raise Error. 539 540 """ 541 signals = np.atleast_2d(signals) 542 header = make_header(technician='pyedflib-quickwrite') 543 labels = ['CH_{}'.format(i) for i in range(len(signals))] 544 pmin, pmax = signals.min(), signals.max() 545 signal_headers = make_signal_headers(labels, sample_frequency=sfreq, 546 physical_min=pmin, physical_max=pmax) 547 return write_edf(edf_file, signals, signal_headers, header, digital=digital) 548 549 550def read_edf_header(edf_file, read_annotations=True): 551 """ 552 Reads the header and signal headers of an EDF file and it's annotations 553 554 Parameters 555 ---------- 556 edf_file : str 557 EDF/BDF file to read. 558 559 Returns 560 ------- 561 summary : dict 562 header of the edf file as dictionary. 563 564 """ 565 assert os.path.isfile(edf_file), 'file {} does not exist'.format(edf_file) 566 with pyedflib.EdfReader(edf_file) as f: 567 568 summary = f.getHeader() 569 summary['Duration'] = f.getFileDuration() 570 summary['SignalHeaders'] = f.getSignalHeaders() 571 summary['channels'] = f.getSignalLabels() 572 if read_annotations: 573 annotations = f.read_annotation() 574 annotations = [[float(t)/10000000, d if d else -1, x.decode()] for t,d,x in annotations] 575 summary['annotations'] = annotations 576 del f 577 return summary 578 579 580def compare_edf(edf_file1, edf_file2, verbose=False): 581 """ 582 Loads two edf files and checks whether the values contained in 583 them are the same. Does not check the header or annotations data. 584 585 Mainly to verify that other options (eg anonymization) produce the 586 same EDF file. 587 588 Parameters 589 ---------- 590 edf_file1 : str 591 edf file 1 to compare. 592 edf_file2 : str 593 edf file 2 to compare. 594 verbose : bool, optional 595 print progress or not. The default is False. 596 597 Returns 598 ------- 599 bool 600 True if signals are equal, else raises error. 601 """ 602 signals1, shead1, _ = read_edf(edf_file1, digital=True, verbose=verbose) 603 signals2, shead2, _ = read_edf(edf_file2, digital=True, verbose=verbose) 604 605 for i, sigs in enumerate(zip(signals1, signals2)): 606 s1, s2 = sigs 607 if np.array_equal(s1, s2): continue # early stopping 608 s1 = np.abs(s1) 609 s2 = np.abs(s2) 610 if np.array_equal(s1, s2): continue # early stopping 611 close = np.mean(np.isclose(s1, s2)) 612 assert close>0.99, 'Error, digital values of {}'\ 613 ' and {} for ch {}: {} are not the same: {:.3f}'.format( 614 edf_file1, edf_file2, shead1[i]['label'], 615 shead2[i]['label'], close) 616 617 dmin1, dmax1 = shead1[i]['digital_min'], shead1[i]['digital_max'] 618 pmin1, pmax1 = shead1[i]['physical_min'], shead1[i]['physical_max'] 619 dmin2, dmax2 = shead2[i]['digital_min'], shead2[i]['digital_max'] 620 pmin2, pmax2 = shead2[i]['physical_min'], shead2[i]['physical_max'] 621 622 for i, sigs in enumerate(zip(signals1, signals2)): 623 s1, s2 = sigs 624 625 # convert to physical values, no need to load all data again 626 s1 = dig2phys(s1, dmin1, dmax1, pmin1, pmax1) 627 s2 = dig2phys(s2, dmin2, dmax2, pmin2, pmax2) 628 629 # compare absolutes in case of inverted signals 630 if np.array_equal(s1, s2): continue # early stopping 631 s1 = np.abs(s1) 632 s2 = np.abs(s2) 633 if np.array_equal(s1, s2): continue # early stopping 634 min_dist = np.abs(dig2phys(1, dmin1, dmax1, pmin1, pmax1)) 635 close = np.mean(np.isclose(s1, s2, atol=min_dist)) 636 assert close>0.99, 'Error, physical values of {}'\ 637 ' and {} for ch {}: {} are not the same: {:.3f}'.format( 638 edf_file1, edf_file2, shead1[i]['label'], 639 shead2[i]['label'], close) 640 return True 641 642 643def drop_channels(edf_source, edf_target=None, to_keep=None, to_drop=None, 644 verbose=False): 645 """ 646 Remove channels from an edf file. Save the file. 647 For safety reasons, no source files can be overwritten. 648 649 Parameters 650 ---------- 651 edf_source : str 652 The source edf file from which to drop channels. 653 edf_target : str, optional 654 Where to save the file.If None, will be edf_source+'dropped.edf'. 655 The default is None. 656 to_keep : list, optional 657 A list of channel names or indices that will be kept. 658 Strings will always be interpreted as channel names. 659 'to_keep' will overwrite any droppings proposed by to_drop. 660 The default is None. 661 to_drop : list, optional 662 A list of channel names/indices that should be dropped. 663 Strings will be interpreted as channel names. The default is None. 664 verbose : bool, optional 665 print progress or not. The default is False. 666 667 Returns 668 ------- 669 edf_target : str 670 the target filename with the dropped channels. 671 672 """ 673 674 # convert to list if necessary 675 if isinstance(to_keep, (int, str)): to_keep = [to_keep] 676 if isinstance(to_drop, (int, str)): to_drop = [to_drop] 677 678 # check all parameters are good 679 assert to_keep is None or to_drop is None,'Supply only to_keep xor to_drop' 680 if to_keep is not None: 681 assert all([isinstance(ch, (str, int)) for ch in to_keep]),\ 682 'channels must be int or string' 683 if to_drop is not None: 684 assert all([isinstance(ch, (str, int)) for ch in to_drop]),\ 685 'channels must be int or string' 686 assert os.path.exists(edf_source), \ 687 'source file {} does not exist'.format(edf_source) 688 assert edf_source!=edf_target, 'For safet, target must not be source file.' 689 690 if edf_target is None: 691 edf_target = os.path.splitext(edf_source)[0] + '_dropped.edf' 692 if os.path.exists(edf_target): 693 warnings.warn('Target file will be overwritten') 694 695 ch_names = read_edf_header(edf_source)['channels'] 696 # convert to all lowercase for compatibility 697 ch_names = [ch.lower() for ch in ch_names] 698 ch_nrs = list(range(len(ch_names))) 699 700 if to_keep is not None: 701 for i,ch in enumerate(to_keep): 702 if isinstance(ch,str): 703 ch_idx = ch_names.index(ch.lower()) 704 to_keep[i] = ch_idx 705 load_channels = list(to_keep) # copy list compatible with py2.7 706 elif to_drop is not None: 707 for i,ch in enumerate(to_drop): 708 if isinstance(ch,str): 709 ch_idx = ch_names.index(ch.lower()) 710 to_drop[i] = ch_idx 711 to_drop = [len(ch_nrs)+ch if ch<0 else ch for ch in to_drop] 712 713 [ch_nrs.remove(ch) for ch in to_drop] 714 load_channels = list(ch_nrs) 715 else: 716 raise ValueError 717 718 signals, signal_headers, header = read_edf(edf_source, 719 ch_nrs=load_channels, 720 digital=True, verbose=verbose) 721 722 write_edf(edf_target, signals, signal_headers, header, digital=True) 723 return edf_target 724 725 726def anonymize_edf(edf_file, new_file=None, 727 to_remove=['patientname', 'birthdate'], 728 new_values=['xxx', ''], verify=False, verbose=False): 729 """Anonymize an EDF file by replacing values of header fields. 730 731 This function can be used to overwrite all header information that is 732 patient specific, for example birthdate and patientname. All header fields 733 can be overwritten this way (i.e., all header.keys() given 734 _, _, header = read_edf(edf_file, digital=True)). 735 736 Parameters 737 ---------- 738 edf_file : str 739 Filename of an EDF/BDF. 740 new_file : str | None 741 The filename of the anonymized file. If None, the input filename 742 appended with '_anonymized' is used. Defaults to None. 743 to_remove : list of str 744 List of attributes to overwrite in the `edf_file`. Defaults to 745 ['patientname', 'birthdate']. 746 new_values : list of str 747 List of values used for overwriting the attributes specified in 748 `to_remove`. Each item in `to_remove` must have a corresponding item 749 in `new_values`. Defaults to ['xxx', '']. 750 verify : bool 751 Compare `edf_file` and `new_file` for equality (i.e., double check that 752 values are same). Defaults to False 753 verbose : bool, optional 754 print progress or not. The default is False. 755 756 Returns 757 ------- 758 bool 759 True if successful, or if `verify` is False. Raises an error otherwise. 760 761 """ 762 if not len(to_remove) == len(new_values): 763 raise AssertionError('Each `to_remove` must have one `new_value`') 764 765 if new_file is None: 766 file, ext = os.path.splitext(edf_file) 767 new_file = file + '_anonymized' + ext 768 769 signals, signal_headers, header = read_edf(edf_file, digital=True, 770 verbose=verbose) 771 772 for new_val, attr in zip(new_values, to_remove): 773 header[attr] = new_val 774 775 write_edf(new_file, signals, signal_headers, header, digital=True) 776 if verify: 777 compare_edf(edf_file, new_file, verbose=verbose) 778 return True 779 780 781def rename_channels(edf_file, mapping, new_file=None, verbose=False): 782 """ 783 A convenience function to rename channels in an EDF file. 784 785 Parameters 786 ---------- 787 edf_file : str 788 an string pointing to an edf file. 789 mapping : dict 790 a dictionary with channel mappings as key:value. 791 eg: {'M1-O2':'A1-O2'} 792 new_file : str, optional 793 the new filename. If None will be edf_file + '_renamed' 794 The default is None. 795 verbose : bool, optional 796 print progress or not. The default is False. 797 798 Returns 799 ------- 800 bool 801 True if successful, False if failed. 802 803 """ 804 header = read_edf_header(edf_file) 805 channels = header['channels'] 806 if new_file is None: 807 file, ext = os.path.splitext(edf_file) 808 new_file = file + '_renamed' + ext 809 810 signal_headers = [] 811 signals = [] 812 for ch_nr in tqdm(range(len(channels)), disable=not verbose): 813 signal, signal_header, _ = read_edf(edf_file, digital=True, 814 ch_nrs=ch_nr, verbose=verbose) 815 ch = signal_header[0]['label'] 816 if ch in mapping : 817 if verbose: print('{} to {}'.format(ch, mapping[ch])) 818 ch = mapping[ch] 819 signal_header[0]['label']=ch 820 else: 821 if verbose: print('no mapping for {}, leave as it is'.format(ch)) 822 signal_headers.append(signal_header[0]) 823 signals.append(signal.squeeze()) 824 825 return write_edf(new_file, signals, signal_headers, header, digital=True) 826 827 828def change_polarity(edf_file, channels, new_file=None, verify=True, 829 verbose=False): 830 """ 831 Change polarity of certain channels 832 833 Parameters 834 ---------- 835 edf_file : str 836 from which file to change polarity. 837 channels : list of int 838 the indices of the channels. 839 new_file : str, optional 840 where to save the edf with inverted channels. The default is None. 841 verify : bool, optional 842 whether to verify the two edfs for similarity. The default is True. 843 verbose : str, optional 844 print progress or not. The default is True. 845 846 Returns 847 ------- 848 bool 849 True if success. 850 851 """ 852 853 if new_file is None: 854 new_file = os.path.splitext(edf_file)[0] + '.edf' 855 856 if isinstance(channels, str): channels=[channels] 857 channels = [c.lower() for c in channels] 858 859 signals, signal_headers, header = read_edf(edf_file, digital=True, 860 verbose=verbose) 861 for i,sig in enumerate(signals): 862 label = signal_headers[i]['label'].lower() 863 if label in channels: 864 if verbose: print('inverting {}'.format(label)) 865 signals[i] = -sig 866 write_edf(new_file, signals, signal_headers, header, 867 digital=True, correct=False, verbose=verbose) 868 if verify: compare_edf(edf_file, new_file) 869 return True 870