Acquisition Module

Acquisition is responsible for managing the process of waveform data acquisition.

Data acquisition has two modes: Test and Run: Test and Run. The Test mode can be understood as a debugging mode, where real-time adjustments to the Cracker configuration and data acquisition settings can be made to achieve better data collection results (currently, this mode is meaningful only in the Jupyter environment, where waveforms can be viewed in real-time through the oscilloscope panel). In this mode, the acquired waveform data is not saved.

In the Run mode, the user cannot make real-time configurations to the device. This mode is mainly used for data collection after various parameter adjustments are completed. In this mode, the user can specify the file path and file format for saving the data.

The acquisition process consists of the following steps:

• Pre-initialization

• Initialization

• Post-initialization

• Pre-acquisition

• Data acquisition

• Post-acquisition

• Pre-completion

• Completion

• Post-completion

The end user mainly needs to focus on the Initialization and Data Acquisition steps, which are the init and do methods of the Acquisition class in the code. Typically, key settings for the device are made in init, while specific operations like encrypted data transmission and result reception are handled in do.

The flowchart is shown below:

Acquisition

class cracknuts.acquisition.acquisition.Acquisition(cracker, trace_count=1000, sample_length=-1, sample_offset=0, data_plaintext_length=None, data_ciphertext_length=None, data_key_length=None, data_extended_length=None, trigger_judge_wait_time=0.01, trigger_judge_timeout=1.0, do_error_handler_strategy=0, do_error_max_count=-1, file_format='zarr', file_path='auto', trace_fetch_interval=0)

Bases: ABC

Trace acquisition process control class

DO_ERROR_HANDLER_STRATEGY_CONTINUE_UNTIL_MAX_ERROR_COUNT: int = 1

DO_ERROR_HANDLER_STRATEGY_EXIT: int = 0

MAX_WAVE_LENGTH = 32000000

STATUS_GLITCH_TEST_RUNNING: int = 3

STATUS_RUNNING: int = 2

STATUS_STOPPED: int = 0

STATUS_TESTING: int = 1

__abstractmethods__ = frozenset({'do', 'init'})

__annotations__ = {'DO_ERROR_HANDLER_STRATEGY_CONTINUE_UNTIL_MAX_ERROR_COUNT': <class 'int'>, 'DO_ERROR_HANDLER_STRATEGY_EXIT': <class 'int'>, 'STATUS_GLITCH_TEST_RUNNING': <class 'int'>, 'STATUS_RUNNING': <class 'int'>, 'STATUS_STOPPED': <class 'int'>, 'STATUS_TESTING': <class 'int'>, '_STATUS_PAUSED_SWITCH': <class 'int'>, '_do_error_max_count': 'int', '_file_format': 'str', '_file_path': 'str', '_last_wave': 'dict[int, np.ndarray] | None', '_on_config_changed_listener': 'list[typing.Callable[[str, typing.Any], None]]', '_on_run_progress_changed_listeners': 'list[typing.Callable[[dict], None]]', '_on_status_change_listeners': 'list[typing.Callable[[int], None]]', '_on_wave_loaded_callback': 'typing.Callable[[typing.Any], None] | None', '_run_thread_pause_event': 'threading.Event', '_sample_offset': 'int', '_status': 'int', '_trace_count': 'int', '_trigger_judge_timeout': 'float', '_trigger_judge_wait_time': 'float', 'cracker': 'CrackerBasic', 'do_error_handler_strategy': 'int', 'metadata_ciphertext_length': 'int', 'metadata_extended_length': 'int', 'metadata_key_length': 'int', 'metadata_plaintext_length': 'int'}

