linting

Author: oerc0042

isatools/isatab/load/core.py

@@ -1,13 +1,13 @@
 from __future__ import annotations
 
-from abc import ABCMeta, abstractmethod
-from glob import glob
-from io import StringIO
-from itertools import zip_longest
-from os import path
-from re import compile
-from functools import lru_cache
-from typing import TextIO
+from abc import ABCMeta, abstractmethod
+from functools import lru_cache
+from glob import glob
+from io import StringIO
+from itertools import zip_longest
+from os import path
+from re import compile
+from typing import TextIO
 
 from numpy import nan
 from pandas import DataFrame, Series, merge, read_csv
@@ -35,60 +35,60 @@
 from .mapping import get_investigation_base_output, investigation_sections_mapping, study_sections_mapping
 
 
-class ISATabReader:
+class ISATabReader:
     """A class to read an ISA-Tab investigation file into a dictionary of DataFrames
 
     :param fp: A file-like buffer object of the investigation file
     """
 
-    def __init__(self, fp: TextIO) -> None:
-        """Constructor for the ISATabReader class"""
-        self.source_file: TextIO = fp
-        self.__next_line: str | None = None
-        self.dataframe_dict: dict[str, DataFrame | str, list[DataFrame]] = {}
-
-    @property
-    def source_file(self) -> TextIO:
-        """Getter for the source file-like buffer object
-
-        :return: A file-like buffer object
-        """
-        return self.__source_file
-
-    @source_file.setter
-    def source_file(self, fp: TextIO) -> None:
-        """Setter for the source file property.
-
-        :param fp: A file-like buffer object
-        """
-        self.__source_file = fp
-
-    def __read_next_data_line(self) -> str:
-        """Read the next non-comment line from the source file."""
-        while True:
-            line = self.source_file.readline()
-            if not line:
-                return ""
-            if not line.lstrip().startswith("#"):
-                return line
-
-    def __peek(self) -> str:
-        """Peek at the next line without moving to the next line. This function get the position of the next line,
-        reads the next line, then resets the file pointer to the original position
-
-        :return: The next line past the current line
-        """
-        if self.__next_line is None:
-            self.__next_line = self.__read_next_data_line()
-        return self.__next_line
-
-    def __readline(self) -> str:
-        """Read the next line, honoring the peek buffer."""
-        if self.__next_line is not None:
-            line = self.__next_line
-            self.__next_line = None
-            return line
-        return self.__read_next_data_line()
+    def __init__(self, fp: TextIO) -> None:
+        """Constructor for the ISATabReader class"""
+        self.source_file: TextIO = fp
+        self.__next_line: str | None = None
+        self.dataframe_dict: dict[str, DataFrame | str, list[DataFrame]] = {}
+
+    @property
+    def source_file(self) -> TextIO:
+        """Getter for the source file-like buffer object
+
+        :return: A file-like buffer object
+        """
+        return self.__source_file
+
+    @source_file.setter
+    def source_file(self, fp: TextIO) -> None:
+        """Setter for the source file property.
+
+        :param fp: A file-like buffer object
+        """
+        self.__source_file = fp
+
+    def __read_next_data_line(self) -> str:
+        """Read the next non-comment line from the source file."""
+        while True:
+            line = self.source_file.readline()
+            if not line:
+                return ""
+            if not line.lstrip().startswith("#"):
+                return line
+
+    def __peek(self) -> str:
+        """Peek at the next line without moving to the next line. This function get the position of the next line,
+        reads the next line, then resets the file pointer to the original position
+
+        :return: The next line past the current line
+        """
+        if self.__next_line is None:
+            self.__next_line = self.__read_next_data_line()
+        return self.__next_line
+
+    def __readline(self) -> str:
+        """Read the next line, honoring the peek buffer."""
+        if self.__next_line is not None:
+            line = self.__next_line
+            self.__next_line = None
+            return line
+        return self.__read_next_data_line()
 
     def __read_tab_section(self, sec_key: str, next_sec_key: str) -> StringIO:
         """Slices a file by section delimited by section keys
@@ -97,19 +97,19 @@ def __read_tab_section(self, sec_key: str, next_sec_key: str) -> StringIO:
         :param next_sec_key: Delimiter key of end of section
         :return: A memory file of the section slice, as a string buffer object
         """
-        fileline: str = self.__readline()
-        normed_line: str = fileline.rstrip().strip('"')
-        memory_file: StringIO = StringIO()
-
-        if normed_line != sec_key:
-            raise IOError(f"Invalid ISA-Tab section order: expected '{sec_key}', got '{normed_line}'")
-        while self.__peek().rstrip() != next_sec_key:
-            fileline = self.__readline()
-            if not fileline:
-                break
-            memory_file.write(fileline.rstrip() + "\n")
-        memory_file.seek(0)
-        return memory_file
+        fileline: str = self.__readline()
+        normed_line: str = fileline.rstrip().strip('"')
+        memory_file: StringIO = StringIO()
+
+        if normed_line != sec_key:
+            raise IOError(f"Invalid ISA-Tab section order: expected '{sec_key}', got '{normed_line}'")
+        while self.__peek().rstrip() != next_sec_key:
+            fileline = self.__readline()
+            if not fileline:
+                break
+            memory_file.write(fileline.rstrip() + "\n")
+        memory_file.seek(0)
+        return memory_file
 
     def __build_section_df(self, current_section_key: str, next_section_key: str) -> DataFrame:
         """Reads a file section into a DataFrame
@@ -149,7 +149,7 @@ def run(self) -> dict[str, DataFrame | str, list[DataFrame]]:
         return output
 
 
-class ISATabLoaderMixin(metaclass=ABCMeta):
+class ISATabLoaderMixin(metaclass=ABCMeta):
     """A mixin to provide modeling for the ISATab loaders. Provides shared methods, attributes and implementations
 
     - Properties:
