18830501663

Committed 24 Oct 2025 10:44AM UTC coverage: 86.433% (-0.3%) from 86.689%

Build # 18830501663

Build Type

push

github

Committed by

web-flow

Commit Message

Cleanup dependencies (#428)

Another attempt

---------

Co-authored-by: Karel Král <karelkral@google.com>

Run Details

55 of 67 new or added lines in 6 files covered. (82.09%)

4 existing lines in 1 file now uncovered.

3077 of 3560 relevant lines covered (86.43%)

0.86 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

43.93

/scaaml/utils.py

# Copyright 2020-2024 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#      http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Utils common to various SCAAML components"""

import functools
import importlib
from multiprocessing import Pool
from random import randint
import time
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Sequence,
    ParamSpec,
    TypeVar,
)

from glob import glob
from termcolor import cprint
from tqdm.auto import tqdm
from chipwhisperer.common.traces import Trace
import numpy as np
import numpy.typing as npt
import tensorflow as tf


def pretty_hex(val: int) -> str:
    "convert a value into a pretty hex"
    s = hex(int(val))
    s = s[2:]  # remove 0x
    if len(s) == 1:
        s = "0" + s
    return s.upper()


def bytelist_to_hex(lst: Sequence[int], spacer: str = " ") -> str:
    h = []

    for e in lst:
        h.append(pretty_hex(e))
    return spacer.join(h)


def hex_display(lst: Sequence[int],
                prefix: str = "",
                color: str = "green") -> None:
    "display a list of int as colored hex"
    h = []
    if len(prefix) > 0:
        prefix += "\t"
    for e in lst:
        h.append(pretty_hex(e))
    cprint(prefix + " ".join(h), color)


def get_model_stub(attack_point: str, attack_byte: int,
                   config: Dict[str, str]) -> str:
    device = config["device"]
    algorithm = config["algorithm"]
    model = config["model"]
    version = config["version"]
    max_trace_len = config["max_trace_len"]
    return (f"{device}-{algorithm}-{model}-v{version}-ap_{attack_point}-"
            f"byte_{attack_byte}-len_{max_trace_len}")


def get_target_stub(config: Dict[str, str]) -> str:
    device = config["device"]
    algorithm = config["algorithm"]
    return f"{device}-{algorithm}"


def get_num_gpu() -> int:
    return len(tf.config.list_physical_devices("GPU"))


def tf_cap_memory() -> None:
    gpus = tf.config.experimental.list_physical_devices("GPU")

    if gpus:
        for gpu in gpus:
            try:
                tf.config.experimental.set_memory_growth(gpu, True)
            except RuntimeError as e:
                # Memory growth must be set before GPUs have been initialized
                print(e)


def convert_shard_to_cw(info: Dict[str, Any]) -> List[Trace]:
    # avoid trashing the HD by de-synchronizing multi process
    time.sleep(randint(0, 100) / 1000)
    cw_traces = []
    shard = np.load(info["fname"])
    # CW traces
    cts = np.transpose(shard["cts"])
    pts = np.transpose(shard["pts"])
    keys = np.transpose(shard["keys"])

    for idx in range(info["num_traces_by_shard"]):
        wave = np.squeeze(shard["traces"][idx])
        wave = wave[:info["trace_len"]]

        t = Trace(wave, pts[idx], cts[idx], keys[idx])
        cw_traces.append(t)
    return cw_traces


def convert_to_chipwhisperer_format(file_pattern: str, num_shards: int,
                                    num_traces_by_shard: int,
                                    trace_len: int) -> List[Trace]:

    filenames = glob(file_pattern)[:num_shards]
    num_traces = len(filenames) * num_traces_by_shard

    # creating info for multiprocessing
    chunks = []
    for fname in filenames:
        chunks.append({
            "fname": fname,
            "num_traces_by_shard": num_traces_by_shard,
            "trace_len": trace_len
        })

    with Pool() as p:
        cw_traces = []
        pb = tqdm(total=num_traces, desc="Converting", unit="traces")
        for traces in p.imap_unordered(convert_shard_to_cw, chunks):
            cw_traces.extend(traces)
            pb.update(num_traces_by_shard)

        pb.close()
        return cw_traces


def display_config(config_name: str, config: Dict[str, str]) -> None:
    """Pretty print a config object in terminal.

    Args:
        config_name (str): name of the config
        config (dict): config to display
    """
    cprint(f"[{config_name}]", "magenta")
    cnt = 1
    for k, v in config.items():
        color: str = "yellow"
        if cnt % 2:
            color = "cyan"
        cprint(f"{k}:{v}", color)
        cnt += 1


def from_categorical(predictions: Sequence[npt.ArrayLike]) -> List[np.intp]:
    "reverse of categorical"
    # note: doing it as a list is significantly faster than a single argmax
    return [np.argmax(p) for p in predictions]