__dict__ = mappingproxy({'__module__': 'cracknuts.acquisition.acquisition', '__annotations__': {'STATUS_STOPPED': <class 'int'>, 'STATUS_TESTING': <class 'int'>, 'STATUS_RUNNING': <class 'int'>, 'STATUS_GLITCH_TEST_RUNNING': <class 'int'>, '_STATUS_PAUSED_SWITCH': <class 'int'>, 'DO_ERROR_HANDLER_STRATEGY_EXIT': <class 'int'>, 'DO_ERROR_HANDLER_STRATEGY_CONTINUE_UNTIL_MAX_ERROR_COUNT': <class 'int'>, '_last_wave': 'dict[int, np.ndarray] | None', '_status': 'int', '_run_thread_pause_event': 'threading.Event', 'cracker': 'CrackerBasic', '_trace_count': 'int', '_sample_offset': 'int', 'metadata_plaintext_length': 'int', 'metadata_ciphertext_length': 'int', 'metadata_key_length': 'int', 'metadata_extended_length': 'int', '_trigger_judge_wait_time': 'float', '_trigger_judge_timeout': 'float', 'do_error_handler_strategy': 'int', '_do_error_max_count': 'int', '_file_format': 'str', '_file_path': 'str', '_on_wave_loaded_callback': 'typing.Callable[[typing.Any], None] | None', '_on_status_change_listeners': 'list[typing.Callable[[int], None]]', '_on_run_progress_changed_listeners': 'list[typing.Callable[[dict], None]]', '_on_config_changed_listener': 'list[typing.Callable[[str, typing.Any], None]]'}, '__doc__': '\n    Trace acquisition process control class\n    ', 'STATUS_STOPPED': 0, 'STATUS_TESTING': 1, 'STATUS_RUNNING': 2, 'STATUS_GLITCH_TEST_RUNNING': 3, '_STATUS_PAUSED_SWITCH': -1, 'DO_ERROR_HANDLER_STRATEGY_EXIT': 0, 'DO_ERROR_HANDLER_STRATEGY_CONTINUE_UNTIL_MAX_ERROR_COUNT': 1, '_DATASET_DEFAULT_PATH': './dataset/', 'MAX_WAVE_LENGTH': 32000000, '__init__': <function Acquisition.__init__>, 'on_config_changed': <function Acquisition.on_config_changed>, 'trace_count': <property object>, 'sample_length': <property object>, 'sample_offset': <property object>, 'trigger_judge_wait_time': <property object>, 'trigger_judge_timeout': <property object>, 'do_error_max_count': <property object>, 'file_format': <property object>, 'file_path': <property object>, 'trace_fetch_interval': <property object>, 'get_status': <function Acquisition.get_status>, 'set_cracker': <function Acquisition.set_cracker>, 'on_status_changed': <function Acquisition.on_status_changed>, 'on_run_progress_changed': <function Acquisition.on_run_progress_changed>, 'on_wave_loaded': <function Acquisition.on_wave_loaded>, 'run': <function Acquisition.run>, 'run_sync': <function Acquisition.run_sync>, 'test': <function Acquisition.test>, 'test_sync': <function Acquisition.test_sync>, 'pause': <function Acquisition.pause>, 'resume': <function Acquisition.resume>, 'stop': <function Acquisition.stop>, '_run': <function Acquisition._run>, '_do_run': <function Acquisition._do_run>, '_status_changed': <function Acquisition._status_changed>, '_loop': <function Acquisition._loop>, '_progress_changed': <function Acquisition._progress_changed>, '_loop_without_log': <function Acquisition._loop_without_log>, 'connect_scrat': <function Acquisition.connect_scrat>, 'connect_cracker': <function Acquisition.connect_cracker>, 'connect_net': <function Acquisition.connect_net>, 'config_scrat': <function Acquisition.config_scrat>, 'config_cracker': <function Acquisition.config_cracker>, 'pre_init': <function Acquisition.pre_init>, 'init': <function Acquisition.init>, '_post_init': <function Acquisition._post_init>, 'transfer': <function Acquisition.transfer>, 'pre_do': <function Acquisition.pre_do>, 'do': <function Acquisition.do>, '_do': <function Acquisition._do>, '_post_do': <function Acquisition._post_do>, '_is_triggered': <function Acquisition._is_triggered>, '_get_waves': <function Acquisition._get_waves>, '_pre_finish': <function Acquisition._pre_finish>, '_post_finish': <function Acquisition._post_finish>, '_save_dataset': <function Acquisition._save_dataset>, 'finish': <function Acquisition.finish>, 'get_last_wave': <function Acquisition.get_last_wave>, 'dump_config': <function Acquisition.dump_config>, 'load_config_from_file': <function Acquisition.load_config_from_file>, 'load_config_from_str': <function Acquisition.load_config_from_str>, 'load_config_from_json': <function Acquisition.load_config_from_json>, 'is_running': <function Acquisition.is_running>, '__dict__': <attribute '__dict__' of 'Acquisition' objects>, '__weakref__': <attribute '__weakref__' of 'Acquisition' objects>, '__abstractmethods__': frozenset({'do', 'init'}), '_abc_impl': <_abc._abc_data object>})