@@ -173,12 +173,12 @@ class ISATabLoaderMixin(metaclass=ABCMeta):
     skip_load_tables: bool
     filepath: str
 
-    @staticmethod
-    @lru_cache(maxsize=64)
-    def _comment_columns(cols: tuple[str, ...]) -> tuple[str, ...]:
-        return tuple(x for x in cols if _RX_COMMENT.match(str(x)))
-
-    def __get_ontology_source(self, term_source_ref) -> OntologySource | None:
+    @staticmethod
+    @lru_cache(maxsize=64)
+    def _comment_columns(cols: tuple[str, ...]) -> tuple[str, ...]:
+        return tuple(x for x in cols if _RX_COMMENT.match(str(x)))
+
+    def __get_ontology_source(self, term_source_ref) -> OntologySource | None:
         """Small wrapper to return an ontology source from the map or None if not found
 
         :param term_source_ref: The term source reference
@@ -196,12 +196,12 @@ def get_contacts(self, contact_dataframe: DataFrame) -> list[Person]:
         contacts: list[Person] = []
         prefix: str
 
-        if "Investigation Person Last Name" in contact_dataframe.columns:
-            prefix = "Investigation "
-        elif "Study Person Last Name" in contact_dataframe.columns:
-            prefix = "Study "
-        else:
-            raise KeyError("Could not resolve contact section prefix from DataFrame columns")
+        if "Investigation Person Last Name" in contact_dataframe.columns:
+            prefix = "Investigation "
+        elif "Study Person Last Name" in contact_dataframe.columns:
+            prefix = "Study "
+        else:
+            raise KeyError("Could not resolve contact section prefix from DataFrame columns")
 
         for current_row in contact_dataframe.to_dict(orient="records"):
             person: Person = Person(
@@ -225,32 +225,32 @@ def get_contacts(self, contact_dataframe: DataFrame) -> list[Person]:
         return contacts
 
     @staticmethod
-    def get_comments(section_df: DataFrame) -> list[Comment]:
+    def get_comments(section_df: DataFrame) -> list[Comment]:
         """Get Comments from a section DataFrame
 
         :param section_df: A section DataFrame
-        :return: A list of Comment objects as found in the section
-        """
-        comments: list[Comment] = []
-        comment_columns = ISATabLoaderMixin._comment_columns(tuple(map(str, section_df.columns)))
-        for col in comment_columns:
-            for _, current_row in section_df.iterrows():
-                comments.append(Comment(name=next(iter(_RX_COMMENT.findall(col))), value=current_row[col]))
-        return comments
+        :return: A list of Comment objects as found in the section
+        """
+        comments: list[Comment] = []
+        comment_columns = ISATabLoaderMixin._comment_columns(tuple(map(str, section_df.columns)))
+        for col in comment_columns:
+            for _, current_row in section_df.iterrows():
+                comments.append(Comment(name=next(iter(_RX_COMMENT.findall(col))), value=current_row[col]))
+        return comments
 
     @staticmethod
-    def get_comments_row(cols, row) -> list[Comment]:
+    def get_comments_row(cols, row) -> list[Comment]:
         """Get Comments in a given DataFrame row
 
         :param cols: List of DataFrame columns
         :param row: DataFrame row as a Series object
-        :return: A list of Comment objects
-        """
-        comments: list[Comment] = []
-        comment_columns = ISATabLoaderMixin._comment_columns(tuple(map(str, cols)))
-        for col in comment_columns:
-            comments.append(Comment(name=next(iter(_RX_COMMENT.findall(col))), value=row[col]))
-        return comments
+        :return: A list of Comment objects
+        """
+        comments: list[Comment] = []
+        comment_columns = ISATabLoaderMixin._comment_columns(tuple(map(str, cols)))
+        for col in comment_columns:
+            comments.append(Comment(name=next(iter(_RX_COMMENT.findall(col))), value=row[col]))
+        return comments
 
     def get_ontology_annotation(self, val, accession, ts_ref) -> OntologyAnnotation | None:
         """Gets an OntologyAnnotation for a given value, accession and term source REF
@@ -294,12 +294,12 @@ def get_publications(self, section_df) -> list[Publication]:
         publications: list[Publication] = []
         prefix: str
 
-        if "Investigation PubMed ID" in section_df.columns:
-            prefix = "Investigation "
-        elif "Study PubMed ID" in section_df.columns:
-            prefix = "Study "
-        else:
-            raise KeyError("Could not resolve publication section prefix from DataFrame columns")
+        if "Investigation PubMed ID" in section_df.columns:
+            prefix = "Investigation "
+        elif "Study PubMed ID" in section_df.columns:
+            prefix = "Study "
+        else:
+            raise KeyError("Could not resolve publication section prefix from DataFrame columns")
 
         for _, current_row in section_df.iterrows():
             publication: Publication = Publication(