diff --git a/.git-blame-ignore-revs b/.git-blame-ignore-revs new file mode 100644 index 00000000..85404fc6 --- /dev/null +++ b/.git-blame-ignore-revs @@ -0,0 +1,2 @@ +# Apply ruff formatting and safe lint fixes +8790758451dfde249db2c31d4cc4ff13f5106fd2 diff --git a/.github/workflows/style.yml b/.github/workflows/style.yml index 6e714186..c9a81bf9 100644 --- a/.github/workflows/style.yml +++ b/.github/workflows/style.yml @@ -18,4 +18,6 @@ jobs: - name: Install Python run: uv python install 3.11 - name: Check style - run: uv run --extra dev black --check --diff . + run: | + uv run --extra dev ruff format --check --diff . + uv run --extra dev ruff check . diff --git a/DEVELOPING.md b/DEVELOPING.md index 3d28d3ec..d172490e 100644 --- a/DEVELOPING.md +++ b/DEVELOPING.md @@ -8,10 +8,17 @@ We welcome community contributions in the form of pull requests. When contributi - Unit tests are written for new features that are not covered by [existing tests](https://github.com/MIT-LCP/wfdb-python/tree/main/tests). - The code style is consistent with the project's formating standards. -Run the formatter with: +Format and lint the code with [Ruff](https://docs.astral.sh/ruff/): ```sh -black . +ruff format . +ruff check . +``` + +Commits that only reformat the code are listed in [.git-blame-ignore-revs](./.git-blame-ignore-revs). GitHub skips them in its blame view automatically. To do the same locally, run: + +```sh +git config blame.ignoreRevsFile .git-blame-ignore-revs ``` ## Package and Dependency Management diff --git a/docs/changes.rst b/docs/changes.rst index a21d7c50..c02048f7 100644 --- a/docs/changes.rst +++ b/docs/changes.rst @@ -17,6 +17,9 @@ Unreleased **Build with uv_build** The package is now built with ``uv_build`` instead of ``hatchling``. The version number is specified in ``pyproject.toml`` only, and ``wfdb.__version__`` is read from the installed package metadata (it is ``"unknown"`` if the package is run from a source tree without being installed). +**Use Ruff for formatting and linting** + The ``dev`` extra now includes ``ruff`` instead of ``black`` and ``pylint``, and the code is formatted with a line length of 88 (the default). + Version 4.3.1 (Feb 2026) ----------------------------- diff --git a/docs/conf.py b/docs/conf.py index d1907b87..cdda06d7 100644 --- a/docs/conf.py +++ b/docs/conf.py @@ -1,5 +1,4 @@ #!/usr/bin/env python3 -# -*- coding: utf-8 -*- # # WFDB documentation build configuration file, created by # sphinx-quickstart on Tue Jan 2 13:19:02 2018. diff --git a/pylintrc b/pylintrc deleted file mode 100644 index 3253109f..00000000 --- a/pylintrc +++ /dev/null @@ -1,564 +0,0 @@ -[MASTER] - -# Specify a configuration file. -#rcfile= - -# Python code to execute, usually for sys.path manipulation such as -# pygtk.require(). -#init-hook= - -# Files or directories to be skipped. They should be base names, not -# paths. -ignore=CVS - -# Add files or directories matching the regex patterns to the ignore-list. The -# regex matches against paths and can be in Posix or Windows format. -ignore-paths= - -# Files or directories matching the regex patterns are skipped. The regex -# matches against base names, not paths. -ignore-patterns=^\.# - -# Pickle collected data for later comparisons. -persistent=yes - -# List of plugins (as comma separated values of python modules names) to load, -# usually to register additional checkers. -load-plugins= - pylint.extensions.check_elif, - pylint.extensions.bad_builtin, - pylint.extensions.docparams, - pylint.extensions.for_any_all, - pylint.extensions.set_membership, - pylint.extensions.code_style, - pylint.extensions.overlapping_exceptions, - pylint.extensions.typing, - pylint.extensions.redefined_variable_type, - pylint.extensions.comparison_placement, - -# Use multiple processes to speed up Pylint. Specifying 0 will auto-detect the -# number of processors available to use. -jobs=1 - -# When enabled, pylint would attempt to guess common misconfiguration and emit -# user-friendly hints instead of false-positive error messages. -suggestion-mode=yes - -# Allow loading of arbitrary C extensions. Extensions are imported into the -# active Python interpreter and may run arbitrary code. -unsafe-load-any-extension=no - -# A comma-separated list of package or module names from where C extensions may -# be loaded. Extensions are loading into the active Python interpreter and may -# run arbitrary code -extension-pkg-allow-list= - -# Minimum supported python version -py-version = 3.7.2 - -# Control the amount of potential inferred values when inferring a single -# object. This can help the performance when dealing with large functions or -# complex, nested conditions. -limit-inference-results=100 - -# Specify a score threshold to be exceeded before program exits with error. -fail-under=10.0 - -# Return non-zero exit code if any of these messages/categories are detected, -# even if score is above --fail-under value. Syntax same as enable. Messages -# specified are enabled, while categories only check already-enabled messages. -fail-on= - - -[MESSAGES CONTROL] - -# Only show warnings with the listed confidence levels. Leave empty to show -# all. Valid levels: HIGH, INFERENCE, INFERENCE_FAILURE, UNDEFINED -# confidence= - -# Enable the message, report, category or checker with the given id(s). You can -# either give multiple identifier separated by comma (,) or put this option -# multiple time (only on the command line, not in the configuration file where -# it should appear only once). See also the "--disable" option for examples. -enable= - use-symbolic-message-instead, - useless-suppression, - fixme - -# Disable the message, report, category or checker with the given id(s). You -# can either give multiple identifiers separated by comma (,) or put this -# option multiple times (only on the command line, not in the configuration -# file where it should appear only once).You can also use "--disable=all" to -# disable everything first and then re-enable specific checks. For example, if -# you want to run only the similarities checker, you can use "--disable=all -# --enable=similarities". If you want to run only the classes checker, but have -# no Warning level messages displayed, use"--disable=all --enable=classes -# --disable=W" - -disable= - attribute-defined-outside-init, - duplicate-code, - invalid-name, - missing-docstring, - protected-access, - too-few-public-methods, - # handled by black - format, - - -[REPORTS] - -# Set the output format. Available formats are text, parseable, colorized, msvs -# (visual studio) and html. You can also give a reporter class, eg -# mypackage.mymodule.MyReporterClass. -output-format=text - -# Tells whether to display a full report or only the messages -reports=no - -# Python expression which should return a note less than 10 (10 is the highest -# note). You have access to the variables 'fatal', 'error', 'warning', 'refactor', 'convention' -# and 'info', which contain the number of messages in each category, as -# well as 'statement', which is the total number of statements analyzed. This -# score is used by the global evaluation report (RP0004). -evaluation=max(0, 0 if fatal else 10.0 - ((float(5 * error + warning + refactor + convention) / statement) * 10)) - -# Template used to display messages. This is a python new-style format string -# used to format the message information. See doc for all details -#msg-template= - -# Activate the evaluation score. -score=yes - - -[LOGGING] - -# Logging modules to check that the string format arguments are in logging -# function parameter format -logging-modules=logging - -# The type of string formatting that logging methods do. `old` means using % -# formatting, `new` is for `{}` formatting. -logging-format-style=old - - -[MISCELLANEOUS] - -# List of note tags to take in consideration, separated by a comma. -notes=FIXME,XXX,TODO - -# Regular expression of note tags to take in consideration. -#notes-rgx= - - -[SIMILARITIES] - -# Minimum lines number of a similarity. -min-similarity-lines=4 - -# Ignore comments when computing similarities. -ignore-comments=yes - -# Ignore docstrings when computing similarities. -ignore-docstrings=yes - -# Ignore imports when computing similarities. -ignore-imports=no - -# Signatures are removed from the similarity computation -ignore-signatures=no - - -[VARIABLES] - -# Tells whether we should check for unused import in __init__ files. -init-import=no - -# A regular expression matching the name of dummy variables (i.e. expectedly -# not used). -dummy-variables-rgx=_$|dummy - -# List of additional names supposed to be defined in builtins. Remember that -# you should avoid defining new builtins when possible. -additional-builtins= - -# List of strings which can identify a callback function by name. A callback -# name must start or end with one of those strings. -callbacks=cb_,_cb - -# Tells whether unused global variables should be treated as a violation. -allow-global-unused-variables=yes - -# List of names allowed to shadow builtins -allowed-redefined-builtins= - -# Argument names that match this expression will be ignored. Default to name -# with leading underscore. -ignored-argument-names=_.* - -# List of qualified module names which can have objects that can redefine -# builtins. -redefining-builtins-modules=six.moves,past.builtins,future.builtins,builtins,io - - -[FORMAT] - -# Maximum number of characters on a single line. -max-line-length=100 - -# Regexp for a line that is allowed to be longer than the limit. -ignore-long-lines=^\s*(# )??$ - -# Allow the body of an if to be on the same line as the test if there is no -# else. -single-line-if-stmt=no - -# Allow the body of a class to be on the same line as the declaration if body -# contains single statement. -single-line-class-stmt=no - -# Maximum number of lines in a module -max-module-lines=2000 - -# String used as indentation unit. This is usually " " (4 spaces) or "\t" (1 -# tab). -indent-string=' ' - -# Number of spaces of indent required inside a hanging or continued line. -indent-after-paren=4 - -# Expected format of line ending, e.g. empty (any line ending), LF or CRLF. -expected-line-ending-format= - - -[BASIC] - -# Good variable names which should always be accepted, separated by a comma -good-names=i,j,k,ex,Run,_ - -# Good variable names regexes, separated by a comma. If names match any regex, -# they will always be accepted -good-names-rgxs= - -# Bad variable names which should always be refused, separated by a comma -bad-names=foo,bar,baz,toto,tutu,tata - -# Bad variable names regexes, separated by a comma. If names match any regex, -# they will always be refused -bad-names-rgxs= - -# Colon-delimited sets of names that determine each other's naming style when -# the name regexes allow several styles. -name-group= - -# Include a hint for the correct naming format with invalid-name -include-naming-hint=no - -# Naming style matching correct function names. -function-naming-style=snake_case - -# Regular expression matching correct function names -function-rgx=[a-z_][a-z0-9_]{2,30}$ - -# Naming style matching correct variable names. -variable-naming-style=snake_case - -# Regular expression matching correct variable names -variable-rgx=[a-z_][a-z0-9_]{2,30}$ - -# Naming style matching correct constant names. -const-naming-style=UPPER_CASE - -# Regular expression matching correct constant names -const-rgx=(([A-Z_][A-Z0-9_]*)|(__.*__))$ - -# Naming style matching correct attribute names. -attr-naming-style=snake_case - -# Regular expression matching correct attribute names -attr-rgx=[a-z_][a-z0-9_]{2,}$ - -# Naming style matching correct argument names. -argument-naming-style=snake_case - -# Regular expression matching correct argument names -argument-rgx=[a-z_][a-z0-9_]{2,30}$ - -# Naming style matching correct class attribute names. -class-attribute-naming-style=any - -# Regular expression matching correct class attribute names -class-attribute-rgx=([A-Za-z_][A-Za-z0-9_]{2,30}|(__.*__))$ - -# Naming style matching correct class constant names. -class-const-naming-style=UPPER_CASE - -# Regular expression matching correct class constant names. Overrides class- -# const-naming-style. -#class-const-rgx= - -# Naming style matching correct inline iteration names. -inlinevar-naming-style=any - -# Regular expression matching correct inline iteration names -inlinevar-rgx=[A-Za-z_][A-Za-z0-9_]*$ - -# Naming style matching correct class names. -class-naming-style=PascalCase - -# Regular expression matching correct class names -class-rgx=[A-Z_][a-zA-Z0-9]+$ - - -# Naming style matching correct module names. -module-naming-style=snake_case - -# Regular expression matching correct module names -module-rgx=(([a-z_][a-z0-9_]*)|([A-Z][a-zA-Z0-9]+))$ - - -# Naming style matching correct method names. -method-naming-style=snake_case - -# Regular expression matching correct method names -method-rgx=[a-z_][a-z0-9_]{2,}$ - -# Regular expression which can overwrite the naming style set by typevar-naming-style. -#typevar-rgx= - -# Regular expression which should only match function or class names that do -# not require a docstring. Use ^(?!__init__$)_ to also check __init__. -no-docstring-rgx=__.*__ - -# Minimum line length for functions/classes that require docstrings, shorter -# ones are exempt. -docstring-min-length=-1 - -# List of decorators that define properties, such as abc.abstractproperty. -property-classes=abc.abstractproperty - - -[TYPECHECK] - -# Regex pattern to define which classes are considered mixins if ignore-mixin- -# members is set to 'yes' -mixin-class-rgx=.*MixIn - -# List of module names for which member attributes should not be checked -# (useful for modules/projects where namespaces are manipulated during runtime -# and thus existing member attributes cannot be deduced by static analysis). It -# supports qualified module names, as well as Unix pattern matching. -ignored-modules= - -# List of class names for which member attributes should not be checked (useful -# for classes with dynamically set attributes). This supports the use of -# qualified names. -ignored-classes=SQLObject, optparse.Values, thread._local, _thread._local - -# List of members which are set dynamically and missed by pylint inference -# system, and so shouldn't trigger E1101 when accessed. Python regular -# expressions are accepted. -generated-members=REQUEST,acl_users,aq_parent,argparse.Namespace - -# List of decorators that create context managers from functions, such as -# contextlib.contextmanager. -contextmanager-decorators=contextlib.contextmanager - -# Tells whether to warn about missing members when the owner of the attribute -# is inferred to be None. -ignore-none=yes - -# This flag controls whether pylint should warn about no-member and similar -# checks whenever an opaque object is returned when inferring. The inference -# can return multiple potential results while evaluating a Python object, but -# some branches might not be evaluated, which results in partial inference. In -# that case, it might be useful to still emit no-member and other checks for -# the rest of the inferred objects. -ignore-on-opaque-inference=yes - -# Show a hint with possible names when a member name was not found. The aspect -# of finding the hint is based on edit distance. -missing-member-hint=yes - -# The minimum edit distance a name should have in order to be considered a -# similar match for a missing member name. -missing-member-hint-distance=1 - -# The total number of similar names that should be taken in consideration when -# showing a hint for a missing member. -missing-member-max-choices=1 - -[SPELLING] - -# Spelling dictionary name. Available dictionaries: none. To make it working -# install python-enchant package. -spelling-dict= - -# List of comma separated words that should not be checked. -spelling-ignore-words= - -# List of comma separated words that should be considered directives if they -# appear and the beginning of a comment and should not be checked. -spelling-ignore-comment-directives=fmt: on,fmt: off,noqa:,noqa,nosec,isort:skip,mypy: - -# A path to a file that contains private dictionary; one word per line. -spelling-private-dict-file= - -# Tells whether to store unknown words to indicated private dictionary in -# --spelling-private-dict-file option instead of raising a message. -spelling-store-unknown-words=no - -# Limits count of emitted suggestions for spelling mistakes. -max-spelling-suggestions=4 - - -[DESIGN] - -# Maximum number of arguments for function / method -max-args=10 - -# Maximum number of locals for function / method body -max-locals=25 - -# Maximum number of return / yield for function / method body -max-returns=11 - -# Maximum number of branch for function / method body -max-branches=27 - -# Maximum number of statements in function / method body -max-statements=100 - -# Maximum number of parents for a class (see R0901). -max-parents=7 - -# List of qualified class names to ignore when counting class parents (see R0901). -ignored-parents= - -# Maximum number of attributes for a class (see R0902). -max-attributes=11 - -# Minimum number of public methods for a class (see R0903). -min-public-methods=2 - -# Maximum number of public methods for a class (see R0904). -max-public-methods=25 - -# Maximum number of boolean expressions in an if statement (see R0916). -max-bool-expr=5 - -# List of regular expressions of class ancestor names to -# ignore when counting public methods (see R0903). -exclude-too-few-public-methods= - -[CLASSES] - -# List of method names used to declare (i.e. assign) instance attributes. -defining-attr-methods=__init__,__new__,setUp,__post_init__ - -# List of valid names for the first argument in a class method. -valid-classmethod-first-arg=cls - -# List of valid names for the first argument in a metaclass class method. -valid-metaclass-classmethod-first-arg=mcs - -# List of member names, which should be excluded from the protected access -# warning. -exclude-protected=_asdict,_fields,_replace,_source,_make - -# Warn about protected attribute access inside special methods -check-protected-access-in-special-methods=no - -[IMPORTS] - -# List of modules that can be imported at any level, not just the top level -# one. -allow-any-import-level= - -# Allow wildcard imports from modules that define __all__. -allow-wildcard-with-all=no - -# Analyse import fallback blocks. This can be used to support both Python 2 and -# 3 compatible code, which means that the block might have code that exists -# only in one or another interpreter, leading to false positives when analysed. -analyse-fallback-blocks=no - -# Deprecated modules which should not be used, separated by a comma -deprecated-modules=regsub,TERMIOS,Bastion,rexec - -# Create a graph of every (i.e. internal and external) dependencies in the -# given file (report RP0402 must not be disabled) -import-graph= - -# Create a graph of external dependencies in the given file (report RP0402 must -# not be disabled) -ext-import-graph= - -# Create a graph of internal dependencies in the given file (report RP0402 must -# not be disabled) -int-import-graph= - -# Force import order to recognize a module as part of the standard -# compatibility libraries. -known-standard-library= - -# Force import order to recognize a module as part of a third party library. -known-third-party=enchant - -# Couples of modules and preferred modules, separated by a comma. -preferred-modules= - - -[EXCEPTIONS] - -# Exceptions that will emit a warning when being caught. Defaults to -# "Exception" -overgeneral-exceptions=Exception - - -[TYPING] - -# Set to ``no`` if the app / library does **NOT** need to support runtime -# introspection of type annotations. If you use type annotations -# **exclusively** for type checking of an application, you're probably fine. -# For libraries, evaluate if some users what to access the type hints at -# runtime first, e.g., through ``typing.get_type_hints``. Applies to Python -# versions 3.7 - 3.9 -runtime-typing = no - - -[DEPRECATED_BUILTINS] - -# List of builtins function names that should not be used, separated by a comma -bad-functions=map,input - - -[REFACTORING] - -# Maximum number of nested blocks for function / method body -max-nested-blocks=5 - -# Complete name of functions that never returns. When checking for -# inconsistent-return-statements if a never returning function is called then -# it will be considered as an explicit return statement and no message will be -# printed. -never-returning-functions=sys.exit,argparse.parse_error - - -[STRING] - -# This flag controls whether inconsistent-quotes generates a warning when the -# character used as a quote delimiter is used inconsistently within a module. -check-quote-consistency=no - -# This flag controls whether the implicit-str-concat should generate a warning -# on implicit string concatenation in sequences defined over several lines. -check-str-concat-over-line-jumps=no - - -[CODE_STYLE] - -# Max line length for which to sill emit suggestions. Used to prevent optional -# suggestions which would get split by a code formatter (e.g., black). Will -# default to the setting for ``max-line-length``. -#max-line-length-suggestions= diff --git a/pyproject.toml b/pyproject.toml index 0c92b787..864bf5a1 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -26,8 +26,7 @@ dependencies = [ dev = [ "pytest >= 8.0.0", "pytest-xdist >= 3.5.0", - "pylint >= 2.13.7", - "black >= 22.3.0", + "ruff >= 0.13.0", "sphinx >= 7.0.0", ] @@ -36,9 +35,31 @@ homepage = "https://github.com/MIT-LCP/wfdb-python/" repository = "https://github.com/MIT-LCP/wfdb-python/" documentation = "https://wfdb.readthedocs.io/" -[tool.black] -line-length = 80 -target-version = ["py311"] +[tool.ruff] +target-version = "py311" +extend-exclude = ["*.ipynb"] + +[tool.ruff.lint] +select = ["E", "F", "W", "I", "UP", "B"] +# Existing violations (other than E501), to be cleaned up separately +ignore = [ + "E501", # handled by the formatter + "E711", + "E721", + "E722", + "E741", + "F841", + "UP031", + "B006", + "B007", + "B020", + "B904", + "B905", +] + +[tool.ruff.lint.per-file-ignores] +# Re-exports +"__init__.py" = ["F401"] [tool.uv.build-backend] module-root = "" diff --git a/tests/io/test_convert.py b/tests/io/test_convert.py index cf97f700..1453a5e2 100644 --- a/tests/io/test_convert.py +++ b/tests/io/test_convert.py @@ -4,9 +4,9 @@ import numpy as np -from wfdb.io.record import rdrecord -from wfdb.io.convert.edf import read_edf from wfdb.io.convert.csv import csv_to_wfdb +from wfdb.io.convert.edf import read_edf +from wfdb.io.record import rdrecord class TestEdfToWfdb: @@ -170,9 +170,7 @@ def test_write_dir(self): self.assertEqual(record_write.sig_name, ["MLII", "V5"]) self.assertEqual(record_write.sig_name, self.record_100_dat.sig_name) self.assertEqual(record_write.p_signal.size, 1300000) - self.assertEqual( - record_write.p_signal.size, self.record_100_dat.p_signal.size - ) + self.assertEqual(record_write.p_signal.size, self.record_100_dat.p_signal.size) if __name__ == "__main__": diff --git a/tests/test_annotation.py b/tests/test_annotation.py index db3e71d0..e70045f8 100644 --- a/tests/test_annotation.py +++ b/tests/test_annotation.py @@ -33,7 +33,7 @@ def test_1(self): # no null to detect in the output text file of rdann. # Target data from WFDB software package - with open("tests/target-output/ann-1", "r") as f: + with open("tests/target-output/ann-1") as f: lines = tuple(f) nannot = len(lines) @@ -109,7 +109,7 @@ def test_2(self): annotation = wfdb.rdann("sample-data/12726", "anI") # Target data from WFDB software package - with open("tests/target-output/ann-2", "r") as f: + with open("tests/target-output/ann-2") as f: lines = tuple(f) nannot = len(lines) @@ -183,7 +183,7 @@ def test_3(self): annotation = wfdb.rdann("sample-data/1003", "atr") # Target data from WFDB software package - with open("tests/target-output/ann-3", "r") as f: + with open("tests/target-output/ann-3") as f: lines = tuple(f) nannot = len(lines) diff --git a/tests/test_datasource.py b/tests/test_datasource.py index 98cd22bc..107dd3b5 100644 --- a/tests/test_datasource.py +++ b/tests/test_datasource.py @@ -1,4 +1,5 @@ import platform + import pytest from wfdb.io import ( @@ -8,7 +9,6 @@ remove_data_source, reset_data_sources, ) - from wfdb.io.datasource import _data_sources LOCAL_PATH = ( diff --git a/tests/test_multi_record.py b/tests/test_multi_record.py index 683c2c44..6c217820 100644 --- a/tests/test_multi_record.py +++ b/tests/test_multi_record.py @@ -93,11 +93,9 @@ def test_contained_combined_ranges_variable_layout(self): (14428365, 14450865), (14458365, 14495865), ] - assert ( - record.contained_combined_ranges(["II", "V", "MCL1", "ABP"]) == [] - ) + assert record.contained_combined_ranges(["II", "V", "MCL1", "ABP"]) == [] - def test_contained_combined_ranges_variable_layout(self): + def test_contained_combined_ranges_all_pairs(self): record = wfdb.rdheader( "sample-data/multi-segment/041s/041s", rd_segments=True, @@ -108,6 +106,4 @@ def test_contained_combined_ranges_variable_layout(self): if sig_1 == sig_2: continue - assert record.contained_combined_ranges([sig_1, sig_2]) == [ - (0, 2000) - ] + assert record.contained_combined_ranges([sig_1, sig_2]) == [(0, 2000)] diff --git a/tests/test_plot.py b/tests/test_plot.py index 289ebbe5..c11ce518 100644 --- a/tests/test_plot.py +++ b/tests/test_plot.py @@ -115,9 +115,7 @@ def test_digital_smooth(self): self.assertEqual(len(fig.axes), record.n_sig) for ch in range(record.n_sig): - self.assertAxesMatchSignal( - fig.axes[ch], record.d_signal[:, ch], record.fs - ) + self.assertAxesMatchSignal(fig.axes[ch], record.d_signal[:, ch], record.fs) def test_physical_multifrequency(self): """ @@ -177,9 +175,7 @@ def test_get_plot_dims(self): record = wfdb.rdrecord( "sample-data/100", physical=True, sampfrom=sampfrom, sampto=sampto ) - ann = wfdb.rdann( - "sample-data/100", "atr", sampfrom=sampfrom, sampto=sampto - ) + ann = wfdb.rdann("sample-data/100", "atr", sampfrom=sampfrom, sampto=sampto) sig_len, n_sig, n_annot, n_subplots = plot._get_plot_dims( signal=record.p_signal, ann_samp=[ann.sample] ) diff --git a/tests/test_processing.py b/tests/test_processing.py index e2b27627..69f9e61b 100644 --- a/tests/test_processing.py +++ b/tests/test_processing.py @@ -18,9 +18,7 @@ def test_resample_single(self): fs = fields["fs"] fs_target = 50 - new_sig, new_ann = processing.resample_singlechan( - sig[:, 0], ann, fs, fs_target - ) + new_sig, new_ann = processing.resample_singlechan(sig[:, 0], ann, fs, fs_target) expected_length = int(sig.shape[0] * fs_target / fs) @@ -33,9 +31,7 @@ def test_resample_multi(self): fs = fields["fs"] fs_target = 50 - new_sig, new_ann = processing.resample_multichan( - sig, ann, fs, fs_target - ) + new_sig, new_ann = processing.resample_multichan(sig, ann, fs, fs_target) expected_length = int(sig.shape[0] * fs_target / fs) diff --git a/tests/test_record.py b/tests/test_record.py index 907a929b..cd2704db 100644 --- a/tests/test_record.py +++ b/tests/test_record.py @@ -1,6 +1,5 @@ import datetime import os -import shutil import tempfile import unittest @@ -48,9 +47,7 @@ def test_1a(self): Target file created with: rdsamp -r sample-data/test01_00s | cut -f 2- > record-1a """ - record = wfdb.rdrecord( - "sample-data/test01_00s", physical=False, return_res=16 - ) + record = wfdb.rdrecord("sample-data/test01_00s", physical=False, return_res=16) sig = record.d_signal sig_target = np.genfromtxt("tests/target-output/record-1a") @@ -188,9 +185,7 @@ def test_1e(self): rdsamp -r sample-data/n8_evoked_raw_95_F1_R9 | cut -f 2- | gzip -9 -n > record-1e.gz """ - record = wfdb.rdrecord( - "sample-data/n8_evoked_raw_95_F1_R9", physical=False - ) + record = wfdb.rdrecord("sample-data/n8_evoked_raw_95_F1_R9", physical=False) sig = record.d_signal sig_target = np.genfromtxt("tests/target-output/record-1e.gz") sig_target[sig_target == -32768] = -(2**23) @@ -201,9 +196,7 @@ def test_1e(self): ) # Test file writing - record_2 = wfdb.rdrecord( - "sample-data/n8_evoked_raw_95_F1_R9", physical=False - ) + record_2 = wfdb.rdrecord("sample-data/n8_evoked_raw_95_F1_R9", physical=False) record_2.wrsamp(write_dir=self.temp_path) record_write = wfdb.rdrecord( os.path.join(self.temp_path, "n8_evoked_raw_95_F1_R9"), @@ -381,10 +374,7 @@ def test_unique_samps_per_frame_e_p_signal(self): isinstance(getattr(record, k), np.ndarray) or ( isinstance(getattr(record, k), list) - and all( - isinstance(item, np.ndarray) - for item in getattr(record, k) - ) + and all(isinstance(item, np.ndarray) for item in getattr(record, k)) ) ) } @@ -466,10 +456,7 @@ def test_unique_samps_per_frame_e_d_signal(self): isinstance(getattr(record, k), np.ndarray) or ( isinstance(getattr(record, k), list) - and all( - isinstance(item, np.ndarray) - for item in getattr(record, k) - ) + and all(isinstance(item, np.ndarray) for item in getattr(record, k)) ) ) } @@ -643,9 +630,7 @@ def test_2d(self): rdsamp -r sample-data/3000003_0003 -f 0 -t 8.21 | cut -f 2- | wrsamp -o 310derive -O 310 rdsamp -r 310derive -f 0.007 | cut -f 2- > record-2d """ - record = wfdb.rdrecord( - "sample-data/310derive", sampfrom=2, physical=False - ) + record = wfdb.rdrecord("sample-data/310derive", sampfrom=2, physical=False) sig = record.d_signal sig_target = np.genfromtxt("tests/target-output/record-2d") assert np.array_equal(sig, sig_target) @@ -658,9 +643,7 @@ def test_2e(self): rdsamp -r sample-data/3000003_0003 -f 0 -t 8.21 -s 1 | cut -f 2- | wrsamp -o 311derive -O 311 rdsamp -r 311derive -f 0.005 -t 3.91 -P | cut -f 2- > record-2e """ - sig, fields = wfdb.rdsamp( - "sample-data/311derive", sampfrom=1, sampto=978 - ) + sig, fields = wfdb.rdsamp("sample-data/311derive", sampfrom=1, sampto=978) sig = np.round(sig, decimals=8) sig_target = np.genfromtxt("tests/target-output/record-2e") sig_target = sig_target.reshape([977, 1]) @@ -679,9 +662,7 @@ def test_3a(self): sig_target = np.genfromtxt("tests/target-output/record-3a") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "s0010_re", physical=False, pn_dir="ptbdb/patient001" - ) + record_pn = wfdb.rdrecord("s0010_re", physical=False, pn_dir="ptbdb/patient001") # Test file writing record.wrsamp(write_dir=self.temp_path) @@ -731,9 +712,7 @@ def test_4a(self): Target file created with: rdsamp -r sample-data/test01_00s_skewframe | cut -f 2- > record-4a """ - record = wfdb.rdrecord( - "sample-data/test01_00s_skewframe", physical=False - ) + record = wfdb.rdrecord("sample-data/test01_00s_skewframe", physical=False) sig = record.d_signal # The WFDB library rdsamp does not return the final N samples for all # channels due to the skew. The WFDB python rdsamp does return the final @@ -777,9 +756,7 @@ def test_4b(self): sig_target = np.genfromtxt("tests/target-output/record-4b") # Compare data streaming from Physionet - record_pn = wfdb.rdrecord( - "03700181", physical=False, pn_dir="mimicdb/037" - ) + record_pn = wfdb.rdrecord("03700181", physical=False, pn_dir="mimicdb/037") # Test file writing. Multiple samples per frame and skew. # Have to read all the samples in the record, ignoring skew @@ -851,9 +828,7 @@ def test_4d(self): Target file created with: rdsamp -r sample-data/test01_00s_skewframe -P -H | cut -f 2- > record-4d """ - record = wfdb.rdrecord( - "sample-data/test01_00s_skewframe", smooth_frames=False - ) + record = wfdb.rdrecord("sample-data/test01_00s_skewframe", smooth_frames=False) # Upsample the channels with lower samples/frame expandsig = np.zeros((7994, 3)) @@ -957,9 +932,7 @@ def test_multi_fixed_b(self): Target file created with: rdsamp -r sample-data/multi-segment/fixed1/v102s -t s75000 -P | cut -f 2- > record-multi-fixed-b """ - record = wfdb.rdrecord( - "sample-data/multi-segment/fixed1/v102s", sampto=75000 - ) + record = wfdb.rdrecord("sample-data/multi-segment/fixed1/v102s", sampto=75000) sig_round = np.round(record.p_signal, decimals=8) sig_target = np.genfromtxt("tests/target-output/record-multi-fixed-b") @@ -1014,9 +987,7 @@ def test_multi_fixed_d(self): for i, s in enumerate(record.e_d_signal): sig[:, i] = np.repeat(s, len(sig[:, i]) // len(s)) - sig_target = np.genfromtxt( - "tests/target-output/record-multi-fixed-d.gz" - ) + sig_target = np.genfromtxt("tests/target-output/record-multi-fixed-d.gz") record_named = wfdb.rdrecord( "sample-data/multi-segment/041s/041s", @@ -1044,9 +1015,7 @@ def test_multi_variable_a(self): sampto=14428375, ) sig_round = np.round(record.p_signal, decimals=8) - sig_target = np.genfromtxt( - "tests/target-output/record-multi-variable-a" - ) + sig_target = np.genfromtxt("tests/target-output/record-multi-variable-a") np.testing.assert_equal(sig_round, sig_target) @@ -1064,9 +1033,7 @@ def test_multi_variable_b(self): sampto=14428375, ) sig_round = np.round(record.p_signal, decimals=8) - sig_target = np.genfromtxt( - "tests/target-output/record-multi-variable-b" - ) + sig_target = np.genfromtxt("tests/target-output/record-multi-variable-b") np.testing.assert_equal(sig_round, sig_target) @@ -1150,9 +1117,7 @@ def test_multi_variable_d(self): channels=[0, 1], physical=False, ) - sig_target = np.genfromtxt( - "tests/target-output/record-multi-variable-d" - ) + sig_target = np.genfromtxt("tests/target-output/record-multi-variable-d") # Option of selecting channels by name record_named = wfdb.rdrecord( diff --git a/tests/test_url.py b/tests/test_url.py index b2ca85df..e96b6fa4 100644 --- a/tests/test_url.py +++ b/tests/test_url.py @@ -224,8 +224,8 @@ def test_whole_file(self): N/A """ - import soundfile import numpy as np + import soundfile data_file_path = "sample-data/flacformats.d2" expected_format = "FLAC" diff --git a/wfdb/__init__.py b/wfdb/__init__.py index 5d4d0b8b..7a16d14e 100644 --- a/wfdb/__init__.py +++ b/wfdb/__init__.py @@ -1,42 +1,37 @@ -from wfdb.io.record import ( - Record, - MultiRecord, - rdheader, - rdrecord, - rdsamp, - wrsamp, - dl_database, - sampfreq, - signame, - wfdbdesc, - wfdbtime, -) from wfdb.io.annotation import ( Annotation, + mrgann, rdann, - wrann, - show_ann_labels, show_ann_classes, - mrgann, -) -from wfdb.io.download import ( - dl_files, - get_dbs, - get_record_list, - set_db_index_url, + show_ann_labels, + wrann, ) - - from wfdb.io.datasource import ( DataSource, DataSourceType, - show_data_sources, add_data_source, remove_data_source, reset_data_sources, + show_data_sources, +) +from wfdb.io.download import ( + dl_files, + get_dbs, + get_record_list, + set_db_index_url, +) +from wfdb.io.record import ( + MultiRecord, + Record, + dl_database, + rdheader, + rdrecord, + rdsamp, + sampfreq, + signame, + wfdbdesc, + wfdbtime, + wrsamp, ) - - -from wfdb.plot.plot import plot_items, plot_wfdb, plot_all_records - +from wfdb.plot.plot import plot_all_records, plot_items, plot_wfdb from wfdb.version import __version__ diff --git a/wfdb/io/__init__.py b/wfdb/io/__init__.py index fb37f566..30969d43 100644 --- a/wfdb/io/__init__.py +++ b/wfdb/io/__init__.py @@ -1,38 +1,37 @@ -from wfdb.io.record import ( - Record, - MultiRecord, - rdheader, - rdrecord, - rdsamp, - wrsamp, - dl_database, - sampfreq, - signame, - wfdbdesc, - wfdbtime, - SIGNAL_CLASSES, -) from wfdb.io._signal import est_res, wr_dat_file from wfdb.io.annotation import ( Annotation, + mrgann, rdann, - wrann, - show_ann_labels, show_ann_classes, - mrgann, -) -from wfdb.io.download import ( - dl_files, - get_dbs, - get_record_list, - set_db_index_url, + show_ann_labels, + wrann, ) - from wfdb.io.datasource import ( DataSource, DataSourceType, - show_data_sources, add_data_source, remove_data_source, reset_data_sources, + show_data_sources, +) +from wfdb.io.download import ( + dl_files, + get_dbs, + get_record_list, + set_db_index_url, +) +from wfdb.io.record import ( + SIGNAL_CLASSES, + MultiRecord, + Record, + dl_database, + rdheader, + rdrecord, + rdsamp, + sampfreq, + signame, + wfdbdesc, + wfdbtime, + wrsamp, ) diff --git a/wfdb/io/_header.py b/wfdb/io/_header.py index d8315eef..dabc6678 100644 --- a/wfdb/io/_header.py +++ b/wfdb/io/_header.py @@ -1,11 +1,11 @@ import datetime -from typing import Any, Dict, List, Optional, Sequence, Tuple +from collections.abc import Sequence +from typing import Any import numpy as np import pandas as pd -from wfdb.io import _signal -from wfdb.io import util +from wfdb.io import _signal, util from wfdb.io.header import HeaderSyntaxError, rx_record, rx_segment, rx_signal """ @@ -129,7 +129,7 @@ FIELD_SPECS = pd.concat((RECORD_SPECS, SIGNAL_SPECS, SEGMENT_SPECS)) -class BaseHeaderMixin(object): +class BaseHeaderMixin: """ Mixin class with multi-segment header methods. Inherited by Record and MultiRecord classes. @@ -186,7 +186,7 @@ def get_write_subset(self, spec_type): write_fields.append(req_field) req_field = record_specs.loc[req_field, "dependency"] # Add comments if any - if getattr(self, "comments") is not None: + if self.comments is not None: write_fields.append("comments") # signal spec field. Need to return a potentially different list for each channel. @@ -211,9 +211,7 @@ def get_write_subset(self, spec_type): # Add the field and its recursive dependencies while req_field is not None: write_fields_ch.append(req_field) - req_field = signal_specs.loc[ - req_field, "dependency" - ] + req_field = signal_specs.loc[req_field, "dependency"] write_fields.append(write_fields_ch) @@ -454,10 +452,7 @@ def set_default(self, field): # Return if no default to set, or if the field is already # present. - if ( - SIGNAL_SPECS.loc[field, "write_default"] is None - or item is not None - ): + if SIGNAL_SPECS.loc[field, "write_default"] is None or item is not None: return # Set more specific defaults if possible @@ -518,10 +513,7 @@ def check_field_cohesion(self, rec_write_fields, sig_write_fields): if self.file_name[ch] not in datoffsets: datoffsets[self.file_name[ch]] = self.byte_offset[ch] else: - if ( - datoffsets[self.file_name[ch]] - != self.byte_offset[ch] - ): + if datoffsets[self.file_name[ch]] != self.byte_offset[ch]: raise ValueError( "Each file_name (dat file) specified must have the same byte offset" ) @@ -553,9 +545,7 @@ def wr_header_file(self, rec_write_fields, sig_write_fields, write_dir): # If the field is being used, add it with its delimiter if field in rec_write_fields: string_field = self.preprocess_header(field) - record_line += ( - RECORD_SPECS.loc[field, "delimiter"] + string_field - ) + record_line += RECORD_SPECS.loc[field, "delimiter"] + string_field # The 'base_counter' field needs to be closed with ')' if field == "base_counter": record_line += ")" @@ -571,13 +561,10 @@ def wr_header_file(self, rec_write_fields, sig_write_fields, write_dir): # If the field is being used, add each of its # elements with the delimiter to the appropriate # line - if ( - field in sig_write_fields - and ch in sig_write_fields[field] - ): - signal_lines[ch] += SIGNAL_SPECS.loc[ - field, "delimiter" - ] + str(getattr(self, field)[ch]) + if field in sig_write_fields and ch in sig_write_fields[field]: + signal_lines[ch] += SIGNAL_SPECS.loc[field, "delimiter"] + str( + getattr(self, field)[ch] + ) # The 'baseline' field needs to be closed with ')' if field == "baseline": signal_lines[ch] += ")" @@ -605,7 +592,7 @@ class MultiHeaderMixin(BaseHeaderMixin): n_seg: int seg_len: Sequence[int] - segments: Optional[Sequence] + segments: Sequence | None def set_defaults(self): """ @@ -763,9 +750,7 @@ def wr_header_file(self, write_fields, write_dir): # If the field is being used, add it with its delimiter if field in write_fields: string_field = self.preprocess_header(field) - record_line += ( - RECORD_SPECS.loc[field, "delimiter"] + string_field - ) + record_line += RECORD_SPECS.loc[field, "delimiter"] + string_field header_lines = [record_line] @@ -775,9 +760,9 @@ def wr_header_file(self, write_fields, write_dir): # to the appropriate line for field in SEGMENT_SPECS.index: for seg_num in range(self.n_seg): - segment_lines[seg_num] += SEGMENT_SPECS.loc[ - field, "delimiter" - ] + str(getattr(self, field)[seg_num]) + segment_lines[seg_num] += SEGMENT_SPECS.loc[field, "delimiter"] + str( + getattr(self, field)[seg_num] + ) header_lines = header_lines + segment_lines @@ -824,10 +809,7 @@ def get_sig_segments(self, sig_name=None): elif isinstance(sig_name, str): sig_segs = [] for i in range(self.n_seg): - if ( - self.seg_name[i] != "~" - and sig_name in self.segments[i].sig_name - ): + if self.seg_name[i] != "~" and sig_name in self.segments[i].sig_name: sig_segs.append(i) return sig_segs else: @@ -862,7 +844,7 @@ def get_sig_name(self): return sig_name - def contained_ranges(self, sig_name: str) -> List[Tuple[int, int]]: + def contained_ranges(self, sig_name: str) -> list[tuple[int, int]]: """ Given a signal name, return the sample ranges that contain signal values, relative to the start of the full record. Does not account for NaNs/missing @@ -931,7 +913,7 @@ def contained_ranges(self, sig_name: str) -> List[Tuple[int, int]]: def contained_combined_ranges( self, sig_names: Sequence[str], - ) -> List[Tuple[int, int]]: + ) -> list[tuple[int, int]]: """ Given a collection of signal name, return the sample ranges that contain all of the specified signals, relative to the start of the @@ -1020,7 +1002,7 @@ def _parse_record_line(record_line: str) -> dict: """ # Dictionary for record fields - record_fields: Dict[str, Any] = {} + record_fields: dict[str, Any] = {} # Read string fields from record line match = rx_record.match(record_line) @@ -1057,9 +1039,7 @@ def _parse_record_line(record_line: str) -> dict: fs = int(fs) record_fields["fs"] = fs elif field == "base_time": - record_fields["base_time"] = wfdb_strptime( - record_fields["base_time"] - ) + record_fields["base_time"] = wfdb_strptime(record_fields["base_time"]) elif field == "base_date": record_fields["base_date"] = datetime.datetime.strptime( record_fields["base_date"], "%d/%m/%Y" @@ -1124,15 +1104,11 @@ def _parse_signal_lines(signal_lines): # Note: Never set a field to None. [None]* n_sig is accurate, indicating # that different channels can be present or missing. if signal_fields[field][ch] == "": - signal_fields[field][ch] = SIGNAL_SPECS.loc[ - field, "read_default" - ] + signal_fields[field][ch] = SIGNAL_SPECS.loc[field, "read_default"] # Special case: missing baseline defaults to ADCzero if present if field == "baseline" and signal_fields["adc_zero"][ch] != "": - signal_fields["baseline"][ch] = int( - signal_fields["adc_zero"][ch] - ) + signal_fields["baseline"][ch] = int(signal_fields["adc_zero"][ch]) # Typecast non-empty strings for numerical fields else: if SIGNAL_SPECS.loc[field, "allowed_types"] is int_types: @@ -1140,10 +1116,7 @@ def _parse_signal_lines(signal_lines): elif SIGNAL_SPECS.loc[field, "allowed_types"] is float_types: signal_fields[field][ch] = float(signal_fields[field][ch]) # Special case: adc_gain of 0 means 200 - if ( - field == "adc_gain" - and signal_fields["adc_gain"][ch] == 0 - ): + if field == "adc_gain" and signal_fields["adc_gain"][ch] == 0: signal_fields["adc_gain"][ch] = 200.0 return signal_fields diff --git a/wfdb/io/_signal.py b/wfdb/io/_signal.py index 5bad70c1..19c22092 100644 --- a/wfdb/io/_signal.py +++ b/wfdb/io/_signal.py @@ -6,7 +6,7 @@ import fsspec import numpy as np -from wfdb.io import download, _coreio, util +from wfdb.io import _coreio, download, util from wfdb.io._coreio import CLOUD_PROTOCOLS MAX_I32 = 2147483647 @@ -109,7 +109,7 @@ } -class SignalMixin(object): +class SignalMixin: """ Mixin class with signal methods. Inherited by Record class. @@ -190,9 +190,7 @@ def check_sig_cohesion(self, write_fields, expanded): # Match the actual signal shape against stated length and number of channels if self.n_sig != len(self.e_d_signal): - raise ValueError( - "n_sig does not match the length of e_d_signal" - ) + raise ValueError("n_sig does not match the length of e_d_signal") for ch in range(self.n_sig): if len(self.e_d_signal[ch]) != spf[ch] * self.sig_len: raise ValueError( @@ -251,9 +249,7 @@ def check_sig_cohesion(self, write_fields, expanded): print("sig_len: ", self.sig_len) print("n_sig: ", self.n_sig) print("d_signal.shape: ", self.d_signal.shape) - raise ValueError( - "sig_len and n_sig do not match shape of d_signal" - ) + raise ValueError("sig_len and n_sig do not match shape of d_signal") # For each channel (if any), make sure the digital format has no values out of bounds for ch in range(self.n_sig): @@ -279,9 +275,7 @@ def check_sig_cohesion(self, write_fields, expanded): if "checksum" in write_fields: realchecksum = self.calc_checksum() if self.checksum != realchecksum: - print( - "The actual checksum of d_signal is: ", realchecksum - ) + print("The actual checksum of d_signal is: ", realchecksum) raise ValueError( "checksum field does not match actual checksum of d_signal" ) @@ -341,9 +335,7 @@ def set_p_features(self, do_dac=False, expanded=False): # Use e_p_signal to set fields self.check_field("e_p_signal", "all") - self.sig_len = int( - len(self.e_p_signal[0]) / self.samps_per_frame[0] - ) + self.sig_len = int(len(self.e_p_signal[0]) / self.samps_per_frame[0]) self.n_sig = len(self.e_p_signal) else: if do_dac: @@ -439,9 +431,7 @@ def set_d_features(self, do_adc=False, single_fmt=True, expanded=False): # Use e_d_signal to set fields self.check_field("e_d_signal", "all") - self.sig_len = int( - len(self.e_d_signal[0]) / self.samps_per_frame[0] - ) + self.sig_len = int(len(self.e_d_signal[0]) / self.samps_per_frame[0]) self.n_sig = len(self.e_d_signal) self.init_value = [sig[0] for sig in self.e_d_signal] self.checksum = self.calc_checksum(expanded) @@ -685,9 +675,7 @@ def dac(self, expanded=False, return_res=64, inplace=False): for ch in range(self.n_sig): # NAN locations for the channel ch_nanlocs = self.e_d_signal[ch] == d_nans[ch] - ch_p_signal = self.e_d_signal[ch].astype( - floatdtype, copy=False - ) + ch_p_signal = self.e_d_signal[ch].astype(floatdtype, copy=False) np.subtract(ch_p_signal, self.baseline[ch], ch_p_signal) np.divide(ch_p_signal, self.adc_gain[ch], ch_p_signal) ch_p_signal[ch_nanlocs] = np.nan @@ -832,9 +820,7 @@ def calc_adc_params(self): maxvals = np.nanmax(self.p_signal, axis=0) for ch in range(np.shape(self.p_signal)[1]): - adc_gain, baseline = self.calc_adc_gain_baseline( - ch, minvals, maxvals - ) + adc_gain, baseline = self.calc_adc_gain_baseline(ch, minvals, maxvals) adc_gains.append(adc_gain) baselines.append(baseline) @@ -851,16 +837,12 @@ def calc_adc_params(self): raise ValueError("Signal contains inf. Cannot perform adc.") for ch, _ in enumerate(self.e_p_signal): - adc_gain, baseline = self.calc_adc_gain_baseline( - ch, minvals, maxvals - ) + adc_gain, baseline = self.calc_adc_gain_baseline(ch, minvals, maxvals) adc_gains.append(adc_gain) baselines.append(baseline) else: - raise Exception( - "Must supply p_signal or e_p_signal to calc_adc_params" - ) + raise Exception("Must supply p_signal or e_p_signal to calc_adc_params") return (adc_gains, baselines) @@ -897,9 +879,7 @@ def convert_dtype(self, physical, return_res, smooth_frames): if smooth_frames: current_dtype = self.p_signal.dtype if current_dtype != return_dtype: - self.p_signal = self.p_signal.astype( - return_dtype, copy=False - ) + self.p_signal = self.p_signal.astype(return_dtype, copy=False) else: for ch in range(self.n_sig): if self.e_p_signal[ch].dtype != return_dtype: @@ -916,17 +896,13 @@ def convert_dtype(self, physical, return_res, smooth_frames): raise Exception( "Cannot convert digital samples to lower dtype. Risk of overflow/underflow." ) - self.d_signal = self.d_signal.astype( - return_dtype, copy=False - ) + self.d_signal = self.d_signal.astype(return_dtype, copy=False) else: for ch in range(self.n_sig): current_dtype = self.e_d_signal[ch].dtype if current_dtype != return_dtype: # Do not allow changing integer dtype to lower value due to over/underflow - if int(str(current_dtype)[3:]) > int( - str(return_dtype)[3:] - ): + if int(str(current_dtype)[3:]) > int(str(return_dtype)[3:]): raise Exception( "Cannot convert digital samples to lower dtype. Risk of overflow/underflow." ) @@ -1067,8 +1043,7 @@ def smooth_frames(self, sigtype="physical"): if len(expanded_signal[ch]) != sig_len * spf[ch]: raise ValueError( "length mismatch: signal %d has %d samples," - " expected %dx%d" - % (ch, len(expanded_signal), sig_len, spf[ch]) + " expected %dx%d" % (ch, len(expanded_signal), sig_len, spf[ch]) ) input_dtypes.add(expanded_signal[ch].dtype) @@ -1402,18 +1377,14 @@ def _rd_dat_signals( sig_data = np.concatenate( ( data_to_read, - np.zeros( - n_extra_bytes, dtype=np.dtype(DATA_LOAD_TYPES[fmt]) - ), + np.zeros(n_extra_bytes, dtype=np.dtype(DATA_LOAD_TYPES[fmt])), ) ) else: sig_data = np.concatenate( ( data_to_read, - np.zeros( - extra_flat_samples, dtype=np.dtype(DATA_LOAD_TYPES[fmt]) - ), + np.zeros(extra_flat_samples, dtype=np.dtype(DATA_LOAD_TYPES[fmt])), ) ) else: @@ -1475,9 +1446,7 @@ def _rd_dat_signals( signal.append(ch_signal) ch_start = ch_end # Skew the signal - signal = _skew_sig( - signal, skew, n_sig, read_len, fmt, nan_replace, samps_per_frame - ) + signal = _skew_sig(signal, skew, n_sig, read_len, fmt, nan_replace, samps_per_frame) # Integrity check of signal shape after reading _check_sig_dims(signal, read_len, n_sig, samps_per_frame) @@ -1749,9 +1718,7 @@ def _blocks_to_samples(sig_data, n_samp, fmt): sig[0::2] = sig_data[0::3] + 256 * np.bitwise_and(sig_data[1::3], 0x0F) # Odd numbered samples (len(sig) always > 1 due to processing of # whole blocks) - sig[1::2] = sig_data[2::3] + 256 * np.bitwise_and( - sig_data[1::3] >> 4, 0x0F - ) + sig[1::2] = sig_data[2::3] + 256 * np.bitwise_and(sig_data[1::3] >> 4, 0x0F) # Remove trailing sample read within the byte block if # originally odd sampled @@ -1768,13 +1735,13 @@ def _blocks_to_samples(sig_data, n_samp, fmt): # One sample triplet is stored in one byte quartet # First sample is 7 msb of first byte and 3 lsb of second byte. - sig[0::3] = (sig_data[0::4] >> 1)[ - 0 : len(sig[0::3]) - ] + 128 * np.bitwise_and(sig_data[1::4], 0x07)[0 : len(sig[0::3])] + sig[0::3] = (sig_data[0::4] >> 1)[0 : len(sig[0::3])] + 128 * np.bitwise_and( + sig_data[1::4], 0x07 + )[0 : len(sig[0::3])] # Second signal is 7 msb of third byte and 3 lsb of forth byte - sig[1::3] = (sig_data[2::4] >> 1)[ - 0 : len(sig[1::3]) - ] + 128 * np.bitwise_and(sig_data[3::4], 0x07)[0 : len(sig[1::3])] + sig[1::3] = (sig_data[2::4] >> 1)[0 : len(sig[1::3])] + 128 * np.bitwise_and( + sig_data[3::4], 0x07 + )[0 : len(sig[1::3])] # Third signal is 5 msb of second byte and 5 msb of forth byte sig[2::3] = ( np.bitwise_and((sig_data[1::4] >> 3), 0x1F)[0 : len(sig[2::3])] @@ -1796,13 +1763,13 @@ def _blocks_to_samples(sig_data, n_samp, fmt): + 256 * np.bitwise_and(sig_data[1::4], 0x03)[0 : len(sig[0::3])] ) # Second sample is 6 msb of second byte and 4 lsb of third byte - sig[1::3] = (sig_data[1::4] >> 2)[ - 0 : len(sig[1::3]) - ] + 64 * np.bitwise_and(sig_data[2::4], 0x0F)[0 : len(sig[1::3])] + sig[1::3] = (sig_data[1::4] >> 2)[0 : len(sig[1::3])] + 64 * np.bitwise_and( + sig_data[2::4], 0x0F + )[0 : len(sig[1::3])] # Third sample is 4 msb of third byte and 6 msb of forth byte - sig[2::3] = (sig_data[2::4] >> 4)[ - 0 : len(sig[2::3]) - ] + 16 * np.bitwise_and(sig_data[3::4], 0x7F)[0 : len(sig[2::3])] + sig[2::3] = (sig_data[2::4] >> 4)[0 : len(sig[2::3])] + 16 * np.bitwise_and( + sig_data[3::4], 0x7F + )[0 : len(sig[2::3])] # Loaded values as un_signed. Convert to 2's complement form. # Values > 2^9-1 are negative. @@ -2003,9 +1970,7 @@ def _rd_compressed_file( return sig_data.reshape(-1) -def _skew_sig( - sig, skew, n_sig, read_len, fmt, nan_replace, samps_per_frame=None -): +def _skew_sig(sig, skew, n_sig, read_len, fmt, nan_replace, samps_per_frame=None): """ Skew the signal, insert nans, and shave off end of array if needed. @@ -2419,9 +2384,7 @@ def wr_dat_file( # Triplet 1 from lowest 8 bits of sample 1 b_write[0::3] = d_signal[0::2] & 255 # Triplet 2 from highest 4 bits of samples 1 (lower) and 2 (upper) - b_write[1::3] = ((d_signal[0::2] & 3840) >> 8) + ( - (d_signal[1::2] & 3840) >> 4 - ) + b_write[1::3] = ((d_signal[0::2] & 3840) >> 8) + ((d_signal[1::2] & 3840) >> 4) # Triplet 3 from lowest 8 bits of sample 2 b_write[2::3] = d_signal[1::2] & 255 @@ -2625,9 +2588,7 @@ def _infer_sig_len( "Cloud paths should be passed under record_name, not under pn_dir" ) - file_size = download._remote_file_size( - file_name=file_name, pn_dir=pn_dir - ) + file_size = download._remote_file_size(file_name=file_name, pn_dir=pn_dir) # If it isn't a cloud path or a PhysioNet path, we treat as a local file else: diff --git a/wfdb/io/_url.py b/wfdb/io/_url.py index a5b3c5d8..adc5245b 100644 --- a/wfdb/io/_url.py +++ b/wfdb/io/_url.py @@ -159,9 +159,7 @@ def __init__(self, url, start, end): } session = _get_session() - self._response = session.request( - method, url, headers=headers, stream=True - ) + self._response = session.request(method, url, headers=headers, stream=True) self._content_iter = self._response.iter_content(4096) try: self._parse_headers(method, self._response) @@ -201,9 +199,7 @@ def _parse_headers(self, method, response): # Raise an exception if an error occurs. if response.status_code >= 400 and response.status_code != 416: - _LOGGER.info( - "%s %s: %s", method, response.url, response.status_code - ) + _LOGGER.info("%s %s: %s", method, response.url, response.status_code) if response.status_code in (401, 403): cls = NetFilePermissionError elif response.status_code == 404: @@ -791,8 +787,6 @@ def openurl( if mode == "rb": return nf elif mode == "r" or mode == "rt": - return io.TextIOWrapper( - nf, encoding=encoding, errors=errors, newline=newline - ) + return io.TextIOWrapper(nf, encoding=encoding, errors=errors, newline=newline) else: return ValueError("invalid mode: %r" % (mode,)) diff --git a/wfdb/io/annotation.py b/wfdb/io/annotation.py index d3b9055a..74588def 100644 --- a/wfdb/io/annotation.py +++ b/wfdb/io/annotation.py @@ -1,20 +1,18 @@ import copy -import fsspec -import numpy as np import os -import pandas as pd -import re import posixpath +import re import sys -from wfdb.io import download -from wfdb.io import _header -from wfdb.io import record -from wfdb.io import util +import fsspec +import numpy as np +import pandas as pd + +from wfdb.io import _header, download, record, util from wfdb.io._coreio import CLOUD_PROTOCOLS -class Annotation(object): +class Annotation: """ The class representing WFDB annotations. @@ -464,9 +462,7 @@ def check_field(self, field): if len(item) != len(set(symbol)): raise ValueError( - "The symbol values of the " - + field - + " field must be unique" + "The symbol values of the " + field + " field must be unique" ) for i in range(len(item)): @@ -478,9 +474,7 @@ def check_field(self, field): + " field must be integer-like" ) - if not isinstance(symbol[i], str_types) or len( - symbol[i] - ) not in [ + if not isinstance(symbol[i], str_types) or len(symbol[i]) not in [ 1, 2, 3, @@ -546,9 +540,7 @@ def check_field(self, field): if len(self.sample) == 1: sampdiffs = np.array([self.sample[0]]) elif len(self.sample) > 1: - sampdiffs = np.concatenate( - ([self.sample[0]], np.diff(self.sample)) - ) + sampdiffs = np.concatenate(([self.sample[0]], np.diff(self.sample))) else: raise ValueError( "The 'sample' field must be a numpy array with length greater than 0" @@ -564,9 +556,7 @@ def check_field(self, field): elif field == "label_store": if min(item) < 1 or max(item) > 49: - raise ValueError( - "The label_store values must be between 1 and 49" - ) + raise ValueError("The label_store values must be between 1 and 49") # The C WFDB library stores num/sub/chan as chars. elif field == "subtype": @@ -721,13 +711,9 @@ def standardize_custom_labels(self): "There are more custom_label definitions than storage values available for them." ) - custom_labels["label_store"] = available_label_stores[ - :n_custom_labels - ] + custom_labels["label_store"] = available_label_stores[:n_custom_labels] - custom_labels = custom_labels.set_index( - custom_labels["label_store"].values - ) + custom_labels = custom_labels.set_index(custom_labels["label_store"].values) custom_labels = custom_labels[list(ann_label_fields)] self.custom_labels = custom_labels @@ -806,9 +792,7 @@ def get_available_label_stores(self, usefield="tryall"): else: # the label_store values from the standard WFDB annotation labels # whose symbols are not contained in this annotation - unused_field = ( - set(ann_label_table[usefield].values) - contained_field - ) + unused_field = set(ann_label_table[usefield].values) - contained_field unused_label_stores = ann_label_table.loc[ ann_label_table[usefield] in unused_field, "label_store" ].values @@ -1184,9 +1168,7 @@ def calc_core_bytes(self): for i in range(len(sampdiff)): # Process the samp (difference) and sym items data_bytes.append( - field2bytes( - "samptype", [sampdiff[i], self.symbol[i]], typecodes - ) + field2bytes("samptype", [sampdiff[i], self.symbol[i]], typecodes) ) # Process the extra optional fields @@ -1469,9 +1451,7 @@ def convert_label_attribute( return label_map = self.create_label_map(inplace=False) - label_map = label_map.set_index( - keys=pd.Index(label_map[source_field].values) - ) + label_map = label_map.set_index(keys=pd.Index(label_map[source_field].values)) try: target_item = label_map.loc[ @@ -1953,9 +1933,7 @@ def rdann( dir_list[0], download.get_version(dir_list[0]), *dir_list[1:] ) - return_label_elements = check_read_inputs( - sampfrom, sampto, return_label_elements - ) + return_label_elements = check_read_inputs(sampfrom, sampto, return_label_elements) # Read the file in byte pairs filebytes = load_byte_pairs(record_name, extension, pn_dir) @@ -1967,9 +1945,7 @@ def rdann( # Get the indices of annotations that hold definition information about # the entire annotation file, and other empty annotations to be removed. - potential_definition_inds, rm_inds = get_special_inds( - sample, label_store, aux_note - ) + potential_definition_inds, rm_inds = get_special_inds(sample, label_store, aux_note) # Try to extract information describing the annotation file fs, custom_labels = interpret_defintion_annotations( @@ -2249,17 +2225,13 @@ def proc_core_fields(filebytes, bpi): # Not a skip - it is the actual sample number + annotation type store value label_store = int(filebytes[bpi, 1]) >> 2 - sample_diff += np.int64(filebytes[bpi, 0]) + 256 * ( - np.int64(filebytes[bpi, 1]) & 3 - ) + sample_diff += np.int64(filebytes[bpi, 0]) + 256 * (np.int64(filebytes[bpi, 1]) & 3) bpi = bpi + 1 return sample_diff, label_store, bpi -def proc_extra_field( - label_store, filebytes, bpi, subtype, chan, num, aux_note, update -): +def proc_extra_field(label_store, filebytes, bpi, subtype, chan, num, aux_note, update): """ Process extra fields belonging to the current annotation. Potential updated fields: subtype, chan, num, aux_note. @@ -2505,9 +2477,7 @@ def interpret_defintion_annotations(potential_definition_inds, aux_note): symbol, description, ) = rx_custom_label.findall(aux_note[i])[0] - custom_labels.append( - (int(label_store), symbol, description) - ) + custom_labels.append((int(label_store), symbol, description)) i += 1 i += 1 else: @@ -2722,14 +2692,13 @@ def mrgann( total_seconds = hours * 60 * 60 + minutes * 60 + seconds if (i == 1) and (total_seconds == 0): raise Exception( - "End time can not be set to the start of " "the record" + "End time can not be set to the start of the record" ) sample = int(total_seconds * ann1.fs) if sample > max([max(ann1.sample), max(ann2.sample)]): if i == 0: raise Exception( - "Start time can not be set to the " - "end of the record" + "Start time can not be set to the end of the record" ) else: print( @@ -2748,21 +2717,15 @@ def mrgann( print("Combining the two files together") # The sample should never be empty but others can (though they # shouldn't be) - both_sample = np.concatenate([ann1.sample, ann2.sample]).astype( - np.int64 - ) + both_sample = np.concatenate([ann1.sample, ann2.sample]).astype(np.int64) # Generate a list of sorted indices then sort the array sort_indices = np.argsort(both_sample) both_sample = np.sort(both_sample) # Find where to filter the array if merge_method == "combine": - sample_range = (both_sample >= start_sample) & ( - both_sample <= end_sample - ) + sample_range = (both_sample >= start_sample) & (both_sample <= end_sample) if merge_method == "delete": - sample_range = (both_sample < start_sample) | ( - both_sample > end_sample - ) + sample_range = (both_sample < start_sample) | (both_sample > end_sample) index_range = np.where(sample_range)[0] both_sample = both_sample[sample_range] # Combine both annotation attributes @@ -2829,9 +2792,9 @@ def mrgann( # shouldn't be) keep_ann_sample = keep_ann.sample[keep_index_range] remove_ann_sample = remove_ann.sample[remove_index_range] - both_sample = np.concatenate( - [keep_ann_sample, remove_ann_sample] - ).astype(np.int64) + both_sample = np.concatenate([keep_ann_sample, remove_ann_sample]).astype( + np.int64 + ) # Generate a list of sorted indices then sort the array sort_indices = np.argsort(both_sample) both_sample = np.sort(both_sample) @@ -2876,15 +2839,11 @@ def mrgann( if remove_cat is not None else blank_array ) - temp_cat = np.concatenate([keep_cat, remove_cat]).astype( - np.int64 - ) + temp_cat = np.concatenate([keep_cat, remove_cat]).astype(np.int64) if temp_cat.shape[0] == 0: ann_attr[cat] = None else: - ann_attr[cat] = np.array( - [temp_cat[i] for i in sort_indices] - ) + ann_attr[cat] = np.array([temp_cat[i] for i in sort_indices]) else: raise Exception( "Invalid value for 'merge_method': options are " @@ -3013,7 +2972,7 @@ def format_ann_from_df(df_in): ann_label_fields = ("label_store", "symbol", "description") -class AnnotationClass(object): +class AnnotationClass: """ Describes the annotations. @@ -3062,13 +3021,11 @@ def __init__(self, extension, description, human_reviewed): ann_class_table = ann_class_table.set_index( keys=pd.Index(ann_class_table["extension"].values), ) -ann_class_table = ann_class_table[ - ["extension", "description", "human_reviewed"] -] +ann_class_table = ann_class_table[["extension", "description", "human_reviewed"]] # Individual annotation labels -class AnnotationLabel(object): +class AnnotationLabel: """ Describes the individual annotation labels. @@ -3166,9 +3123,7 @@ def __str__(self): AnnotationLabel(6, "F", "FUSION", "Fusion of ventricular and normal beat"), AnnotationLabel(7, "J", "NPC", "Nodal (junctional) premature beat"), AnnotationLabel(8, "A", "APC", "Atrial premature contraction"), - AnnotationLabel( - 9, "S", "SVPB", "Premature or ectopic supraventricular beat" - ), + AnnotationLabel(9, "S", "SVPB", "Premature or ectopic supraventricular beat"), AnnotationLabel(10, "E", "VESC", "Ventricular escape beat"), AnnotationLabel(11, "j", "NESC", "Nodal (junctional) escape beat"), AnnotationLabel(12, "/", "PACE", "Paced beat"), @@ -3191,24 +3146,16 @@ def __str__(self): AnnotationLabel(29, "u", "UWAVE", "U-wave peak"), AnnotationLabel(30, "?", "LEARN", "Learning"), AnnotationLabel(31, "!", "FLWAV", "Ventricular flutter wave"), - AnnotationLabel( - 32, "[", "VFON", "Start of ventricular flutter/fibrillation" - ), - AnnotationLabel( - 33, "]", "VFOFF", "End of ventricular flutter/fibrillation" - ), + AnnotationLabel(32, "[", "VFON", "Start of ventricular flutter/fibrillation"), + AnnotationLabel(33, "]", "VFOFF", "End of ventricular flutter/fibrillation"), AnnotationLabel(34, "e", "AESC", "Atrial escape beat"), AnnotationLabel(35, "n", "SVESC", "Supraventricular escape beat"), - AnnotationLabel( - 36, "@", "LINK", "Link to external data (aux_note contains URL)" - ), + AnnotationLabel(36, "@", "LINK", "Link to external data (aux_note contains URL)"), AnnotationLabel(37, "x", "NAPC", "Non-conducted P-wave (blocked APB)"), AnnotationLabel(38, "f", "PFUS", "Fusion of paced and normal beat"), AnnotationLabel(39, "(", "WFON", "Waveform onset"), AnnotationLabel(40, ")", "WFOFF", "Waveform end"), - AnnotationLabel( - 41, "r", "RONT", "R-on-T premature ventricular contraction" - ), + AnnotationLabel(41, "r", "RONT", "R-on-T premature ventricular contraction"), # AnnotationLabel(42, None, None, None), # AnnotationLabel(43, None, None, None), # AnnotationLabel(44, None, None, None), @@ -3221,14 +3168,10 @@ def __str__(self): ann_label_table = pd.DataFrame( { - "label_store": np.array( - [al.label_store for al in ann_labels], dtype="int" - ), + "label_store": np.array([al.label_store for al in ann_labels], dtype="int"), "symbol": [al.symbol for al in ann_labels], "description": [al.description for al in ann_labels], } ) -ann_label_table = ann_label_table.set_index( - ann_label_table["label_store"].values -) +ann_label_table = ann_label_table.set_index(ann_label_table["label_store"].values) ann_label_table = ann_label_table[["label_store", "symbol", "description"]] diff --git a/wfdb/io/convert/__init__.py b/wfdb/io/convert/__init__.py index 54cd440e..18680966 100644 --- a/wfdb/io/convert/__init__.py +++ b/wfdb/io/convert/__init__.py @@ -2,4 +2,4 @@ from wfdb.io.convert.edf import rdedfann, read_edf, wfdb_to_edf from wfdb.io.convert.matlab import wfdb_to_mat from wfdb.io.convert.tff import rdtff -from wfdb.io.convert.wav import wfdb_to_wav, read_wav +from wfdb.io.convert.wav import read_wav, wfdb_to_wav diff --git a/wfdb/io/convert/csv.py b/wfdb/io/convert/csv.py index e4fa8b68..6f9e78f3 100644 --- a/wfdb/io/convert/csv.py +++ b/wfdb/io/convert/csv.py @@ -3,7 +3,7 @@ import numpy as np import pandas as pd -from wfdb.io.annotation import format_ann_from_df, Annotation, wrann +from wfdb.io.annotation import Annotation, format_ann_from_df, wrann from wfdb.io.record import Record, wrsamp @@ -302,10 +302,10 @@ def csv_to_wfdb( # The dataframe should be in (`sig_len`, `n_sig`) dimensions sig_len = p_signal.shape[0] if verbose: - print("Signal length: {}".format(sig_len)) + print(f"Signal length: {sig_len}") n_sig = p_signal.shape[1] if verbose: - print("Number of signals: {}".format(n_sig)) + print(f"Number of signals: {n_sig}") # Check if signal names are valid and set defaults if not sig_name: @@ -316,24 +316,22 @@ def csv_to_wfdb( "WARNING: One or more of your signal names are numbers, this " "is not recommended:\n- Does your CSV have a header line " "which defines the signal names?\n- If not, please set the " - "parameter 'header' to False.\nSignal names: {}".format( - sig_name - ) + f"parameter 'header' to False.\nSignal names: {sig_name}" ) else: sig_name = ["ch_" + str(i) for i in range(n_sig)] if verbose: - print("Signal names: {}".format(sig_name)) + print(f"Signal names: {sig_name}") record_name = os.path.splitext(os.path.basename(file_name))[0] if verbose: - print("Record name: {}.hea".format(record_name)) + print(f"Record name: {record_name}.hea") # Replace the CSV file tag with DAT dat_file_name = record_name + ".dat" dat_file_name = [dat_file_name] * n_sig if verbose: - print("Output record: {}".format(dat_file_name[0])) + print(f"Output record: {dat_file_name[0]}") # Convert `units` from string to list if necessary units = [units] * n_sig if type(units) is str else units @@ -343,14 +341,14 @@ def csv_to_wfdb( fmt = ["16"] * n_sig fmt = [fmt] * n_sig if type(fmt) is str else fmt if verbose: - print("Signal format: {}".format(fmt)) + print(f"Signal format: {fmt}") # Set the default `adc_gain` if none exists if not adc_gain: adc_gain = [200] * n_sig adc_gain = [adc_gain] * n_sig if type(adc_gain) is int else adc_gain if verbose: - print("Signal ADC gain: {}".format(adc_gain)) + print(f"Signal ADC gain: {adc_gain}") # Set the default `baseline` if none exists if not baseline: @@ -360,36 +358,32 @@ def csv_to_wfdb( baseline = [0] * n_sig baseline = [baseline] * n_sig if type(baseline) is int else baseline if verbose: - print("Signal baseline: {}".format(baseline)) + print(f"Signal baseline: {baseline}") # Convert `samps_per_frame` from int to list if necessary samps_per_frame = ( - [samps_per_frame] * n_sig - if type(samps_per_frame) is int - else samps_per_frame + [samps_per_frame] * n_sig if type(samps_per_frame) is int else samps_per_frame ) # Convert `skew` from int to list if necessary skew = [skew] * n_sig if type(skew) is int else skew # Convert `byte_offset` from int to list if necessary - byte_offset = ( - [byte_offset] * n_sig if type(byte_offset) is int else byte_offset - ) + byte_offset = [byte_offset] * n_sig if type(byte_offset) is int else byte_offset # Set the default `adc_res` if none exists if not adc_res: adc_res = [12] * n_sig adc_res = [adc_res] * n_sig if type(adc_res) is int else adc_res if verbose: - print("Signal ADC resolution: {}".format(adc_res)) + print(f"Signal ADC resolution: {adc_res}") # Set the default `adc_zero` if none exists if not adc_zero: adc_zero = [0] * n_sig adc_zero = [adc_zero] * n_sig if type(adc_zero) is int else adc_zero if verbose: - print("Signal ADC zero: {}".format(adc_zero)) + print(f"Signal ADC zero: {adc_zero}") # Set the default `init_value` # NOTE: Initial value (and subsequently the digital signal) won't be correct @@ -400,20 +394,20 @@ def csv_to_wfdb( init_value = baseline + (np.array(adc_gain) * init_value) init_value = [int(i) for i in init_value.tolist()] if verbose: - print("Signal initial value: {}".format(init_value)) + print(f"Signal initial value: {init_value}") # Set the default `checksum` if not checksum: checksum = [int(np.sum(v) % 65536) for v in np.transpose(p_signal)] if verbose: - print("Signal checksum: {}".format(checksum)) + print(f"Signal checksum: {checksum}") # Set the default `block_size` if not block_size: block_size = [0] * n_sig block_size = [block_size] * n_sig if type(block_size) is int else block_size if verbose: - print("Signal block size: {}".format(block_size)) + print(f"Signal block size: {block_size}") # Convert array to floating point p_signal = p_signal.astype("float64") diff --git a/wfdb/io/convert/edf.py b/wfdb/io/convert/edf.py index f6f49415..887ca0bf 100644 --- a/wfdb/io/convert/edf.py +++ b/wfdb/io/convert/edf.py @@ -8,10 +8,9 @@ import numpy as np import pandas as pd +from wfdb.io import _url, download from wfdb.io.annotation import Annotation, format_ann_from_df, wrann -from wfdb.io.record import Record, rdrecord, SIG_UNITS -from wfdb.io import _url -from wfdb.io import download +from wfdb.io.record import SIG_UNITS, Record, rdrecord def read_edf( @@ -147,17 +146,15 @@ def read_edf( # Check to see that the input is an EDF file. (This check will detect # most but not all other types of files.) if version != "0 ": - raise Exception( - "Input does not appear to be EDF -- no conversion attempted" - ) + raise Exception("Input does not appear to be EDF -- no conversion attempted") else: if verbose: - print("EDF version number: {}".format(version.strip())) + print(f"EDF version number: {version.strip()}") # Local patient identification (80 bytes) patient_id = struct.unpack("<80s", edf_file.read(80))[0].decode(encoding) if verbose: - print("Patient ID: {}".format(patient_id)) + print(f"Patient ID: {patient_id}") # Local recording identification (80 bytes) # Bob Kemp recommends using this field to encode the start date @@ -166,12 +163,12 @@ def read_edf( # record. EDF+ requires this. record_id = struct.unpack("<80s", edf_file.read(80))[0].decode(encoding) if verbose: - print("Recording ID: {}".format(record_id)) + print(f"Recording ID: {record_id}") # Start date of recording (dd.mm.yy) (8 bytes) start_date = struct.unpack("<8s", edf_file.read(8))[0].decode(encoding) if verbose: - print("Recording Date: {}".format(start_date)) + print(f"Recording Date: {start_date}") start_day, start_month, start_year = [int(i) for i in start_date.split(".")] # This should work for a while if start_year < 1970: @@ -182,17 +179,13 @@ def read_edf( # Start time of recording (hh.mm.ss) (8 bytes) start_time = struct.unpack("<8s", edf_file.read(8))[0].decode(encoding) if verbose: - print("Recording Time: {}".format(start_time)) - start_hour, start_minute, start_second = [ - int(i) for i in start_time.split(".") - ] + print(f"Recording Time: {start_time}") + start_hour, start_minute, start_second = [int(i) for i in start_time.split(".")] # Number of bytes in header (8 bytes) - header_bytes = int( - struct.unpack("<8s", edf_file.read(8))[0].decode(encoding) - ) + header_bytes = int(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)) if verbose: - print("Number of bytes in header record: {}".format(header_bytes)) + print(f"Number of bytes in header record: {header_bytes}") # Reserved (44 bytes) reserved_notes = ( @@ -206,46 +199,36 @@ def read_edf( # Pages 1755-1761, ISSN 1388-2457 pass elif reserved_notes[:5] == "EDF+D": - raise Exception( - "EDF+ File: interrupted data records (not currently supported)" - ) + raise Exception("EDF+ File: interrupted data records (not currently supported)") else: if verbose: - print("Free Space: {}".format(reserved_notes)) + print(f"Free Space: {reserved_notes}") # Number of blocks (-1 if unknown) (8 bytes) num_blocks = int(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)) if verbose: - print("Number of data records: {}".format(num_blocks)) + print(f"Number of data records: {num_blocks}") if num_blocks == -1: - raise Exception( - "Number of data records in unknown (not currently supported)" - ) + raise Exception("Number of data records in unknown (not currently supported)") # Duration of a block, in seconds (8 bytes) - block_duration = float( - struct.unpack("<8s", edf_file.read(8))[0].decode(encoding) - ) + block_duration = float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)) if verbose: - print( - "Duration of each data record in seconds: {}".format(block_duration) - ) + print(f"Duration of each data record in seconds: {block_duration}") if block_duration <= 0.0: block_duration = 1.0 # Number of signals (4 bytes) n_sig = int(struct.unpack("<4s", edf_file.read(4))[0].decode(encoding)) if verbose: - print("Number of signals: {}".format(n_sig)) + print(f"Number of signals: {n_sig}") if n_sig < 1: raise Exception("Done: not any signals left to read") # Label (e.g., EEG FpzCz or Body temp) (16 bytes each) sig_name = [] for _ in range(n_sig): - temp_sig = ( - struct.unpack("<16s", edf_file.read(16))[0].decode(encoding).strip() - ) + temp_sig = struct.unpack("<16s", edf_file.read(16))[0].decode(encoding).strip() if temp_sig == "EDF Annotations" and not rdedfann_flag: print( "*** This may be an EDF+ Annotation file instead, please see " @@ -253,7 +236,7 @@ def read_edf( ) sig_name.append(temp_sig) if verbose: - print("Signal Labels: {}".format(sig_name)) + print(f"Signal Labels: {sig_name}") # Transducer type (e.g., AgAgCl electrode) (80 bytes each) transducer_types = [] @@ -262,7 +245,7 @@ def read_edf( struct.unpack("<80s", edf_file.read(80))[0].decode(encoding).strip() ) if verbose: - print("Transducer Types: {}".format(transducer_types)) + print(f"Transducer Types: {transducer_types}") # Physical dimension (e.g., uV or degreeC) (8 bytes each) physical_dims = [] @@ -271,7 +254,7 @@ def read_edf( struct.unpack("<8s", edf_file.read(8))[0].decode(encoding).strip() ) if verbose: - print("Physical Dimensions: {}".format(physical_dims)) + print(f"Physical Dimensions: {physical_dims}") # Physical minimum (e.g., -500 or 34) (8 bytes each) physical_min = np.array([]) @@ -281,7 +264,7 @@ def read_edf( float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)), ) if verbose: - print("Physical Minimums: {}".format(physical_min)) + print(f"Physical Minimums: {physical_min}") # Physical maximum (e.g., 500 or 40) (8 bytes each) physical_max = np.array([]) @@ -291,7 +274,7 @@ def read_edf( float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)), ) if verbose: - print("Physical Maximums: {}".format(physical_max)) + print(f"Physical Maximums: {physical_max}") # Digital minimum (e.g., -2048) (8 bytes each) digital_min = np.array([]) @@ -301,7 +284,7 @@ def read_edf( float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)), ) if verbose: - print("Digital Minimums: {}".format(digital_min)) + print(f"Digital Minimums: {digital_min}") # Digital maximum (e.g., 2047) (8 bytes each) digital_max = np.array([]) @@ -311,7 +294,7 @@ def read_edf( float(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)), ) if verbose: - print("Digital Maximums: {}".format(digital_max)) + print(f"Digital Maximums: {digital_max}") # Prefiltering (e.g., HP:0.1Hz LP:75Hz) (80 bytes each) prefilter_info = [] @@ -320,7 +303,7 @@ def read_edf( struct.unpack("<80s", edf_file.read(80))[0].decode(encoding).strip() ) if verbose: - print("Prefiltering Information: {}".format(prefilter_info)) + print(f"Prefiltering Information: {prefilter_info}") # Number of samples per block (8 bytes each) samps_per_block = [] @@ -329,7 +312,7 @@ def read_edf( int(struct.unpack("<8s", edf_file.read(8))[0].decode(encoding)) ) if verbose: - print("Number of Samples per Record: {}".format(samps_per_block)) + print(f"Number of Samples per Record: {samps_per_block}") # The last 32*nsig bytes in the header are unused for _ in range(n_sig): @@ -638,12 +621,10 @@ def wfdb_to_edf( if bytes_per_block > edf_max_block: print( - ( - "Can't convert record %s to EDF: EDF blocks can't be larger " - "than {} bytes, but each input frame requires {} bytes. Use " - "'channels' to select a subset of the input signals or trim " - "using 'sampfrom' and 'sampto'." - ).format(edf_max_block, samples_per_frame * 2) + "Can't convert record %s to EDF: EDF blocks can't be larger " + f"than {edf_max_block} bytes, but each input frame requires {samples_per_frame * 2} bytes. Use " + "'channels' to select a subset of the input signals or trim " + "using 'sampfrom' and 'sampto'." ) # Calculate the number of blocks to be written. The calculation rounds @@ -676,16 +657,10 @@ def wfdb_to_edf( else: temp_adc_res = record.adc_res[i] # Determine the physical and digital extrema - digital_max.append( - int(record.adc_zero[i] + (1 << (temp_adc_res - 1)) - 1) - ) + digital_max.append(int(record.adc_zero[i] + (1 << (temp_adc_res - 1)) - 1)) digital_min.append(int(record.adc_zero[i] - (1 << (temp_adc_res - 1)))) - physical_max.append( - (digital_max[i] - record.baseline[i]) / record.adc_gain[i] - ) - physical_min.append( - (digital_min[i] - record.baseline[i]) / record.adc_gain[i] - ) + physical_max.append((digital_max[i] - record.baseline[i]) / record.adc_gain[i]) + physical_min.append((digital_min[i] - record.baseline[i]) / record.adc_gain[i]) # The maximum record name length to write is 80 bytes if len(record_name_out) > 80: @@ -779,9 +754,9 @@ def wfdb_to_edf( # Local patient identification (80 bytes) f.write( - struct.pack( - "<80s", "{}".format(record_name_write).encode("ascii") - ).replace(b"\x00", b"\x20") + struct.pack("<80s", f"{record_name_write}".encode("ascii")).replace( + b"\x00", b"\x20" + ) ) # Local recording identification (80 bytes) @@ -793,16 +768,14 @@ def wfdb_to_edf( f.write( struct.pack( "<80s", - "Startdate {}-{}-{}".format( - start_day, month_names[start_month - 1], start_year - ).encode("ascii"), + f"Startdate {start_day}-{month_names[start_month - 1]}-{start_year}".encode( + "ascii" + ), ).replace(b"\x00", b"\x20") ) else: f.write( - struct.pack("<80s", b"Startdate not recorded").replace( - b"\x00", b"\x20" - ) + struct.pack("<80s", b"Startdate not recorded").replace(b"\x00", b"\x20") ) if edf_plus: print("WARNING: EDF+ requires start date (not specified)") @@ -811,9 +784,9 @@ def wfdb_to_edf( f.write( struct.pack( "<8s", - "{:02d}.{:02d}.{:02d}".format( - start_day, start_month, start_year % 100 - ).encode("ascii"), + f"{start_day:02d}.{start_month:02d}.{start_year % 100:02d}".encode( + "ascii" + ), ).replace(b"\x00", b"\x20") ) @@ -821,17 +794,17 @@ def wfdb_to_edf( f.write( struct.pack( "<8s", - "{:02d}.{:02d}.{:02d}".format( - start_hour, start_minute, start_second - ).encode("ascii"), + f"{start_hour:02d}.{start_minute:02d}.{start_second:02d}".encode( + "ascii" + ), ).replace(b"\x00", b"\x20") ) # Number of bytes in header (8 bytes) f.write( - struct.pack( - "<8s", "{:d}".format(header_bytes).encode("ascii") - ).replace(b"\x00", b"\x20") + struct.pack("<8s", f"{header_bytes:d}".encode("ascii")).replace( + b"\x00", b"\x20" + ) ) # Reserved (44 bytes) @@ -842,31 +815,29 @@ def wfdb_to_edf( # Number of blocks (-1 if unknown) (8 bytes) f.write( - struct.pack( - "<8s", "{:d}".format(num_blocks).encode("ascii") - ).replace(b"\x00", b"\x20") + struct.pack("<8s", f"{num_blocks:d}".encode("ascii")).replace( + b"\x00", b"\x20" + ) ) # Duration of a block, in seconds (8 bytes) f.write( - struct.pack( - "<8s", "{:g}".format(seconds_per_block_write).encode("ascii") - ).replace(b"\x00", b"\x20") + struct.pack("<8s", f"{seconds_per_block_write:g}".encode("ascii")).replace( + b"\x00", b"\x20" + ) ) # Number of signals (4 bytes) f.write( - struct.pack( - "<4s", "{:d}".format(record.n_sig).encode("ascii") - ).replace(b"\x00", b"\x20") + struct.pack("<4s", f"{record.n_sig:d}".encode("ascii")).replace( + b"\x00", b"\x20" + ) ) # Label (e.g., EEG FpzCz or Body temp) (16 bytes each) for i in sig_name_write: f.write( - struct.pack("<16s", "{}".format(i).encode("ascii")).replace( - b"\x00", b"\x20" - ) + struct.pack("<16s", f"{i}".encode("ascii")).replace(b"\x00", b"\x20") ) # Transducer type (e.g., AgAgCl electrode) (80 bytes each) @@ -880,15 +851,13 @@ def wfdb_to_edf( # Physical dimension (e.g., uV or degreeC) (8 bytes each) for i in units_write: f.write( - struct.pack("<8s", "{}".format(i).encode("ascii")).replace( - b"\x00", b"\x20" - ) + struct.pack("<8s", f"{i}".encode("ascii")).replace(b"\x00", b"\x20") ) # Physical minimum (e.g., -500 or 34) (8 bytes each) for pmin in physical_min: f.write( - struct.pack("<8s", "{:g}".format(pmin).encode("ascii")).replace( + struct.pack("<8s", f"{pmin:g}".encode("ascii")).replace( b"\x00", b"\x20" ) ) @@ -896,7 +865,7 @@ def wfdb_to_edf( # Physical maximum (e.g., 500 or 40) (8 bytes each) for pmax in physical_max: f.write( - struct.pack("<8s", "{:g}".format(pmax).encode("ascii")).replace( + struct.pack("<8s", f"{pmax:g}".encode("ascii")).replace( b"\x00", b"\x20" ) ) @@ -904,7 +873,7 @@ def wfdb_to_edf( # Digital minimum (e.g., -2048) (8 bytes each) for dmin in digital_min: f.write( - struct.pack("<8s", "{:d}".format(dmin).encode("ascii")).replace( + struct.pack("<8s", f"{dmin:d}".encode("ascii")).replace( b"\x00", b"\x20" ) ) @@ -912,7 +881,7 @@ def wfdb_to_edf( # Digital maximum (e.g., 2047) (8 bytes each) for dmax in digital_max: f.write( - struct.pack("<8s", "{:d}".format(dmax).encode("ascii")).replace( + struct.pack("<8s", f"{dmax:d}".encode("ascii")).replace( b"\x00", b"\x20" ) ) @@ -928,9 +897,7 @@ def wfdb_to_edf( # Number of samples per block (8 bytes each) for spr in samps_per_record: f.write( - struct.pack("<8s", "{:d}".format(spr).encode("ascii")).replace( - b"\x00", b"\x20" - ) + struct.pack("<8s", f"{spr:d}".encode("ascii")).replace(b"\x00", b"\x20") ) # The last 32*nsig bytes in the header are unused @@ -947,11 +914,9 @@ def wfdb_to_edf( for i in range(num_to_write): f.write(b"\x00\x80") - print("Header block size: {:d} bytes".format((record.n_sig + 1) * 256)) + print(f"Header block size: {(record.n_sig + 1) * 256:d} bytes") print( - "Data block size: {:g} seconds ({:d} frames or {:d} bytes)".format( - seconds_per_block, int(frames_per_block), int(bytes_per_block) - ) + f"Data block size: {seconds_per_block:g} seconds ({int(frames_per_block):d} frames or {int(bytes_per_block):d} bytes)" ) print( "Recording length: {:d} ({:d} data blocks, {:d} frames, {:d} bytes)".format( @@ -968,22 +933,18 @@ def wfdb_to_edf( ) ) print( - "Total length of file to be written: {:d} bytes".format( - int((record.n_sig + 1) * 256 + num_blocks * bytes_per_block) - ) + f"Total length of file to be written: {int((record.n_sig + 1) * 256 + num_blocks * bytes_per_block):d} bytes" ) if edf_plus: print( - ( - "WARNING: EDF+ requires the subject's gender, birthdate, and name, as " - "well as additional information about the recording that is not usually " - "available. This information is not saved in the output file even if " - "available. EDF+ also requires the use of standard names for signals and " - "for physical units; these requirements are not enforced by this program. " - "To make the output file fully EDF+ compliant, its header must be edited " - "manually." - ) + "WARNING: EDF+ requires the subject's gender, birthdate, and name, as " + "well as additional information about the recording that is not usually " + "available. This information is not saved in the output file even if " + "available. EDF+ also requires the use of standard names for signals and " + "for physical units; these requirements are not enforced by this program. " + "To make the output file fully EDF+ compliant, its header must be edited " + "manually." ) if "EDF-Annotations" not in record.sig_name: @@ -998,11 +959,7 @@ def wfdb_to_edf( header_test = open(output_filename, "rb").read((record.n_sig + 1) * 256) for i, val in enumerate(header_test): if (val < 32) or (val > 126): - print( - "WARNING: output contains an invalid character, {}, at byte {}".format( - val, i - ) - ) + print(f"WARNING: output contains an invalid character, {val}, at byte {i}") def rdedfann( @@ -1104,8 +1061,7 @@ def rdedfann( # Some preliminary checks if info_only and record_only: raise Exception( - "Both `info_only` and `record_only` are set. Only one " - "can be set at a time." + "Both `info_only` and `record_only` are set. Only one can be set at a time." ) # According to the EDF+ docs: @@ -1126,12 +1082,8 @@ def rdedfann( if chunk + 1 == 0: continue else: - adjusted_hex = hex( - struct.unpack("H", chunk + 1))[0] - ) - annotation_string += bytes.fromhex(adjusted_hex[2:]).decode( - encoding - ) + adjusted_hex = hex(struct.unpack("H", chunk + 1))[0]) + annotation_string += bytes.fromhex(adjusted_hex[2:]).decode(encoding) # Remove all of the whitespace for rep in ["\x00", "\x14", "\x15"]: annotation_string = annotation_string.replace(rep, " ") @@ -1156,9 +1108,7 @@ def rdedfann( duration = float(ann_split[1]) comment = " ".join(ann_split[2:]) if verbose: - print( - f"{onset_time}\t{sample_num}\t{comment}\t\tduration: {duration}" - ) + print(f"{onset_time}\t{sample_num}\t{comment}\t\tduration: {duration}") onsets.append(onset) onset_times.append(onset_time) sample_nums.append(sample_num) diff --git a/wfdb/io/convert/matlab.py b/wfdb/io/convert/matlab.py index 61d06deb..cd932927 100644 --- a/wfdb/io/convert/matlab.py +++ b/wfdb/io/convert/matlab.py @@ -7,9 +7,7 @@ from wfdb.io.record import rdrecord -def wfdb_to_mat( - record_name, pn_dir=None, sampfrom=0, sampto=None, channels=None -): +def wfdb_to_mat(record_name, pn_dir=None, sampfrom=0, sampto=None, channels=None): """ This program converts the signals of any PhysioNet record (or one in any compatible format) into a .mat file that can be read directly using any version @@ -93,9 +91,7 @@ def wfdb_to_mat( The output file name is 100m.mat and 100m.hea """ - record = rdrecord( - record_name, pn_dir=pn_dir, sampfrom=sampfrom, sampto=sampto - ) + record = rdrecord(record_name, pn_dir=pn_dir, sampfrom=sampfrom, sampto=sampto) record_name_out = record_name.split(os.sep)[-1].replace("-", "_") + "m" # Some variables describing the format of the .mat file @@ -225,11 +221,7 @@ def wfdb_to_mat( # = 192 # Copy the selected data into the .mat file - out_data = ( - record.p_signal * record.adc_gain - + record.baseline - - record.adc_zero - ) + out_data = record.p_signal * record.adc_gain + record.baseline - record.adc_zero # Cast the data to the correct type base on the bytes_per_element out_data = np.around(out_data).astype(out_type) # out_data should be [r1c1, r1c2, r2c1, r2c2, etc.] @@ -242,50 +234,30 @@ def wfdb_to_mat( if record.base_date is None: datetime_string = "[None]" else: - datetime_string = "[{}]".format( - record.base_date.strftime("%d/%m/%Y") - ) + datetime_string = "[{}]".format(record.base_date.strftime("%d/%m/%Y")) else: if record.base_date is None: - datetime_string = "[{}]".format( - record.base_time.strftime("%H:%M:%S.%f") - ) + datetime_string = "[{}]".format(record.base_time.strftime("%H:%M:%S.%f")) else: datetime_string = "[{} {}]".format( record.base_time.strftime("%H:%M:%S.%f"), record.base_date.strftime("%d/%m/%Y"), ) - print("Source: record {}\t\tStart: {}".format(record_name, datetime_string)) + print(f"Source: record {record_name}\t\tStart: {datetime_string}") print( - "val has {} rows (signals) and {} columns (samples/signal)".format( - record.n_sig, desired_length - ) - ) - duration_string = str( - datetime.timedelta(seconds=desired_length / record.fs) + f"val has {record.n_sig} rows (signals) and {desired_length} columns (samples/signal)" ) - print("Duration: {}".format(duration_string)) + duration_string = str(datetime.timedelta(seconds=desired_length / record.fs)) + print(f"Duration: {duration_string}") + print(f"Sampling frequency: {record.fs} Hz\tSampling interval: {1 / record.fs} sec") print( - "Sampling frequency: {} Hz\tSampling interval: {} sec".format( - record.fs, 1 / record.fs - ) - ) - print( - "{:<7}{:<20}{:<17}{:<10}{:<10}".format( - "Row", "Signal", "Gain", "Base", "Units" - ) + "{:<7}{:<20}{:<17}{:<10}{:<10}".format("Row", "Signal", "Gain", "Base", "Units") ) record.sig_name = [s.replace(" ", "_") for s in record.sig_name] for i in range(record.n_sig): print( - "{:<7}{:<20}{:<17}{:<10}{:<10}".format( - i, - record.sig_name[i], - record.adc_gain[i], - record.baseline[i] - record.adc_zero[i] + offset, - record.units[i], - ) + f"{i:<7}{record.sig_name[i]:<20}{record.adc_gain[i]:<17}{record.baseline[i] - record.adc_zero[i] + offset:<10}{record.units[i]:<10}" ) # Modify the record file to reflect the new data @@ -296,9 +268,7 @@ def wfdb_to_mat( record.file_name = num_channels * [output_file] record.fmt = num_channels * [wfdb_type] record.byte_offset = num_channels * [192] - record.baseline = [ - b - record.adc_zero[i] for i, b in enumerate(record.baseline) - ] + record.baseline = [b - record.adc_zero[i] for i, b in enumerate(record.baseline)] record.adc_zero = num_channels * [0] record.init_value = out_data[: record.n_sig].tolist() @@ -307,4 +277,4 @@ def wfdb_to_mat( # Append the following lines to create a signature with open(record_name_out + ".hea", "a") as f: f.write("#Creator: wfdb2mat\n") - f.write("#Source: record {}\n".format(record_name)) + f.write(f"#Source: record {record_name}\n") diff --git a/wfdb/io/convert/tff.py b/wfdb/io/convert/tff.py index 355a3eaf..312d8715 100644 --- a/wfdb/io/convert/tff.py +++ b/wfdb/io/convert/tff.py @@ -92,9 +92,7 @@ def _rdheader(fp): # storage method if tag == 1001: storage_method = fs = struct.unpack("B", fp.read(1))[0] - storage_method = {0: "recording", 1: "manual", 2: "online"}[ - storage_method - ] + storage_method = {0: "recording", 1: "manual", 2: "online"}[storage_method] # fs, unit16 elif tag == 1003: fs = struct.unpack(">H", fp.read(2))[0] @@ -130,9 +128,7 @@ def _rdheader(fp): if item[0] <= data <= item[1]: base_name = item[2] break - existing_count = [base_name in name for name in sig_name].count( - True - ) + existing_count = [base_name in name for name in sig_name].count(True) sig_name.append("%s_%d" % (base_name, existing_count)) # Display scale. Probably not useful. elif tag == 1009: diff --git a/wfdb/io/convert/wav.py b/wfdb/io/convert/wav.py index eb616ef3..a0091dc1 100644 --- a/wfdb/io/convert/wav.py +++ b/wfdb/io/convert/wav.py @@ -4,9 +4,7 @@ import numpy as np -from wfdb.io import Record -from wfdb.io import download -from wfdb.io import _url +from wfdb.io import Record, _url, download from wfdb.io.record import rdrecord @@ -128,15 +126,9 @@ def wfdb_to_wav( if write_header: record.adc_zero = record.n_sig * [0] record.adc_res = record.n_sig * [16] - record.adc_gain = [ - (r * (1 << shift[i])) for i, r in enumerate(record.adc_gain) - ] - record.baseline = [ - (b - offset[i]) for i, b in enumerate(record.baseline) - ] - record.baseline = [ - (b * (1 << shift[i])) for i, b in enumerate(record.baseline) - ] + record.adc_gain = [(r * (1 << shift[i])) for i, r in enumerate(record.adc_gain)] + record.baseline = [(b - offset[i]) for i, b in enumerate(record.baseline)] + record.baseline = [(b * (1 << shift[i])) for i, b in enumerate(record.baseline)] record.file_name = record.n_sig * [record_name_out + ".wav"] record.block_size = record.n_sig * [0] record.fmt = record.n_sig * ["16"] @@ -242,24 +234,20 @@ def read_wav(record_name, pn_dir=None, delete_file=True, record_only=False): record_name.split(os.sep)[-1].replace("-", "_").replace(".wav", "") ) - chunk_ID = "".join( - [s.decode() for s in struct.unpack(">4s", wave_file.read(4))] - ) + chunk_ID = "".join([s.decode() for s in struct.unpack(">4s", wave_file.read(4))]) if chunk_ID != "RIFF": - raise Exception("{} is not a .wav-format file".format(record_name)) + raise Exception(f"{record_name} is not a .wav-format file") correct_chunk_size = os.path.getsize(record_name) - 8 chunk_size = struct.unpack("4s", wave_file.read(4))[0].decode() if fmt != "WAVE": - raise Exception("{} is not a .wav-format file".format(record_name)) + raise Exception(f"{record_name} is not a .wav-format file") subchunk1_ID = struct.unpack(">4s", wave_file.read(4))[0].decode() if subchunk1_ID != "fmt ": @@ -268,10 +256,10 @@ def read_wav(record_name, pn_dir=None, delete_file=True, record_only=False): subchunk1_size = struct.unpack(" 1: - print("PCM has compression of {}".format(audio_format)) + print(f"PCM has compression of {audio_format}") if (subchunk1_size != 16) or (audio_format != 1): - raise Exception("Unsupported format {}".format(audio_format)) + raise Exception(f"Unsupported format {audio_format}") num_channels = struct.unpack("4s", wave_file.read(4))[0].decode() if subchunk2_ID != "data": @@ -315,9 +299,7 @@ def read_wav(record_name, pn_dir=None, delete_file=True, record_only=False): subchunk2_size = struct.unpack(" datetime.time: def parse_header_content( header_content: str, -) -> Tuple[List[str], List[str]]: +) -> tuple[list[str], list[str]]: """ Parse the text of a header file. diff --git a/wfdb/io/record.py b/wfdb/io/record.py index 6cb5ef0b..d232bcea 100644 --- a/wfdb/io/record.py +++ b/wfdb/io/record.py @@ -1,19 +1,14 @@ import datetime import multiprocessing.dummy -import posixpath import os +import posixpath import re import fsspec import numpy as np import pandas as pd -from wfdb.io import _header -from wfdb.io import _signal -from wfdb.io import _url -from wfdb.io import download -from wfdb.io import header -from wfdb.io import util +from wfdb.io import _header, _signal, _url, download, header, util from wfdb.io._coreio import CLOUD_PROTOCOLS # -------------- WFDB Signal Calibration and Classification ---------- # @@ -157,7 +152,7 @@ } -class BaseRecord(object): +class BaseRecord: """ The base WFDB class extended by the Record and MultiRecord classes. @@ -222,13 +217,9 @@ def __init__( self.sig_len = sig_len if base_datetime is not None: if base_time is not None: - raise TypeError( - "cannot specify both base_time and base_datetime" - ) + raise TypeError("cannot specify both base_time and base_datetime") if base_date is not None: - raise TypeError( - "cannot specify both base_date and base_datetime" - ) + raise TypeError("cannot specify both base_date and base_datetime") self.base_datetime = base_datetime else: self.base_time = base_time @@ -241,9 +232,7 @@ def base_datetime(self): if self.base_date is None or self.base_time is None: return None else: - return datetime.datetime.combine( - date=self.base_date, time=self.base_time - ) + return datetime.datetime.combine(date=self.base_date, time=self.base_time) @base_datetime.setter def base_datetime(self, value): @@ -405,9 +394,9 @@ def check_field(self, field, required_channels="all"): item=item, field_name=field, ndim=2, - parent_class=( - lambda f: np.integer if f == "d_signal" else np.floating - )(field), + parent_class=(lambda f: np.integer if f == "d_signal" else np.floating)( + field + ), ) elif field in ["e_d_signal", "e_p_signal"]: for ch in range(len(item)): @@ -416,9 +405,7 @@ def check_field(self, field, required_channels="all"): field_name=field, ndim=1, parent_class=( - lambda f: ( - np.integer if f == "e_d_signal" else np.floating - ) + lambda f: np.integer if f == "e_d_signal" else np.floating )(field), channel_num=ch, ) @@ -427,10 +414,7 @@ def check_field(self, field, required_channels="all"): elif field == "record_name": # Allow letters, digits, hyphens, and underscores. accepted_string = re.match(r"[-\w]+", self.record_name) - if ( - not accepted_string - or accepted_string.string != self.record_name - ): + if not accepted_string or accepted_string.string != self.record_name: raise ValueError( "record_name must only comprise of letters, digits, hyphens, and underscores." ) @@ -467,10 +451,7 @@ def check_field(self, field, required_channels="all"): if field == "file_name": # Check for file_name characters accepted_string = re.match(r"[-\w]+\.?[\w]+", item[ch]) - if ( - not accepted_string - or accepted_string.string != item[ch] - ): + if not accepted_string or accepted_string.string != item[ch]: raise ValueError( "File names should only contain alphanumerics, hyphens, and an extension. eg. record-100.dat" ) @@ -492,9 +473,7 @@ def check_field(self, field, required_channels="all"): ) elif field == "skew": if item[ch] < 0: - raise ValueError( - "skew values must be non-negative integers" - ) + raise ValueError("skew values must be non-negative integers") elif field == "byte_offset": if item[ch] < 0: raise ValueError( @@ -511,14 +490,10 @@ def check_field(self, field, required_channels="all"): ) elif field == "units": if re.search(r"\s", item[ch]): - raise ValueError( - "units strings may not contain whitespaces." - ) + raise ValueError("units strings may not contain whitespaces.") elif field == "adc_res": if item[ch] < 0: - raise ValueError( - "adc_res values must be non-negative integers" - ) + raise ValueError("adc_res values must be non-negative integers") elif field == "block_size": if item[ch] < 0: raise ValueError( @@ -527,13 +502,11 @@ def check_field(self, field, required_channels="all"): elif field == "sig_name": if item[ch][:1].isspace() or item[ch][-1:].isspace(): raise ValueError( - "sig_name strings may not begin or end with " - "whitespace." + "sig_name strings may not begin or end with whitespace." ) if re.search(r"[\x00-\x1f\x7f-\x9f]", item[ch]): raise ValueError( - "sig_name strings may not contain " - "control characters." + "sig_name strings may not contain control characters." ) if len(set(item)) != len(item): raise ValueError("sig_name strings must be unique.") @@ -546,10 +519,7 @@ def check_field(self, field, required_channels="all"): if item[ch] == "~": continue accepted_string = re.match(r"[-\w]+", item[ch]) - if ( - not accepted_string - or accepted_string.string != item[ch] - ): + if not accepted_string or accepted_string.string != item[ch]: raise ValueError( "Non-null segment names may only contain alphanumerics and dashes. Null segment names must be set to '~'" ) @@ -635,18 +605,14 @@ def check_read_inputs( # Channel Ranges if len(channels): if min(channels) < 0: - raise ValueError( - "Input channels must all be non-negative integers" - ) + raise ValueError("Input channels must all be non-negative integers") if max(channels) > self.n_sig - 1: raise ValueError( "Input channels must all be lower than the total number of channels" ) if return_res not in [64, 32, 16, 8]: - raise ValueError( - "return_res must be one of the following: 64, 32, 16, 8" - ) + raise ValueError("return_res must be one of the following: 64, 32, 16, 8") if physical and return_res == 8: raise ValueError( "return_res must be one of the following when physical is True: 64, 32, 16" @@ -823,7 +789,7 @@ def __init__( # Note the lack of the 'n_seg' field. Single segment records cannot # have this field. Even n_seg = 1 makes the header a multi-segment # header. - super(Record, self).__init__( + super().__init__( record_name=record_name, n_sig=n_sig, fs=fs, @@ -974,16 +940,12 @@ def _arrange_fields(self, channels, sampfrom, smooth_frames): if not smooth_frames: # Checksum and init_value to be updated if present # unless the whole signal length was input - if self.sig_len != int( - len(self.e_d_signal[0]) / self.samps_per_frame[0] - ): + if self.sig_len != int(len(self.e_d_signal[0]) / self.samps_per_frame[0]): self.checksum = self.calc_checksum(True) self.init_value = [s[0] for s in self.e_d_signal] self.n_sig = len(channels) - self.sig_len = int( - len(self.e_d_signal[0]) / self.samps_per_frame[0] - ) + self.sig_len = int(len(self.e_d_signal[0]) / self.samps_per_frame[0]) # MxN numpy array d_signal else: @@ -1144,7 +1106,7 @@ def __init__( sig_name=None, sig_segments=None, ): - super(MultiRecord, self).__init__( + super().__init__( record_name=record_name, n_sig=n_sig, fs=fs, @@ -1271,9 +1233,7 @@ def _required_segments(self, sampfrom, sampto): if sampto == cumsumlengths[len(cumsumlengths) - 1]: seg_numbers.append(len(cumsumlengths) - 1) else: - seg_numbers.append( - [sampto <= cs for cs in cumsumlengths].index(True) - ) + seg_numbers.append([sampto <= cs for cs in cumsumlengths].index(True)) # Add 1 for variable layout records seg_numbers = list(np.add(seg_numbers, startseg)) @@ -1420,9 +1380,7 @@ def _arrange_fields( # the signal was read. if not force_channels: # The desired signal names. - desired_sig_names = [ - self.segments[0].sig_name[ch] for ch in channels - ] + desired_sig_names = [self.segments[0].sig_name[ch] for ch in channels] # Actual contained signal names of individual segments # contained_sig_names = [seg.sig_name for seg in self.segments[1:]] contained_sig_names = set( @@ -1435,14 +1393,10 @@ def _arrange_fields( ) # Remove non-present names. Keep the order. sig_name = [ - name - for name in desired_sig_names - if name in contained_sig_names + name for name in desired_sig_names if name in contained_sig_names ] # Channel indices to keep for signal specification fields - channels = [ - self.segments[0].sig_name.index(name) for name in sig_name - ] + channels = [self.segments[0].sig_name.index(name) for name in sig_name] # Rearrange signal specification fields for field in _header.SIGNAL_SPECS.index: @@ -1928,9 +1882,7 @@ def rdheader(record_name, pn_dir=None, rd_segments=False): if s == "~": record.segments.append(None) else: - record.segments.append( - rdheader(os.path.join(dir_name, s), pn_dir) - ) + record.segments.append(rdheader(os.path.join(dir_name, s), pn_dir)) # Fill in the sig_name attribute record.sig_name = record.get_sig_name() # Fill in the sig_segments attribute @@ -2449,7 +2401,7 @@ def sampfreq(record_name, pn_dir=None): sig_name = record.sig_name for sig, samp in zip(sig_name, samps_per_frame): - print("{}\t{}".format(sig, samp)) + print(f"{sig}\t{samp}") def signame(record_name, pn_dir=None, sig_nums=[]): @@ -2507,7 +2459,7 @@ def signame(record_name, pn_dir=None, sig_nums=[]): try: print(record.sig_name[n]) except IndexError: - raise Exception("sig_nums value {} out of range".format(n)) + raise Exception(f"sig_nums value {n} out of range") else: print(*record.sig_name, sep="\n") @@ -2614,7 +2566,7 @@ def wfdbdesc(record_name, pn_dir=None): print(f"Length: {record_length} ({record.sig_len} sample intervals)") print(f"Sampling frequency: {record.fs} Hz") - print(f'{record.n_sig} signal{"" if record.n_sig==1 else "s"}') + print(f"{record.n_sig} signal{'' if record.n_sig == 1 else 's'}") if type(record) is not MultiRecord: for i in range(record.n_sig): @@ -2722,9 +2674,7 @@ def wfdbtime(record_name, input_times, pn_dir=None): record = rdheader(record_name, pn_dir=pn_dir) try: - start_time = datetime.datetime.combine( - record.base_date, record.base_time - ) + start_time = datetime.datetime.combine(record.base_date, record.base_time) except TypeError: try: start_time = record.base_time @@ -2747,9 +2697,7 @@ def wfdbtime(record_name, input_times, pn_dir=None): new_times = float(sample_num / record.fs) times_split = [f"{new_times}"] elif "/" in times: - out_date = datetime.datetime.strptime( - times, "%H:%M:%S.%f %d/%m/%Y" - ) + out_date = datetime.datetime.strptime(times, "%H:%M:%S.%f %d/%m/%Y") try: start_time = datetime.datetime.combine( datetime.date.min, start_time @@ -2759,17 +2707,13 @@ def wfdbtime(record_name, input_times, pn_dir=None): try: elapsed_time = out_date - start_time except TypeError: - raise Exception( - "No start date or time provided in the record." - ) + raise Exception("No start date or time provided in the record.") total_seconds = elapsed_time.total_seconds() - sample_num = "s" + str( - int(elapsed_time.total_seconds() * record.fs) - ) + sample_num = "s" + str(int(elapsed_time.total_seconds() * record.fs)) hours = int(total_seconds // 60 // 60) minutes = int(total_seconds // 60) out_time = f"{hours:02}:{minutes:02}:{total_seconds:06.3f}" - out_date = f'[{out_date.strftime("%H:%M:%S.%f %d/%m/%Y")}]' + out_date = f"[{out_date.strftime('%H:%M:%S.%f %d/%m/%Y')}]" print(f"{sample_num:>12}{out_time:>24}{out_date:>32}") break else: @@ -2792,9 +2736,7 @@ def wfdbtime(record_name, input_times, pn_dir=None): if minutes >= 60: raise Exception("Minutes not in correct format") out_time = f"{hours:02}:{minutes:02}:{seconds:06.3f}" - out_date = _get_date_from_time( - start_time, hours, minutes, seconds, out_time - ) + out_date = _get_date_from_time(start_time, hours, minutes, seconds, out_time) if not times.startswith("s"): sample_num = int( sum(x * 60**i for i, x in enumerate([seconds, minutes, hours])) @@ -2843,7 +2785,9 @@ def _get_date_from_time(start_time, hours, minutes, seconds, out_time): microseconds=microseconds, ) out_date = start_time + elapsed_time - out_date = f'[{(datetime.datetime.min+out_date).strftime("%H:%M:%S.%f %d/%m/%Y")}]' + out_date = ( + f"[{(datetime.datetime.min + out_date).strftime('%H:%M:%S.%f %d/%m/%Y')}]" + ) else: out_date = out_time return out_date @@ -2878,9 +2822,7 @@ def _get_wanted_channels(wanted_sig_names, record_sig_names, pad=False): ] else: return [ - record_sig_names.index(s) - for s in wanted_sig_names - if s in record_sig_names + record_sig_names.index(s) for s in wanted_sig_names if s in record_sig_names ] @@ -3003,9 +2945,7 @@ def wrsamp( raise Exception( "When using d_signal or e_d_signal, must also specify 'fmt', 'gain', and 'baseline' fields" ) - if ( - e_p_signal is not None or e_d_signal is not None - ) and samps_per_frame is None: + if (e_p_signal is not None or e_d_signal is not None) and samps_per_frame is None: raise Exception( "When passing e_p_signal or e_d_signal, you also need to specify samples per frame for each channel" ) @@ -3022,9 +2962,9 @@ def wrsamp( ) if isinstance(first_valid_signal, np.ndarray): num_sig_channels = first_valid_signal.shape[1] - channel_samples = [ - first_valid_signal.shape[0] - ] * first_valid_signal.shape[1] + channel_samples = [first_valid_signal.shape[0]] * first_valid_signal.shape[ + 1 + ] elif isinstance(first_valid_signal, list): num_sig_channels = len(first_valid_signal) channel_samples = [len(channel) for channel in first_valid_signal] @@ -3187,9 +3127,7 @@ def dl_database( # If MIT format, have to figure out all associated files all_files.append(rec + ".hea") dir_name, base_rec_name = os.path.split(rec) - record = rdheader( - base_rec_name, pn_dir=posixpath.join(db_dir, dir_name) - ) + record = rdheader(base_rec_name, pn_dir=posixpath.join(db_dir, dir_name)) # Single segment record if isinstance(record, Record): @@ -3217,9 +3155,7 @@ def dl_database( if annotators is not None: for a in annotators: ann_file = rec + "." + a - url = posixpath.join( - download.config.db_index_url, db_dir, ann_file - ) + url = posixpath.join(download.config.db_index_url, db_dir, ann_file) try: _url.openurl(url, check_access=True) all_files.append(ann_file) diff --git a/wfdb/io/util.py b/wfdb/io/util.py index d0c34b8f..d45cbc49 100644 --- a/wfdb/io/util.py +++ b/wfdb/io/util.py @@ -5,11 +5,10 @@ import io import math import os +from collections.abc import Sequence import numpy as np -from typing import List, Sequence, Tuple - def lines_to_file(file_name: str, write_dir: str, lines: Sequence[str]): """ @@ -106,9 +105,9 @@ def upround(x, base): def overlapping_ranges( - ranges_1: Sequence[Tuple[int, int]], - ranges_2: Sequence[Tuple[int, int]], -) -> List[Tuple[int, int]]: + ranges_1: Sequence[tuple[int, int]], + ranges_2: Sequence[tuple[int, int]], +) -> list[tuple[int, int]]: """ Given two collections of integer ranges, return a list of ranges in which both input inputs overlap. diff --git a/wfdb/plot/__init__.py b/wfdb/plot/__init__.py index bf801834..fb78fb36 100644 --- a/wfdb/plot/__init__.py +++ b/wfdb/plot/__init__.py @@ -2,4 +2,4 @@ The plot subpackage contains tools for plotting signals and annotations. """ -from wfdb.plot.plot import plot_items, plot_wfdb, plot_all_records +from wfdb.plot.plot import plot_all_records, plot_items, plot_wfdb diff --git a/wfdb/plot/plot.py b/wfdb/plot/plot.py index ad01cab5..e7853e17 100644 --- a/wfdb/plot/plot.py +++ b/wfdb/plot/plot.py @@ -2,9 +2,8 @@ import numpy as np -from wfdb.io.record import Record, rdrecord +from wfdb.io.record import rdrecord from wfdb.io.util import downround, upround -from wfdb.io.annotation import Annotation def _expand_channels(signal): @@ -34,16 +33,12 @@ def _expand_channels(signal): elif signal.ndim == 2: return list(signal.transpose()) else: - raise ValueError( - "invalid shape for signal array: {}".format(signal.shape) - ) + raise ValueError(f"invalid shape for signal array: {signal.shape}") else: signal = list(signal) if any(s.ndim != 1 for s in signal): raise ValueError( - "invalid shape for signal array(s): {}".format( - [s.shape for s in signal] - ) + f"invalid shape for signal array(s): {[s.shape for s in signal]}" ) return signal @@ -74,8 +69,8 @@ def _get_sampling_freq(sampling_freq, n_sig, frame_freq): elif hasattr(sampling_freq, "__len__"): if len(sampling_freq) != n_sig: raise ValueError( - "length mismatch: n_sig = {}, " - "len(sampling_freq) = {}".format(n_sig, len(sampling_freq)) + f"length mismatch: n_sig = {n_sig}, " + f"len(sampling_freq) = {len(sampling_freq)}" ) return list(sampling_freq) else: @@ -108,8 +103,7 @@ def _get_ann_freq(ann_freq, n_annot, frame_freq): elif hasattr(ann_freq, "__len__"): if len(ann_freq) != n_annot: raise ValueError( - "length mismatch: n_annot = {}, " - "len(ann_freq) = {}".format(n_annot, len(ann_freq)) + f"length mismatch: n_annot = {n_annot}, len(ann_freq) = {len(ann_freq)}" ) return list(ann_freq) else: @@ -301,14 +295,14 @@ def plot_items( if len(xlabel) != signal.shape[1]: raise Exception( "The length of the xlabel must be the same as the " - "signal: {} values".format(signal.shape[1]) + f"signal: {signal.shape[1]} values" ) if ylabel: if len(ylabel) != n_subplots: raise Exception( "The length of the ylabel must be the same as the " - "signal: {} values".format(n_subplots) + f"signal: {n_subplots} values" ) _label_figure( @@ -609,20 +603,15 @@ def _plot_annotation( y = np.zeros(len(ann_samp[ch])) except IndexError: raise Exception( - "IndexError: try setting shift_samps=True in " - 'the "rdann" function?' + 'IndexError: try setting shift_samps=True in the "rdann" function?' ) - axes[ch].plot( - ann_samp[ch] / downsample_factor, y, ann_style[ch], zorder=4 - ) + axes[ch].plot(ann_samp[ch] / downsample_factor, y, ann_style[ch], zorder=4) # Plot the annotation symbols if any if ann_sym is not None and ann_sym[ch] is not None: for i, s in enumerate(ann_sym[ch]): - axes[ch].annotate( - s, (ann_samp[ch][i] / downsample_factor, y[i]) - ) + axes[ch].annotate(s, (ann_samp[ch][i] / downsample_factor, y[i])) def _plot_ecg_grids(ecg_grids, fs, units, time_units, axes, sampling_freq=None): @@ -844,9 +833,7 @@ def _label_figure( # put placeholders n_missing_labels = n_subplots - len(ylabel) if n_missing_labels: - ylabel = ylabel + [ - "ch_%d/NU" % i for i in range(len(ylabel), n_subplots) - ] + ylabel = ylabel + ["ch_%d/NU" % i for i in range(len(ylabel), n_subplots)] for ch in range(n_subplots): axes[ch].set_ylabel(ylabel[ch]) @@ -949,9 +936,7 @@ def plot_wfdb( ylabel, record_name, sig_units, - ) = _get_wfdb_plot_items( - record=record, annotation=annotation, plot_sym=plot_sym - ) + ) = _get_wfdb_plot_items(record=record, annotation=annotation, plot_sym=plot_sym) if record: if record.e_p_signal is not None or record.e_d_signal is not None: @@ -1184,9 +1169,7 @@ def plot_all_records(directory=""): directory = directory or os.getcwd() headers = [ - f - for f in os.listdir(directory) - if os.path.isfile(os.path.join(directory, f)) + f for f in os.listdir(directory) if os.path.isfile(os.path.join(directory, f)) ] headers = [f for f in headers if f.endswith(".hea")] diff --git a/wfdb/processing/__init__.py b/wfdb/processing/__init__.py index 295aa89e..2aadc09d 100644 --- a/wfdb/processing/__init__.py +++ b/wfdb/processing/__init__.py @@ -1,17 +1,17 @@ from wfdb.processing.basic import ( + get_filter_gain, + normalize_bound, resample_ann, + resample_multichan, resample_sig, resample_singlechan, - resample_multichan, - normalize_bound, - get_filter_gain, ) from wfdb.processing.evaluate import ( Comparitor, - compare_annotations, benchmark_mitdb, + compare_annotations, ) -from wfdb.processing.hr import compute_hr, calc_rr, calc_mean_hr, ann2rr, rr2ann -from wfdb.processing.peaks import find_peaks, find_local_peaks, correct_peaks -from wfdb.processing.qrs import XQRS, xqrs_detect, gqrs_detect from wfdb.processing.filter import sigavg +from wfdb.processing.hr import ann2rr, calc_mean_hr, calc_rr, compute_hr, rr2ann +from wfdb.processing.peaks import correct_peaks, find_local_peaks, find_peaks +from wfdb.processing.qrs import XQRS, gqrs_detect, xqrs_detect diff --git a/wfdb/processing/basic.py b/wfdb/processing/basic.py index 2cad9a4a..0390b754 100644 --- a/wfdb/processing/basic.py +++ b/wfdb/processing/basic.py @@ -59,10 +59,7 @@ def resample_sig(x, fs, fs_target): if np.isnan(x).any(): x = pd.Series(x.reshape((-1,))).interpolate().values resampled_x, resampled_t = signal.resample(x, num=new_length, t=t) - assert ( - resampled_x.shape == resampled_t.shape - and resampled_x.shape[0] == new_length - ) + assert resampled_x.shape == resampled_t.shape and resampled_x.shape[0] == new_length assert np.all(np.diff(resampled_t) > 0) return resampled_x, resampled_t diff --git a/wfdb/processing/evaluate.py b/wfdb/processing/evaluate.py index 3e960286..7d49457c 100644 --- a/wfdb/processing/evaluate.py +++ b/wfdb/processing/evaluate.py @@ -1,4 +1,4 @@ -from multiprocessing import cpu_count, Pool +from multiprocessing import Pool, cpu_count import numpy as np @@ -7,7 +7,7 @@ from wfdb.io.record import rdsamp -class Comparitor(object): +class Comparitor: """ The class to implement and hold comparisons between two sets of annotations. See methods `compare`, `print_summary` and `plot`. @@ -48,10 +48,7 @@ def __init__(self, ref_sample, test_sample, window_width, signal=None): if len(ref_sample) > 1 and len(test_sample) > 1: if min(np.diff(ref_sample)) < 0 or min(np.diff(test_sample)) < 0: raise ValueError( - ( - "The sample locations must be monotonically" - + " increasing" - ) + "The sample locations must be monotonically" + " increasing" ) self.ref_sample = ref_sample @@ -204,9 +201,7 @@ def compare(self): ) # Assign the reference-test pair if close enough if smallest_samp_diff < self.window_width: - self.matching_sample_nums[ref_samp_num] = ( - closest_samp_num - ) + self.matching_sample_nums[ref_samp_num] = closest_samp_num # Set the starting test sample number to inspect # for the next reference sample. test_samp_num = closest_samp_num + 1 @@ -302,10 +297,7 @@ def print_summary(self): print("False Positives (unmatched test samples): %d" % self.fp) print("False Negatives (unmatched reference samples): %d\n" % self.fn) - print( - "Sensitivity: %.4f (%d/%d)" - % (self.sensitivity, self.tp, self.n_ref) - ) + print("Sensitivity: %.4f (%d/%d)" % (self.sensitivity, self.tp, self.n_ref)) print( "Positive Predictivity: %.4f (%d/%d)" % (self.positive_predictivity, self.tp, self.n_test) @@ -510,9 +502,7 @@ def benchmark_mitdb(detector, verbose=False, print_results=False): # Calculate aggregate stats sensitivity = np.mean([c.sensitivity for c in comparitors]) - positive_predictivity = np.mean( - [c.positive_predictivity for c in comparitors] - ) + positive_predictivity = np.mean([c.positive_predictivity for c in comparitors]) comparitors = dict(zip(record_list, comparitors)) diff --git a/wfdb/processing/filter.py b/wfdb/processing/filter.py index f8739b73..5b108987 100644 --- a/wfdb/processing/filter.py +++ b/wfdb/processing/filter.py @@ -3,8 +3,7 @@ import numpy as np import pandas as pd -from wfdb.io import annotation -from wfdb.io import download +from wfdb.io import annotation, download from wfdb.io.record import rdrecord diff --git a/wfdb/processing/hr.py b/wfdb/processing/hr.py index 5e6bfc36..b2ea28e0 100644 --- a/wfdb/processing/hr.py +++ b/wfdb/processing/hr.py @@ -2,8 +2,8 @@ import numpy as np -from wfdb.io.annotation import rdann, wrann from wfdb.io import download +from wfdb.io.annotation import rdann, wrann def compute_hr(sig_len, qrs_inds, fs): diff --git a/wfdb/processing/qrs.py b/wfdb/processing/qrs.py index 3460ec13..da0e3a1a 100644 --- a/wfdb/processing/qrs.py +++ b/wfdb/processing/qrs.py @@ -3,12 +3,12 @@ import numpy as np from scipy import signal +from wfdb.io.record import Record from wfdb.processing.basic import get_filter_gain, normalize from wfdb.processing.peaks import find_local_peaks -from wfdb.io.record import Record -class XQRS(object): +class XQRS: """ The QRS detector class for the XQRS algorithm. The `XQRS.Conf` class is the configuration class that stores initial parameters @@ -75,7 +75,7 @@ def __init__(self, sig, fs, conf=None): self.conf = conf or XQRS.Conf() self._set_conf() - class Conf(object): + class Conf: """ Initial signal configuration object for this QRS detector. @@ -217,9 +217,7 @@ def _mwi(self): """ wavelet_filter = ricker(self.qrs_width, 4) - self.sig_i = ( - signal.filtfilt(wavelet_filter, [1], self.sig_f, axis=0) ** 2 - ) + self.sig_i = signal.filtfilt(wavelet_filter, [1], self.sig_f, axis=0) ** 2 # Save the MWI gain (x2 due to double filtering) and the total # gain from raw to MWI @@ -289,9 +287,7 @@ def _learn_init_params(self, n_calib_beats=8): # Skip if no peaks in range if not peak_inds_f.size or not peak_nums_r.size or not peak_nums_l.size: if self.verbose: - print( - "Failed to find %d beats during learning." % n_calib_beats - ) + print("Failed to find %d beats during learning." % n_calib_beats) self._set_default_init_params() return @@ -362,9 +358,7 @@ def _learn_init_params(self, n_calib_beats=8): # Failed to find enough calibration beats. Use default values. else: if self.verbose: - print( - "Failed to find %d beats during learning." % n_calib_beats - ) + print("Failed to find %d beats during learning." % n_calib_beats) self._set_default_init_params() @@ -513,13 +507,9 @@ def _update_qrs(self, peak_num, backsearch=False): # was found via backsearch if backsearch: self.backsearch_qrs_inds.append(i) - self.qrs_amp_recent = ( - 0.75 * self.qrs_amp_recent + 0.25 * self.sig_i[i] - ) + self.qrs_amp_recent = 0.75 * self.qrs_amp_recent + 0.25 * self.sig_i[i] else: - self.qrs_amp_recent = ( - 0.875 * self.qrs_amp_recent + 0.125 * self.sig_i[i] - ) + self.qrs_amp_recent = 0.875 * self.qrs_amp_recent + 0.125 * self.sig_i[i] self.qrs_thr = max( (0.25 * self.qrs_amp_recent + 0.75 * self.noise_amp_recent), @@ -582,9 +572,7 @@ def _update_noise(self, peak_num): """ i = self.peak_inds_i[peak_num] - self.noise_amp_recent = ( - 0.875 * self.noise_amp_recent + 0.125 * self.sig_i[i] - ) + self.noise_amp_recent = 0.875 * self.noise_amp_recent + 0.125 * self.sig_i[i] return def _require_backsearch(self): @@ -628,9 +616,7 @@ def _backsearch(self): """ if self.last_qrs_peak_num is not None: - for peak_num in range( - self.last_qrs_peak_num + 1, self.peak_num + 1 - ): + for peak_num in range(self.last_qrs_peak_num + 1, self.peak_num + 1): if self._is_qrs(peak_num=peak_num, backsearch=True): self._update_qrs(peak_num=peak_num, backsearch=True) # No need to update noise parameters if it was classified as @@ -740,9 +726,7 @@ def detect(self, sampfrom=0, sampto="end", learn=True, verbose=True): self._run_detection() -def xqrs_detect( - sig, fs, sampfrom=0, sampto="end", conf=None, learn=True, verbose=True -): +def xqrs_detect(sig, fs, sampfrom=0, sampto="end", conf=None, learn=True, verbose=True): """ Run the 'xqrs' QRS detection algorithm on a signal. See the docstring of the XQRS class for algorithm details. @@ -809,7 +793,7 @@ def time_to_sample_number(seconds, frequency): return seconds * frequency + 0.5 -class GQRS(object): +class GQRS: """ GQRS detection class. @@ -819,7 +803,7 @@ class GQRS(object): """ - class Conf(object): + class Conf: """ Initial signal configuration object for this QRS detector. @@ -909,9 +893,7 @@ def __init__( self.dt = int(self.QS * self.sps / 4) if self.dt < 1: - raise Exception( - "Sampling rate is too low. Unable to use signal." - ) + raise Exception("Sampling rate is too low. Unable to use signal.") self.rtmin = int(self.RTmin * self.sps) self.rtmean = int(0.75 * self.QT * self.sps) @@ -935,7 +917,7 @@ def __init__( self.smt = 0 self.smt0 = 0 + self.smdt - class Peak(object): + class Peak: """ Holds all of the peak information for the QRS object. @@ -957,7 +939,7 @@ def __init__(self, peak_time, peak_amp, peak_type): self.next_peak = None self.prev_peak = None - class Annotation(object): + class Annotation: """ Holds all of the annotation information for the QRS object. @@ -1265,18 +1247,12 @@ def qf(self): # needed below are in the buffer dv2 = self.sm(self.t + self.c.dt4) dv2 -= self.smv_at(self.t - self.c.dt4) - dv1 = int( - self.smv_at(self.t + self.c.dt) - self.smv_at(self.t - self.c.dt) - ) + dv1 = int(self.smv_at(self.t + self.c.dt) - self.smv_at(self.t - self.c.dt)) dv = dv1 << 1 - dv -= int( - self.smv_at(self.t + self.c.dt2) - self.smv_at(self.t - self.c.dt2) - ) + dv -= int(self.smv_at(self.t + self.c.dt2) - self.smv_at(self.t - self.c.dt2)) dv = dv << 1 dv += dv1 - dv -= int( - self.smv_at(self.t + self.c.dt3) - self.smv_at(self.t - self.c.dt3) - ) + dv -= int(self.smv_at(self.t + self.c.dt3) - self.smv_at(self.t - self.c.dt3)) dv = dv << 1 dv += dv2 self.v1 += dv @@ -1462,20 +1438,12 @@ def find_missing(r, p): q1 = self.qfv_at(self.t - 1) q2 = self.qfv_at(self.t - 2) # state == RUNNING only - if ( - q1 > self.c.pthr - and q2 < q1 - and q1 >= q0 - and self.t > self.c.dt4 - ): + if q1 > self.c.pthr and q2 < q1 and q1 >= q0 and self.t > self.c.dt4: add_peak(self.t - 1, q1, 0) last_peak = self.t - 1 p = self.current_peak.next_peak while p.time < self.t - self.c.rtmax: - if ( - p.time >= self.annot.time + self.c.rrmin - and peaktype(p) == 1 - ): + if p.time >= self.annot.time + self.c.rrmin and peaktype(p) == 1: if p.amp > self.c.qthr: rr = p.time - self.annot.time q = find_missing(r, p) @@ -1542,8 +1510,7 @@ def find_missing(r, p): "TWAVE", ( 1 - if tmp_time - > self.annot.time + self.c.rtmean + if tmp_time > self.annot.time + self.c.rtmean else 0 ), rtdmin, @@ -1567,10 +1534,7 @@ def find_missing(r, p): self.c.qthr -= self.c.qthr >> 4 # end: p = p.next_peak - elif ( - self.t - last_peak > self.c.rrmax - and self.c.pthr > self.c.pthmin - ): + elif self.t - last_peak > self.c.rrmax and self.c.pthr > self.c.pthmin: self.c.pthr -= self.c.pthr >> 4 self.t += 1