Skip to content
This repository has been archived by the owner on Mar 9, 2022. It is now read-only.

Remove assert statement from non-test files #80

Merged
merged 1 commit into from
Oct 19, 2021
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
50 changes: 30 additions & 20 deletions ifpd2/asserts.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,48 +10,58 @@


def ert_type(x, stype, label):
assert isinstance(x, stype), f"{label} should be {stype}, {type(x)} instead"
if not isinstance(x, stype):
raise AssertionError(f"{label} should be {stype}, {type(x)} instead")


def ert_multiTypes(x, types, label):
cond = any(isinstance(x, t) for t in types)
assert cond, f"{label} should be one of {types}, {type(x)} instead"
if not cond:
raise AssertionError(f"{label} should be one of {types}, {type(x)} instead")


def ert_nonNeg(x, label, include_zero=False):
if not include_zero:
assert x > 0, f"{label} should be greater than 0"
if x <= 0:
raise AssertionError(f"{label} should be greater than 0")
else:
assert x >= 0, f"{label} should be greater than or equal to 0"
if x < 0:
raise AssertionError(f"{label} should be greater than or equal to 0")


def ert_inInterv(x, vmin, vmax, label, leftClose=False, rightClose=True):
if leftClose:
if rightClose:
assert x >= vmin and x <= vmax, f"expected {vmin}<={label}<={vmax}"
if not (x >= vmin and x <= vmax):
raise AssertionError(f"expected {vmin}<={label}<={vmax}")
else:
assert x >= vmin and x < vmax, f"expected {vmin}<={label}<{vmax}"
if not (x >= vmin and x < vmax):
raise AssertionError(f"expected {vmin}<={label}<{vmax}")
elif rightClose:
assert x > vmin and x <= vmax, f"expected {vmin}<{label}<={vmax}"
if not (x > vmin and x <= vmax):
raise AssertionError(f"expected {vmin}<{label}<={vmax}")
else:
assert x > vmin and x < vmax, f"expected {vmin}<{label}<{vmax}"
if not (x > vmin and x < vmax):
raise AssertionError(f"expected {vmin}<{label}<{vmax}")


def ert_in_dtype(x, dtype):
if dtype.startswith("f"):
assert x <= np.finfo(dtype).max, " ".join(
[
"expected to be lower than {dtype} max:",
f"{x} < {np.finfo(dtype).max}",
]
)
if x > np.finfo(dtype).max:
raise AssertionError(" ".join(
[
"expected to be lower than {dtype} max:",
f"{x} < {np.finfo(dtype).max}",
]
))
elif dtype.startswith("u") or dtype.startswith("i"):
assert x <= np.iinfo(dtype).max, " ".join(
[
"expected to be lower than {dtype} max:",
f"{x} < {np.iinfo(dtype).max}",
]
)
if x > np.iinfo(dtype).max:
raise AssertionError(" ".join(
[
"expected to be lower than {dtype} max:",
f"{x} < {np.iinfo(dtype).max}",
]
))
else:
logging.warning(f"assert not implemented for dtype '{dtype}'")

Expand Down
30 changes: 20 additions & 10 deletions ifpd2/chromosome.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,8 @@ class ChromosomeIndex(object):

def __init__(self, bin_size: int):
super(ChromosomeIndex, self).__init__()
assert bin_size >= 1
if bin_size < 1:
raise AssertionError
self._bin_size = bin_size

