aboutsummaryrefslogtreecommitdiffhomepage
diff options
context:
space:
mode:
authorTomáš Jusko2023-09-18 15:27:28 +0200
committerTomáš Jusko2023-09-18 15:27:28 +0200
commit80f834f8b27d04a41974cec545a1f280f26f0698 (patch)
treecccacd49c167a702a0ec8a79bc767425349016ca
parenta8649aba83999b8695610ab7ed1cd46a408b2440 (diff)
downloadpyecsca-80f834f8b27d04a41974cec545a1f280f26f0698.tar.gz
pyecsca-80f834f8b27d04a41974cec545a1f280f26f0698.tar.zst
pyecsca-80f834f8b27d04a41974cec545a1f280f26f0698.zip
fix: Deepsource errors
-rw-r--r--pyecsca/sca/stacked_traces/combine.py69
1 files changed, 42 insertions, 27 deletions
diff --git a/pyecsca/sca/stacked_traces/combine.py b/pyecsca/sca/stacked_traces/combine.py
index 7dbf89a..7cf4691 100644
--- a/pyecsca/sca/stacked_traces/combine.py
+++ b/pyecsca/sca/stacked_traces/combine.py
@@ -141,27 +141,8 @@ class GPUTraceManager(BaseTraceManager):
item_size=self._traces.samples.itemsize,
chunk_item_count=self._traces.samples.shape[0])
- def _check_init_args(self,
- chunk_size: Optional[int],
- chunk_memory_ratio: Optional[float],
- tpb: TPB) -> None:
- if not cuda.is_available():
- raise RuntimeError("CUDA is not available, "
- "use CPUTraceManager instead")
-
- if chunk_size and chunk_memory_ratio:
- raise ValueError("Only one of chunk_size and chunk_memory_ratio "
- "can be specified")
-
- if chunk_memory_ratio is not None \
- and (chunk_memory_ratio <= 0 or chunk_memory_ratio > 0.5):
- raise ValueError("Chunk memory ratio should be in (0, 0.5], "
- "because two chunks are stored in memory "
- "at once")
-
- if chunk_size is not None and chunk_size <= 0:
- raise ValueError("Chunk size should be positive")
-
+ @staticmethod
+ def _check_tpb(tpb: TPB) -> None:
dev = cuda.get_current_device()
warp_size = dev.WARP_SIZE
max_tpb = dev.MAX_THREADS_PER_BLOCK
@@ -185,6 +166,33 @@ class GPUTraceManager(BaseTraceManager):
)
@staticmethod
+ def _check_chunk_sizing(chunk_size: Optional[int],
+ chunk_memory_ratio: Optional[float]) -> None:
+ if chunk_size and chunk_memory_ratio:
+ raise ValueError("Only one of chunk_size and chunk_memory_ratio "
+ "can be specified")
+
+ if chunk_memory_ratio is not None \
+ and (chunk_memory_ratio <= 0 or chunk_memory_ratio > 0.5):
+ raise ValueError("Chunk memory ratio should be in (0, 0.5], "
+ "because two chunks are stored in memory "
+ "at once")
+
+ if chunk_size is not None and chunk_size <= 0:
+ raise ValueError("Chunk size should be positive")
+
+ @staticmethod
+ def _check_init_args(chunk_size: Optional[int],
+ chunk_memory_ratio: Optional[float],
+ tpb: TPB) -> None:
+ if not cuda.is_available():
+ raise RuntimeError("CUDA is not available, "
+ "use CPUTraceManager instead")
+
+ GPUTraceManager._check_chunk_sizing(chunk_size, chunk_memory_ratio)
+ GPUTraceManager._check_tpb(tpb)
+
+ @staticmethod
def chunk_size_from_ratio(chunk_memory_ratio: float,
element_size: int | None = None,
item_size: int | None = None,
@@ -230,8 +238,9 @@ class GPUTraceManager(BaseTraceManager):
:param output_count: Number of outputs expected from the GPU function.
:return: Combined trace output from the GPU function
"""
- assert self._chunk_size is None
- assert isinstance(self._tpb, int)
+ if not isinstance(self._tpb, int):
+ raise ValueError("Something went wrong. "
+ "TPB should be an int")
device_input = cuda.to_device(self._traces.samples)
device_outputs = [
@@ -246,9 +255,15 @@ class GPUTraceManager(BaseTraceManager):
def _gpu_combine1D_chunked(self, func, output_count: int = 1) \
-> List[npt.NDArray[np.number]]:
- assert self._chunk_size is not None
- assert self._stream_count is not None
- assert isinstance(self._tpb, int)
+ if self._chunk_size is None:
+ raise ValueError("Something went wrong. "
+ "Chunk size should be specified")
+ if self._stream_count is None:
+ raise ValueError("Something went wrong. "
+ "Stream count should be specified")
+ if not isinstance(self._tpb, int):
+ raise ValueError("Something went wrong. "
+ "TPB should be an int")
chunk_count = (
self._traces.samples.shape[1] + self._chunk_size - 1
@@ -265,7 +280,7 @@ class GPUTraceManager(BaseTraceManager):
]
chunk_results: List[List[npt.NDArray[np.number]]] = [
- list() for _ in range(output_count)]
+ [] for _ in range(output_count)]
with cuda.defer_cleanup():
for chunk in range(chunk_count):