diff --git a/src/lava/utils/serialization.py b/src/lava/utils/serialization.py index 246e64550..9ac6dc871 100644 --- a/src/lava/utils/serialization.py +++ b/src/lava/utils/serialization.py @@ -2,7 +2,7 @@ # SPDX-License-Identifier: BSD-3-Clause # See: https://spdx.org/licenses/ -import pickle +import pickle # noqa: S403 import typing as ty import os @@ -13,7 +13,7 @@ class SerializationObject: """This class is used to serialize a process or a list of processes together with a corresponding executable. - + Parameters ---------- processes: AbstractProcess, ty.List[AbstractProcess] @@ -36,7 +36,7 @@ def save(processes: ty.Union[AbstractProcess, ty.List[AbstractProcess]], executable: ty.Optional[Executable] = None) -> None: """Saves a given process or list of processes with an (optional) corresponding executable to file . - + Parameters ---------- processes: AbstractProcess, ty.List[AbstractProcess] @@ -47,7 +47,6 @@ def save(processes: ty.Union[AbstractProcess, ty.List[AbstractProcess]], executable: Executable, optional The corresponding executable of the compiled processes which should be stored in a file. - """ # Check parameter types if not isinstance(processes, list) and not isinstance(processes, @@ -58,15 +57,15 @@ def save(processes: ty.Union[AbstractProcess, ty.List[AbstractProcess]], if not isinstance(filename, str): raise AssertionError(f"Parameter must be string" f" but got {filename}.") - if not executable is None and not isinstance(executable, Executable): + if executable is not None and not isinstance(executable, Executable): raise AssertionError(f"Parameter must be Executable" f" but got {executable}.") # Create object which is stored obj = SerializationObject(processes, executable) - # Add default file extension if no extension is present - if not "." in filename: + # Add default file extension if no extension is present + if "." not in filename: filename = filename + ".pickle" # Store object at @@ -79,30 +78,32 @@ def load(filename: str) -> ty.Tuple[ty.Union[AbstractProcess, ty.Union[None, Executable]]: """Loads a process or list of processes with an (optional) corresponding executable from file . - + Parameters ---------- filename: str The path + name of the file. If no file extension is given, '.pickle' will be added automatically. - + Returns ------- tuple - Returns a tuple of a process or list of processes and a executable or None. + Returns a tuple of a process or list of processes and a executable or + None. """ # Check parameter types if not isinstance(filename, str): raise AssertionError(f"Parameter must be string" f" but got {filename}.") + # Check if filename exists if not os.path.isfile(filename): raise AssertionError(f"File {filename} could not be found.") # Load serialized object from with open(filename, 'rb') as f: - obj = pickle.load(f) + obj = pickle.load(f) # noqa: S301 # Check loaded object if not isinstance(obj, SerializationObject): diff --git a/tests/lava/utils/test_serialization.py b/tests/lava/utils/test_serialization.py index 7e9398460..8f6f2c848 100644 --- a/tests/lava/utils/test_serialization.py +++ b/tests/lava/utils/test_serialization.py @@ -16,7 +16,7 @@ from lava.magma.core.process.variable import Var -# A minimal hierarchical process +# A minimal hierarchical process class HP(AbstractProcess): def __init__(self, **kwargs): super().__init__(**kwargs) @@ -30,7 +30,7 @@ class PyProcModelHP(AbstractSubProcessModel): def __init__(self, proc): """Builds sub Process structure of the Process.""" - + pre_size = 2 post_size = 3 @@ -38,7 +38,7 @@ def __init__(self, proc): self.lif_in = LIF(shape=(pre_size,), bias_mant=100, vth=120, name="LIF_neuron input") - self.dense = Dense(weights = weights * 10, name="Dense") + self.dense = Dense(weights=weights * 10, name="Dense") self.lif_out = LIF(shape=(post_size,), bias_mant=0, vth=50000, name="LIF_neuron output") @@ -83,7 +83,7 @@ def test_save_load_processes(self): lif_procs = [] for i in range(5): lif_procs.append(LIF(shape=(1,), name="LIF" + str(i))) - + # Store the processes in file test.pickle with tempfile.TemporaryDirectory() as tmpdirname: save(lif_procs + [dense], tmpdirname + "test") @@ -101,7 +101,7 @@ def test_save_load_processes(self): for i in range(5): self.assertTrue(isinstance(procs[i], LIF)) self.assertTrue(procs[i].name == "LIF" + str(i)) - + def test_save_load_executable(self): """Checks storing and loading of executable.""" @@ -123,7 +123,6 @@ def test_save_load_executable(self): loaded_lif = exec.process_list[0] self.assertTrue(lif.name == loaded_lif.name) - def test_save_load_hierarchical_proc(self): """Checks saving, loading and execution of a workload using a hierarchical process.""" @@ -161,11 +160,11 @@ def test_save_load_hierarchical_proc(self): # Run the loaded executable proc_loaded.create_runtime(executable=ex_loaded) try: - for i in range(num_steps): - proc_loaded.run(condition=RunSteps(num_steps=1)) + for i in range(num_steps): + proc_loaded.run(condition=RunSteps(num_steps=1)) - output_lif_in_v_loaded[:, i] = proc_loaded.lif_in_v.get() - output_lif_out_u_loaded[:, i] = proc_loaded.lif_out_u.get() + output_lif_in_v_loaded[:, i] = proc_loaded.lif_in_v.get() + output_lif_out_u_loaded[:, i] = proc_loaded.lif_out_u.get() finally: proc_loaded.stop()