def is_module_present(module_name: str) -> bool:
    """Return True iff `module_name` can be imported. As a side-effect the
    module is imported.

    Args:

      module_name (str): The module which might or might not be importable.

    Raises: does not raise.
    """
    try:
        importlib.import_module(module_name)
        return True
    except ImportError:
        return False


# Type specifications for the requires decorator (which does not change the
# type of the function).
P = ParamSpec("P")
R = TypeVar("R")


def requires(module_name: str) -> Callable[[Callable[P, R]], Callable[P, R]]:
    """A decorator which marks a function or a method which needs an optional
    module to function. The first call will import the module if installed
    otherwise it will raise an ImportError.

    Performance penalty: The first call might be slower due to the deferred
    import. There is an additional performance penalty due to import checking
    in each call. It is visible in very fast functions however for most
    functions it should be ok:

    ```python
    >>> N = 10_000_000
    >>> f = lambda: 1
    >>> timeit.timeit(lambda: f(), number=N)
    0.6438868753612041
    >>> f = requires("numpy")(f)
    >>> timeit.timeit(lambda: f(), number=N)
    6.547926515340805
    ```
    """

    def decorator_requires(func: Callable[P, R]) -> Callable[P, R]:

        @functools.wraps(func)
        def wrapper_func(*args: P.args, **kwargs: P.kwargs) -> R:
            if is_module_present(module_name):
                return func(*args, **kwargs)
            raise ImportError(f"{func.__qualname__} needs the module "
                              f"{module_name}")

        return wrapper_func

    return decorator_requires