__init__(cracker, trace_count=1000, sample_length=-1, sample_offset=0, data_plaintext_length=None, data_ciphertext_length=None, data_key_length=None, data_extended_length=None, trigger_judge_wait_time=0.01, trigger_judge_timeout=1.0, do_error_handler_strategy=0, do_error_max_count=-1, file_format='zarr', file_path='auto', trace_fetch_interval=0)

Parameters:

• cracker (CrackerBasic) – The controlled Cracker object.

• trace_count (int) – The number of traces to acquire.

• sample_length (int) – the sample length to be acquired, If it is -1, it will be set to the value of cracker.current_count().sample_len.

• sample_offset (int) – the sample offset to be acquired.

• trigger_judge_timeout (float) – The trigger judge timeout in seconds.

• trigger_judge_wait_time (float) – The trigger judge wait time in seconds.

• do_error_handler_strategy (int) – The strategy to handle error handling. 0: Exit immediately, 1: Exit after exceeding the error count.

• do_error_max_count (int) – The maximum number of error count. If do_error_handler_strategy is 0, this setting is invalid.

• file_format (str) – The file format of the trace dataset. “zarr” or “numpy”.

• file_path (str) – The file path of the trace dataset. If set to “auto”, a folder with a timestamp format will be created in the current working directory to save the data.

__module__ = 'cracknuts.acquisition.acquisition'

__weakref__

list of weak references to the object

config_cracker()

config_scrat()

connect_cracker()

Connect to cracker device.

connect_net()

connect_scrat()

Connect to scrat device.

abstractmethod do(count)

The do logic, which the user should implement in the subclass.

The returned dictionary should contain plaintext and ciphertext keys and may optionally include key and extended keys. The plaintext is the data to be encrypted, the ciphertext is the encrypted result, the key is used to encrypt the plaintext, and the extended field is for any custom data you want to save. Expected structure: {"plaintext": bytes, "ciphertext": bytes, "key": bytes, "extended": bytes}.

Parameters:

count (int) – The loop count, starting from 0.

Returns:

Data in dictionary format to be saved.

Return type:

dict[str, bytes]

property do_error_max_count

dump_config(path=None)

Dump the current config to a JSON file if a path is specified, or to a JSON string if no path is specified.

Parameters:

path (str | None) – The path to the JSON file.

Returns:

The JSON string content, or None if a file path was specified.

Return type:

str | None

property file_format

property file_path

finish()

get_last_wave()

Get the last acquired wave.

Returns:

Dictionary mapping channel index to wave data array, or None if not yet available.

Return type:

dict[int, np.ndarray] | None

get_status()

abstractmethod init()

The init logic, which the user should implement in the subclass.

is_running()

load_config_from_file(path)

Load config from a JSON file.

Parameters:

path (str) – The path to the JSON file.

Return type:

None

Returns:

None

load_config_from_json(config)

Load config from a JSON object.

Parameters:

config (dict) – The JSON object containing configuration values.

Returns:

None

load_config_from_str(json_str)

Load config from a JSON string.

Parameters:

json_str (str) – The JSON string.

Return type:

None

Returns:

None

on_config_changed(listener)

on_run_progress_changed(callback)

Return type:

None

on_status_changed(callback)

User should not use this function. If you need to perform actions when the ACQ state changes, please use the node functions in the ACQ lifecycle point: init, do, and finish.

Status values: 0 = stopped, 1 = testing, 2 = running, 3 = paused.

Parameters:

callback (Callable[[int], None]) – Callable that receives the new status integer.

Return type:

None

Returns:

None

on_wave_loaded(callback)

Return type:

None

pause()

Pause test or run mode.

pre_do()

pre_init()

resume()

Resume from paused test or run mode.

run(count=None, sample_length=None, sample_offset=None, data_plaintext_length=None, data_ciphertext_length=None, data_key_length=None, data_extended_length=None, trigger_judge_wait_time=None, trigger_judge_timeout=None, do_error_max_count=None, do_error_handler_strategy=None, file_format=None, file_path=None)

Start run mode in the background. The parameters configured here will override the settings specified in __init__.

Parameters:

• count (int) – The number of traces to acquire.

• sample_length (int) – The sample length to be acquired.

• sample_offset (int) – The sample offset to be acquired.

• trigger_judge_wait_time (float) – The trigger judge wait time in seconds.

• trigger_judge_timeout (float) – The trigger judge timeout in seconds.

• do_error_max_count (int) – The maximum number of error count. If do_error_handler_strategy is 0, this setting is invalid.

• do_error_handler_strategy (int) – The strategy to handle error handling. 0: Exit immediately, 1: Exit after exceeding the error count.

• file_format (str) – The file format of the trace dataset. “zarr” or “numpy”.

• file_path (str) – The file path of the trace dataset. If set to “auto”, a folder with a timestamp format will be created in the current working directory to save the data.

run_sync(count=1, sample_length=1024, sample_offset=0, trigger_judge_wait_time=None, trigger_judge_timeout=None, do_error_max_count=None, do_error_handler_strategy=None, file_format='zarr', file_path='auto')

Start run mode in the foreground, which will cause blocking.

Parameters:

• count (int) – The number of traces to acquire.

• sample_length (int) – The sample length to be acquired.

• sample_offset (int) – The sample offset to be acquired.

• trigger_judge_wait_time (float) – The trigger judge wait time in seconds.

• trigger_judge_timeout (float) – The trigger judge timeout in seconds.

• do_error_max_count (int) – The maximum number of error count. If do_error_handler_strategy is 0, this setting is invalid.

• do_error_handler_strategy (int) – The strategy to handle error handling. 0: Exit immediately, 1: Exit after exceeding the error count.

• file_format (str) – The file format of the trace dataset. “zarr” or “numpy”.

• file_path (str) – The file path of the trace dataset. If set to “auto”, a folder with a timestamp format will be created in the current working directory to save the data.

property sample_length

property sample_offset

set_cracker(cracker)

stop()

Stop test or run mode.

test(sample_length=None, sample_offset=None, trigger_judge_wait_time=None, trigger_judge_timeout=None, do_error_max_count=None, do_error_handler_strategy=None, trace_fetch_interval=2.0)

Start test mode in background.

Parameters:

• count (int) – The number of traces to acquire. normally this value should be -1.

• sample_length (int) – The sample length to be acquired.

• sample_offset (int) – The sample offset to be acquired.

• trigger_judge_wait_time (float) – The trigger judge wait time in seconds.

• trigger_judge_timeout (float) – The trigger judge timeout in seconds.

• do_error_max_count (int) – The maximum number of error count. If do_error_handler_strategy is 0, this setting is invalid.

• do_error_handler_strategy (int | None) – The strategy to handle error handling. 0: Exit immediately, 1: Exit after exceeding the error count.

test_sync(count=-1, sample_length=None, sample_offset=None, trigger_judge_wait_time=None, trigger_judge_timeout=None, do_error_max_count=None, do_error_handler_strategy=None)

Start test mode in foreground, which will cause blocking.

Parameters:

• count (int) – The number of traces to acquire. normally this value should be -1.

• sample_length (int) – The sample length to be acquired.

• sample_offset (int) – The sample offset to be acquired.

• trigger_judge_wait_time (float) – The trigger judge wait time in seconds.

• trigger_judge_timeout (float) – The trigger judge timeout in seconds.

• do_error_max_count (int) – The maximum number of error count. If do_error_handler_strategy is 0, this setting is invalid.

• do_error_handler_strategy (int | None) – The strategy to handle error handling. 0: Exit immediately, 1: Exit after exceeding the error count.

property trace_count

property trace_fetch_interval

transfer()

property trigger_judge_timeout

property trigger_judge_wait_time