def __init_index(self, chrom_db: pd.DataFrame) -> None:
Expand Down Expand Up @@ -63,7 +64,8 @@ def __populate_bins(
for i in range(chrom_db.shape[0]):
track[0].update(track[1], advance=1)
position_in_nt = chrom_db["start"].values[i]
assert position_in_nt > current_position
if position_in_nt <= current_position:
raise AssertionError
current_position = position_in_nt

position_in_bytes = record_byte_size * i
Expand Down Expand Up @@ -101,16 +103,19 @@ def build(
track {Tuple[Progress, TaskID]} -- progress bar details
"""
for colname in ("chromosome", "start", "end"):
assert colname in chrom_db.columns, f"missing '{colname}' column"
if colname not in chrom_db.columns:
raise AssertionError(f"missing '{colname}' column")
chromosome_set: Set[bytes] = set(chrom_db["chromosome"].values)
assert len(chromosome_set) == 1
if len(chromosome_set) != 1:
raise AssertionError

self.__init_index(chrom_db)
self.__populate_bins(chrom_db, record_byte_size, track)
self.__fill_empty_bins()

def __getitem__(self, position_in_nt: int) -> int:
assert self._index is not None
if self._index is None:
raise AssertionError
binned_to = position_in_nt // self._bin_size
if binned_to not in self._index:
return -1
Expand Down Expand Up @@ -145,12 +150,15 @@ def __init__(
):
super(ChromosomeData, self).__init__()

assert "chromosome" in chromosome_db.columns
if "chromosome" not in chromosome_db.columns:
raise AssertionError
selected_chrom = chromosome_db["chromosome"][0]
assert len(set(chromosome_db["chromosome"].values)) == 1
if len(set(chromosome_db["chromosome"].values)) != 1:
raise AssertionError

self._record_byte_size = get_dtype_length(dtype)
assert self._record_byte_size > 0
if self._record_byte_size <= 0:
raise AssertionError

self._name = selected_chrom
self._recordno = chromosome_db.shape[0]
Expand Down Expand Up @@ -195,7 +203,8 @@ def _build_index(
index_bin_size {int} -- index bin size
progress {Progress} -- progress instance for progress bar with rich
"""
assert index_bin_size > 0
if index_bin_size <= 0:
raise AssertionError
indexing_track = progress.add_task(
f"indexing {self._name.decode()}.bin",
total=chromosome_db.shape[0],
Expand Down Expand Up @@ -226,7 +235,8 @@ class ChromosomeDict(object):
def __init__(self, index_bin_size: int = const.DEFAULT_DATABASE_INDEX_BIN_SIZE):
super(ChromosomeDict, self).__init__()
self._data = {}
assert index_bin_size > 0
if index_bin_size <= 0:
raise AssertionError
self._index_bin_size = index_bin_size

def __len__(self) -> int:
Expand Down
20 changes: 13 additions & 7 deletions ifpd2/database.py
Original file line number Diff line number Diff line change
Expand Up @@ -38,7 +38,8 @@ def __parse_bytes(self, record: bytes, column_dtypes: Dict[str, str]) -> None:
record {bytes} -- record bytes
column_dtypes {Dict[str, str]} -- column dtypes
"""
assert len(record) == get_dtype_length(column_dtypes)
if len(record) != get_dtype_length(column_dtypes):
raise AssertionError
self._data = {}
current_location = 0
for label in const.database_columns:
Expand Down Expand Up @@ -137,9 +138,11 @@ def __init__(self, path: str):
path {str} -- absolute path to database folder
"""
super(DataBase, self).__init__()
assert os.path.isdir(path), f"cannot find database folder '{path}'"
if not os.path.isdir(path):
raise AssertionError(f"cannot find database folder '{path}'")
db_pickle_path = os.path.join(path, "db.pickle")
assert os.path.isfile(db_pickle_path), f"'db.pickle is missing in '{path}'"
if not os.path.isfile(db_pickle_path):
raise AssertionError(f"'db.pickle is missing in '{path}'")

with open(db_pickle_path, "rb") as IH:
details = pickle.load(IH)
Expand All @@ -150,15 +153,18 @@ def __init__(self, path: str):
else:
self._args = dict(details["args"])

assert isinstance(self._chromosomes, ChromosomeDict)
if not isinstance(self._chromosomes, ChromosomeDict):
raise AssertionError
for chromosome in self.chromosome_list:
chromosome_path = os.path.join(path, f"{chromosome.decode()}.bin")
assert os.path.isfile(
if not os.path.isfile(
chromosome_path
), f"missing expected chromosome file: '{chromosome_path}'"
):
raise AssertionError(f"missing expected chromosome file: '{chromosome_path}'")

self._record_byte_size = get_dtype_length(self._dtype)
assert self._record_byte_size > 0
if self._record_byte_size <= 0:
raise AssertionError

self._root = path

Expand Down
63 changes: 42 additions & 21 deletions ifpd2/dataclasses.py
Original file line number Diff line number Diff line change
Expand Up @@ -19,9 +19,11 @@ class Folder:

def __post_init__(self):
if self.exists:
assert isdir(self.path)
if not isdir(self.path):
raise AssertionError
else:
assert not path_exists(self.path)
if path_exists(self.path):
raise AssertionError


@dataclass(frozen=True)
Expand All @@ -31,33 +33,38 @@ class File:

def __post_init__(self):
if self.exists:
assert isfile(self.path)
if not isfile(self.path):
raise AssertionError
else:
assert not path_exists(self.path)
if path_exists(self.path):
raise AssertionError


@dataclass(frozen=True)
class PositiveInteger:
n: int

def __post_init__(self):
assert self.n >= 1
if self.n < 1:
raise AssertionError


@dataclass(frozen=True)
class NonNegativeFloat:
n: float

def __post_init__(self):
assert self.n >= 0
if self.n < 0:
raise AssertionError


@dataclass(frozen=True)
class NonNegativeInteger:
n: int

def __post_init__(self):
assert self.n > 0
if self.n <= 0:
raise AssertionError


@dataclass(frozen=True)
Expand All @@ -68,12 +75,15 @@ class PositiveFloat:

def __post_init__(self):
if self.limit is None:
assert 0 < self.n
if 0 >= self.n:
raise AssertionError

elif self.limit_included:
assert 0 < self.n <= self.limit
if not 0 < self.n <= self.limit:
raise AssertionError
else:
assert 0 < self.n < self.limit
if not 0 < self.n < self.limit:
raise AssertionError


@dataclass(frozen=True)
Expand All @@ -82,8 +92,10 @@ class GenomicRegion:
end: int

def __post_init__(self):
assert self.start >= 0
assert self.end >= self.start or self.end == -1
if self.start < 0:
raise AssertionError
if not (self.end >= self.start or self.end == -1):
raise AssertionError

def astuple(self) -> Tuple[int, int]:
return (self.start, self.end)
Expand All @@ -95,8 +107,10 @@ class NonNegativeIntInterval:
end: int

def __post_init__(self):
assert self._from >= 0
assert self.end >= self.first
if self._from < 0:
raise AssertionError
if self.end < self.first:
raise AssertionError

def astuple(self) -> Tuple[int, int]:
return (self.start, self.end)
Expand All @@ -109,8 +123,10 @@ class QueryWindow:

def __post_init__(self):
if self.size is not None:
assert self.size >= 1
assert 0 < self.shift <= 1
if self.size < 1:
raise AssertionError
if not 0 < self.shift <= 1:
raise AssertionError

def astuple(self) -> Tuple[Optional[int], Optional[float]]:
return (self.size, self.shift)
Expand All @@ -122,8 +138,10 @@ class QueryFocus:
step: float

def __post_init__(self):
assert self.size > 0
assert self.step > 0
if self.size <= 0:
raise AssertionError
if self.step <= 0:
raise AssertionError


@dataclass(frozen=True)
Expand All @@ -141,6 +159,9 @@ class GCRange:
high: float

def __post_init__(self):
assert 0 <= self.low <= 1
assert 0 <= self.high <= 1
assert self.low <= self.high
if not 0 <= self.low <= 1:
raise AssertionError
if not 0 <= self.high <= 1:
raise AssertionError
if self.low > self.high:
raise AssertionError
3 changes: 2 additions & 1 deletion ifpd2/fasta.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@


def extract_kmers(input_path: str, kmer_size: int) -> List[Tuple[str, str]]:
assert isfile(input_path)
if not isfile(input_path):
raise AssertionError
with open(input_path) as FH:
oligos_list: List[Tuple[str, str]] = []
for record_header, record_sequence in tqdm(
Expand Down
12 changes: 8 additions & 4 deletions ifpd2/io.py
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,8 @@ def parse_hush(path: str) -> pd.DataFrame:
Returns:
pd.DataFrame -- parsed HUSH data
"""
assert os.path.isfile(path), f"cannot find file '{path}'"
if not os.path.isfile(path):
raise AssertionError(f"cannot find file '{path}'")
logging.info(f"parsing: '{path}'")
sequence_lengths: Set[int] = set()
parsed_lines: List[Tuple[str, str, int]] = []
Expand Down Expand Up @@ -81,11 +82,13 @@ def parse_melting(path: str, sep: str = "\t", header: bool = True) -> pd.DataFra
Returns:
pd.DataFrame -- parsed oligo-melting data
"""
assert os.path.isfile(path), f"cannot find file '{path}'"
if not os.path.isfile(path):
raise AssertionError(f"cannot find file '{path}'")
logging.info(f"parsing: '{path}'")
expected_columns = copy.copy(const.dtype_melting)
melting_df = pd.read_csv(path, sep=sep, header=None, skiprows=1 if header else 0)
assert melting_df.shape[1] == len(expected_columns)
if melting_df.shape[1] != len(expected_columns):
raise AssertionError
melting_df.columns = list(expected_columns.keys())
melting_df.set_index("name", inplace=True)
expected_columns.pop("name", None)
Expand All @@ -104,7 +107,8 @@ def parse_secondary(path: str) -> pd.DataFrame:
Returns:
pd.DataFrame -- parsed OligoArrayAux .ct data
"""
assert os.path.isfile(path), f"cannot find file '{path}'"
if not os.path.isfile(path):
raise AssertionError(f"cannot find file '{path}'")
logging.info(f"parsing: '{path}'")
parsed_lines: List[Tuple[str, float]] = []
with open(path, "r") as IH:
Expand Down
Loading