Acquisition

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.05, trigger_judge_timeout=1.0, do_error_handler_strategy=0, do_error_max_count=-1, file_format='scarr', 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

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_RUNNING': <class 'int'>, 'STATUS_STOPPED': <class 'int'>, 'STATUS_TESTING': <class 'int'>, '_STATUS_PAUSED_SWITCH': <class 'int'>, '_last_wave': 'dict[int, np.ndarray] | 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', '_status': 'int', 'cracker': 'CrackerBasic', 'do_error_handler_strategy': 'int', 'do_error_max_count': 'int', 'file_format': 'str', 'file_path': 'str', 'metadata_ciphertext_length': 'int', 'metadata_extended_length': 'int', 'metadata_key_length': 'int', 'metadata_plaintext_length': 'int', 'sample_offset': 'int', 'trace_count': 'int', 'trigger_judge_timeout': 'float', 'trigger_judge_wait_time': 'float'}

__dict__ = mappingproxy({'__module__': 'cracknuts.acquisition.acquisition', '__annotations__': {'STATUS_STOPPED': <class 'int'>, 'STATUS_TESTING': <class 'int'>, 'STATUS_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]]'}, '__doc__': '\n    Trace acquisition process control class\n    ', 'STATUS_STOPPED': 0, 'STATUS_TESTING': 1, 'STATUS_RUNNING': 2, '_STATUS_PAUSED_SWITCH': -1, 'DO_ERROR_HANDLER_STRATEGY_EXIT': 0, 'DO_ERROR_HANDLER_STRATEGY_CONTINUE_UNTIL_MAX_ERROR_COUNT': 1, '_DATASET_DEFAULT_PATH': './dataset/', '__init__': <function Acquisition.__init__>, '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>, 'builder': <staticmethod(<function Acquisition.builder>)>, '__dict__': <attribute '__dict__' of 'Acquisition' objects>, '__weakref__': <attribute '__weakref__' of 'Acquisition' objects>, '__abstractmethods__': frozenset({'init', 'do'}), '_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.05, trigger_judge_timeout=1.0, do_error_handler_strategy=0, do_error_max_count=-1, file_format='scarr', 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. “scarr” 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

static builder()

config_cracker()

config_scrat()

connect_cracker()

Connect to cracker device. :return:

connect_net()

connect_scrat()

Connect to scrat device :return:

abstractmethod do()

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

Returns:

The data in dictionary format to be saved should contain the ciphertext and plaintext keys,

and may optionally include the 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. The data is structured as follows: {

”plaintext”: plaintext_data_bytes, “ciphertext”: ciphertext_data_bytes, “key”: key_data_bytes, “extended”: extended_data_bytes,

}

Return type:

dict[str, bytes]

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) – the path to the JSON file

Return type:

str | None

Returns:

the content of JSON string or None if no path is specified.

get_last_wave()

Get the last wave.

Return type:

dict[int, 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

load_config_from_json(config)

Load config from a JSON object.

Parameters:

config (dict) – the JSON object

load_config_from_str(json_str)

Load config from a JSON string.

Parameters:

json_str (str) – the JSON string

Return type:

None

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: 0 stopped, 1 testing, 2 running 3 paused

Return type:

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=1, sample_length=1024, sample_offset=0, 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='scarr', file_path='auto')

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. “scarr” 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='scarr', 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. “scarr” 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.

set_cracker(cracker)

stop()

Stop test or run mode.

test(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, 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.

transfer()