1	# Copyright 2020-2024 Google LLC
2	#
3	# Licensed under the Apache License, Version 2.0 (the "License");
4	# you may not use this file except in compliance with the License.
5	# You may obtain a copy of the License at
6	#
7	# http://www.apache.org/licenses/LICENSE-2.0
8	#
9	# Unless required by applicable law or agreed to in writing, software
10	# distributed under the License is distributed on an "AS IS" BASIS,
11	# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12	# See the License for the specific language governing permissions and
13	# limitations under the License.
14	"""Utils common to various SCAAML components"""
15
16	import functools	1✔
17	import importlib	1✔
18	from multiprocessing import Pool	1✔
19	from random import randint	1✔
20	import time	1✔
21	from typing import (	1✔
22	Any,
23	Callable,
24	Dict,
25	List,
26	Sequence,
27	ParamSpec,
28	TypeVar,
29	)
30
31	from glob import glob	1✔
32	from termcolor import cprint	1✔
33	from tqdm.auto import tqdm	1✔
34	from chipwhisperer.common.traces import Trace	1✔
35	import numpy as np	1✔
36	import numpy.typing as npt	1✔
37	import tensorflow as tf	1✔
38
39
40	def pretty_hex(val: int) -> str:	1✔
41	"convert a value into a pretty hex"
42	s = hex(int(val))	1✔
43	s = s[2:] # remove 0x	1✔
44	if len(s) == 1:	1✔
45	s = "0" + s	1✔
46	return s.upper()	1✔
47
48
49	def bytelist_to_hex(lst: Sequence[int], spacer: str = " ") -> str:	1✔
50	h = []	1✔
51
52	for e in lst:	1✔
53	h.append(pretty_hex(e))	1✔
54	return spacer.join(h)	1✔
55
56
57	def hex_display(lst: Sequence[int],	1✔
58	prefix: str = "",
59	color: str = "green") -> None:
60	"display a list of int as colored hex"
61	h = []	×
62	if len(prefix) > 0:	×
63	prefix += "\t"	×
64	for e in lst:	×
65	h.append(pretty_hex(e))	×
66	cprint(prefix + " ".join(h), color)	×
67
68
69	def get_model_stub(attack_point: str, attack_byte: int,	1✔
70	config: Dict[str, str]) -> str:
71	device = config["device"]	×
72	algorithm = config["algorithm"]	×
73	model = config["model"]	×
74	version = config["version"]	×
75	max_trace_len = config["max_trace_len"]	×
76	return (f"{device}-{algorithm}-{model}-v{version}-ap_{attack_point}-"	×
77	f"byte_{attack_byte}-len_{max_trace_len}")
78
79
80	def get_target_stub(config: Dict[str, str]) -> str:	1✔
81	device = config["device"]	×
82	algorithm = config["algorithm"]	×
83	return f"{device}-{algorithm}"	×
84
85
86	def get_num_gpu() -> int:	1✔
87	return len(tf.config.list_physical_devices("GPU"))	×
88
89
90	def tf_cap_memory() -> None:	1✔
91	gpus = tf.config.experimental.list_physical_devices("GPU")	×
92
93	if gpus:	×
94	for gpu in gpus:	×
95	try:	×
96	tf.config.experimental.set_memory_growth(gpu, True)	×
97	except RuntimeError as e:	×
98	# Memory growth must be set before GPUs have been initialized
99	print(e)	×
100
101
102	def convert_shard_to_cw(info: Dict[str, Any]) -> List[Trace]:	1✔
103	# avoid trashing the HD by de-synchronizing multi process
104	time.sleep(randint(0, 100) / 1000)	×
105	cw_traces = []	×
106	shard = np.load(info["fname"])	×
107	# CW traces
108	cts = np.transpose(shard["cts"])	×
109	pts = np.transpose(shard["pts"])	×
110	keys = np.transpose(shard["keys"])	×
111
112	for idx in range(info["num_traces_by_shard"]):	×
113	wave = np.squeeze(shard["traces"][idx])	×
114	wave = wave[:info["trace_len"]]	×
115
116	t = Trace(wave, pts[idx], cts[idx], keys[idx])	×
117	cw_traces.append(t)	×
118	return cw_traces	×
119
120
121	def convert_to_chipwhisperer_format(file_pattern: str, num_shards: int,	1✔
122	num_traces_by_shard: int,
123	trace_len: int) -> List[Trace]:
124
125	filenames = glob(file_pattern)[:num_shards]	×
126	num_traces = len(filenames) * num_traces_by_shard	×
127
128	# creating info for multiprocessing
129	chunks = []	×
130	for fname in filenames:	×
131	chunks.append({	×
132	"fname": fname,
133	"num_traces_by_shard": num_traces_by_shard,
134	"trace_len": trace_len
135	})
136
137	with Pool() as p:	×
138	cw_traces = []	×
139	pb = tqdm(total=num_traces, desc="Converting", unit="traces")	×
140	for traces in p.imap_unordered(convert_shard_to_cw, chunks):	×
141	cw_traces.extend(traces)	×
142	pb.update(num_traces_by_shard)	×
143
144	pb.close()	×
145	return cw_traces	×
146
147
148	def display_config(config_name: str, config: Dict[str, str]) -> None:	1✔
149	"""Pretty print a config object in terminal.
150
151	Args:
152	config_name (str): name of the config
153	config (dict): config to display
154	"""
155	cprint(f"[{config_name}]", "magenta")	×
156	cnt = 1	×
157	for k, v in config.items():	×
158	color: str = "yellow"	×
159	if cnt % 2:	×
160	color = "cyan"	×
161	cprint(f"{k}:{v}", color)	×
162	cnt += 1	×
163
164
165	def from_categorical(predictions: Sequence[npt.ArrayLike]) -> List[np.intp]:	1✔
166	"reverse of categorical"
167	# note: doing it as a list is significantly faster than a single argmax
168	return [np.argmax(p) for p in predictions]	×
169
170
171	def is_module_present(module_name: str) -> bool:	1✔
172	"""Return True iff `module_name` can be imported. As a side-effect the
173	module is imported.
174
175	Args:
176
177	module_name (str): The module which might or might not be importable.
178
179	Raises: does not raise.
180	"""
181	try:	1✔
182	importlib.import_module(module_name)	1✔
183	return True	1✔
NEW 184	except ImportError:	×
NEW 185	return False	×
186
187
188	# Type specifications for the requires decorator (which does not change the
189	# type of the function).
190	P = ParamSpec("P")	1✔
191	R = TypeVar("R")	1✔
192
193
194	def requires(module_name: str) -> Callable[[Callable[P, R]], Callable[P, R]]:	1✔
195	"""A decorator which marks a function or a method which needs an optional
196	module to function. The first call will import the module if installed
197	otherwise it will raise an ImportError.
198
199	Performance penalty: The first call might be slower due to the deferred
200	import. There is an additional performance penalty due to import checking
201	in each call. It is visible in very fast functions however for most
202	functions it should be ok:
203
204	```python
205	>>> N = 10_000_000
206	>>> f = lambda: 1
207	>>> timeit.timeit(lambda: f(), number=N)
208	0.6438868753612041
209	>>> f = requires("numpy")(f)
210	>>> timeit.timeit(lambda: f(), number=N)
211	6.547926515340805
212	```
213	"""
214
215	def decorator_requires(func: Callable[P, R]) -> Callable[P, R]:	1✔
216
217	@functools.wraps(func)	1✔
218	def wrapper_func(args: P.args, *kwargs: P.kwargs) -> R:	1✔
219	if is_module_present(module_name):	1✔
220	return func(args, *kwargs)	1✔
NEW 221	raise ImportError(f"{func.__qualname__} needs the module "	×
222	f"{module_name}")
223
224	return wrapper_func	1✔
225
226	return decorator_requires	1✔

google / scaaml / 18830501663

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous