14359872492

Committed 09 Apr 2025 02:33PM UTC coverage: 97.602% (+0.6%) from 97.035%

Build # 14359872492

Build Type

Pull #86

github

Committed by

web-flow

Commit Message

Merge ec719935a into e9c323c4f

Pull Request Pull Request #86: Smallclasses

Run Details

2291 of 2418 new or added lines in 54 files covered. (94.75%)

3 existing lines in 1 file now uncovered.

11683 of 11970 relevant lines covered (97.6%)

0.98 hits per line

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

99.44

/nnodely/operators/network.py

import copy, torch

import numpy as np

from nnodely.basic.modeldef import ModelDef
from nnodely.basic.model import Model
from nnodely.support.utils import check, log, TORCH_DTYPE, NP_DTYPE, argmax_dict, argmin_dict, enforce_types
from nnodely.basic.relation import Stream
from nnodely.layers.input import State
from nnodely.layers.output import Output

class Network():
    def __init__(self):
        check(type(self) is not Network, TypeError, "Network class cannot be instantiated directly")

    def __addInfo(self):
        total_params = sum(p.numel() for p in self._model.parameters() if p.requires_grad)
        self._model_def['Info']['num_parameters'] = total_params
        from nnodely import __version__
        self._model_def['Info']['nnodely_version'] = __version__

    @enforce_types
    def addModel(self, name:str, stream_list:list|Output|Stream) -> None:
        """
        Adds a new model with the given name along with a list of Outputs.

        Parameters
        ----------
        name : str
            The name of the model.
        stream_list : list of Stream
            The list of Outputs stream in the model.

        Example
        -------
        Example usage:
            >>> model = Modely()
            >>> x = Input('x')
            >>> out = Output('out', Fir(x.last()))
            >>> model.addModel('example_model', [out])
        """
        try:
            self._model_def.addModel(name, stream_list)
        except Exception as e:
            self._model_def.removeModel(name)
            raise e

    @enforce_types
    def removeModel(self, name_list:list) -> None:
        """
        Removes models with the given list of names.

        Parameters
        ----------
        name_list : list of str
            The list of model names to remove.

        Example
        -------
        Example usage:
            >>> model.removeModel(['sub_model1', 'sub_model2'])
        """
        self._model_def.removeModel(name_list)

    @enforce_types
    def addConnect(self, stream_out:Output|Stream, state_list_in:State) -> None:
        """
        Adds a connection from a relation stream to an input state.

        Parameters
        ----------
        stream_out : Stream
            The relation stream to connect from.
        state_list_in : State
            The states to connect to.

        Examples
        --------
        .. image:: https://colab.research.google.com/assets/colab-badge.svg
            :target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
            :alt: Open in Colab

        Example:
            >>> model = Modely()
            >>> x = Input('x')
            >>> y = State('y')
            >>> relation = Fir(x.last())
            >>> model.addConnect(relation, y)
        """
        self._model_def.addConnect(stream_out, state_list_in)

    @enforce_types
    def addClosedLoop(self, stream_out:Output|Stream, state_list_in:State) -> None:
        """
        Adds a closed loop connection from a relation stream to an input state.

        Parameters
        ----------
        stream_out : Stream
            The relation stream to connect from.
        state_list_in : list of State
            The list of input states to connect to.

        Examples
        --------
        .. image:: https://colab.research.google.com/assets/colab-badge.svg
            :target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
            :alt: Open in Colab

        Example:
            >>> model = Modely()
            >>> x = Input('x')
            >>> y = State('y')
            >>> relation = Fir(x.last())
            >>> model.addClosedLoop(relation, y)
        """
        self._model_def.addClosedLoop(stream_out, state_list_in)

    @enforce_types
    def neuralizeModel(self, sample_time:float|int|None = None, clear_model:bool = False, model_def:dict|None = None) -> None:
        """
        Neuralizes the model, preparing it for inference and training. This method creates a neural network model starting from the model definition.
        It will also create all the time windows for the inputs and states.

        Parameters
        ----------
        sample_time : float or None, optional
            The sample time for the model. Default is None.
        clear_model : bool, optional
            Whether to clear the existing model definition. Default is False.
        model_def : dict or None, optional
            A dictionary defining the model. If provided, it overrides the existing model definition. Default is None.

        Raises
        ------
        ValueError
            If sample_time is not None and model_def is provided.
            If clear_model is True and model_def is provided.

        Example
        -------
        Example usage:
            >>> model = Modely(name='example_model')
            >>> model.neuralizeModel(sample_time=0.1, clear_model=True)
        """
        if model_def is not None:
            check(sample_time == None, ValueError, 'The sample_time must be None if a model_def is provided')
            check(clear_model == False, ValueError, 'The clear_model must be False if a model_def is provided')
            self._model_def = ModelDef(model_def)
        else:
            if clear_model:
                self._model_def.update()
            else:
                self._model_def.updateParameters(self._model)

        for key, state in self._model_def['States'].items():
            check("connect" in state.keys() or  'closedLoop' in state.keys(), KeyError, f'The connect or closed loop missing for state "{key}"')

        self._model_def.setBuildWindow(sample_time)
        self._model = Model(self._model_def.getJson())
        self.__addInfo()

        self._input_ns_backward = {key:value['ns'][0] for key, value in (self._model_def['Inputs']|self._model_def['States']).items()}
        self._input_ns_forward = {key:value['ns'][1] for key, value in (self._model_def['Inputs']|self._model_def['States']).items()}
        self._max_samples_backward = max(self._input_ns_backward.values())
        self._max_samples_forward = max(self._input_ns_forward.values())
        self._input_n_samples = {}
        for key, value in (self._model_def['Inputs'] | self._model_def['States']).items():
            self._input_n_samples[key] = self._input_ns_backward[key] + self._input_ns_forward[key]
        self._max_n_samples = max(self._input_ns_backward.values()) + max(self._input_ns_forward.values())

        ## Initialize States
        self.resetStates()

        self._neuralized = True
        self._traced = False
        self.visualizer.showModel(self._model_def.getJson())
        self.visualizer.showModelInputWindow()
        self.visualizer.showBuiltModel()

    @enforce_types
    def __call__(self, inputs:dict={}, sampled:bool=False, closed_loop:dict={}, connect:dict={}, prediction_samples:str|int|None='auto',
                 num_of_samples:int|None=None) -> dict:  ##, align_input=False):
        """
        Performs inference on the model.

        Parameters
        ----------
        inputs : dict, optional
            A dictionary of input data. The keys are input names and the values are the corresponding data. Default is an empty dictionary.
        sampled : bool, optional
            A boolean indicating whether the inputs are already sampled. Default is False.
        closed_loop : dict, optional
            A dictionary specifying closed loop connections. The keys are input names and the values are output names. Default is an empty dictionary.
        connect : dict, optional
            A dictionary specifying connections. The keys are input names and the values are output names. Default is an empty dictionary.
        prediction_samples : str or int, optional
            The number of prediction samples. Can be 'auto', None or an integer. Default is 'auto'.
        num_of_samples : str or int, optional
            The number of samples. Can be 'auto', None or an integer. Default is 'auto'.

        Returns
        -------
        dict
            A dictionary containing the model's prediction outputs.

        Raises
        ------
        RuntimeError
            If the network is not neuralized.
        ValueError
            If an input variable is not in the model definition or if an output variable is not in the model definition.

        Examples
        --------
        .. image:: https://colab.research.google.com/assets/colab-badge.svg
            :target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/inference.ipynb
            :alt: Open in Colab

        Example usage:
            >>> model = Modely()
            >>> x = Input('x')
            >>> out = Output('out', Fir(x.last()))
            >>> model.addModel('example_model', [out])
            >>> model.neuralizeModel()
            >>> predictions = model(inputs={'x': [1, 2, 3]})
        """

        ## Copy dict for avoid python bug
        inputs = copy.deepcopy(inputs)
        closed_loop = copy.deepcopy(closed_loop)
        connect = copy.deepcopy(connect)

        ## Check neuralize
        check(self.neuralized, RuntimeError, "The network is not neuralized.")

        ## Check closed loop integrity
        for close_in, close_out in (closed_loop | connect).items():
            check(close_in in self._model_def['Inputs'], ValueError, f'the tag "{close_in}" is not an input variable.')
            check(close_out in self._model_def['Outputs'], ValueError,
                  f'the tag "{close_out}" is not an output of the network')

        ## List of keys
        model_inputs = list(self._model_def['Inputs'].keys())
        model_states = list(self._model_def['States'].keys())
        json_inputs = self._model_def['Inputs'] | self._model_def['States']
        state_closed_loop = [key for key, value in self._model_def['States'].items() if
                             'closedLoop' in value.keys()] + list(closed_loop.keys())
        state_connect = [key for key, value in self._model_def['States'].items() if 'connect' in value.keys()] + list(
            connect.keys())
        extra_inputs = list(set(list(inputs.keys())) - set(model_inputs) - set(model_states))
        non_mandatory_inputs = state_closed_loop + state_connect
        mandatory_inputs = list(set(model_inputs) - set(non_mandatory_inputs))

        ## Remove extra inputs
        for key in extra_inputs:
            log.warning(
                f'The provided input {key} is not used inside the network. the inference will continue without using it')
            del inputs[key]

        ## Get the number of data windows for each input/state
        num_of_windows = {key: len(value) for key, value in inputs.items()} if sampled else {
            key: len(value) - self._input_n_samples[key] + 1 for key, value in inputs.items()}

        ## Get the maximum inference window
        if num_of_samples:
            window_dim = num_of_samples
            for key in inputs.keys():
                input_dim = self._model_def['Inputs'][key]['dim'] if key in model_inputs else \
                self._model_def['States'][key]['dim']
                new_samples = num_of_samples - (len(inputs[key]) - self._input_n_samples[key] + 1)
                if input_dim > 1:
                    log.warning(f'The variable {key} is filled with {new_samples} samples equal to zeros.')
                    inputs[key] += [[0 for _ in range(input_dim)] for _ in range(new_samples)]
                else:
                    log.warning(f'The variable {key} is filled with {new_samples} samples equal to zeros.')
                    inputs[key] += [0 for _ in range(new_samples)]
        elif inputs:
            windows = []
            for key in inputs.keys():
                if key in mandatory_inputs:
                    n_samples = len(inputs[key]) if sampled else len(inputs[key]) - self._model_def['Inputs'][key][
                        'ntot'] + 1
                    windows.append(n_samples)
            if not windows:
                for key in inputs.keys():
                    if key in non_mandatory_inputs:
                        if key in model_inputs:
                            n_samples = len(inputs[key]) if sampled else len(inputs[key]) - \
                                                                         self._model_def['Inputs'][key]['ntot'] + 1
                        else:
                            n_samples = len(inputs[key]) if sampled else len(inputs[key]) - \
                                                                         self._model_def['States'][key]['ntot'] + 1
                        windows.append(n_samples)
            window_dim = min(windows) if windows else 0
        else:  ## No inputs
            window_dim = 1 if non_mandatory_inputs else 0
        check(window_dim > 0, StopIteration, f'Missing samples in the input window')

        if len(set(num_of_windows.values())) > 1:
            max_ind_key, max_dim = argmax_dict(num_of_windows)
            min_ind_key, min_dim = argmin_dict(num_of_windows)
            log.warning(
                f'Different number of samples between inputs [MAX {num_of_windows[max_ind_key]} = {max_dim}; MIN {num_of_windows[min_ind_key]} = {min_dim}]')

        ## Autofill the missing inputs
        provided_inputs = list(inputs.keys())
        missing_inputs = list(set(mandatory_inputs) - set(provided_inputs))
        if missing_inputs:
            log.warning(f'Inputs not provided: {missing_inputs}. Autofilling with zeros..')
            for key in missing_inputs:
                inputs[key] = np.zeros(
                    shape=(self._input_n_samples[key] + window_dim - 1, self._model_def['Inputs'][key]['dim']),
                    dtype=NP_DTYPE).tolist()

        ## Transform inputs in 3D Tensors
        for key in inputs.keys():
            input_dim = json_inputs[key]['dim']
            inputs[key] = torch.from_numpy(np.array(inputs[key])).to(TORCH_DTYPE)

            if input_dim > 1:
                correct_dim = 3 if sampled else 2
                check(len(inputs[key].shape) == correct_dim, ValueError,
                      f'The input {key} must have {correct_dim} dimensions')
                check(inputs[key].shape[correct_dim - 1] == input_dim, ValueError,
                      f'The second dimension of the input "{key}" must be equal to {input_dim}')

            if input_dim == 1 and inputs[key].shape[-1] != 1:  ## add the input dimension
                inputs[key] = inputs[key].unsqueeze(-1)
            if inputs[key].ndim <= 1:  ## add the batch dimension
                inputs[key] = inputs[key].unsqueeze(0)
            if inputs[key].ndim <= 2:  ## add the time dimension
                inputs[key] = inputs[key].unsqueeze(0)

        ## initialize the resulting dictionary
        result_dict = {}
        for key in self._model_def['Outputs'].keys():
            result_dict[key] = []

        ## Inference
        calculate_grad = False
        for key, value in json_inputs.items():
            if 'type' in value.keys():
                calculate_grad = True
                break
        with torch.enable_grad() if calculate_grad else torch.inference_mode():
            ## Update with virtual states
            if prediction_samples is not None:
                self._model.update(closed_loop=closed_loop, connect=connect)
            else:
                prediction_samples = 0
            X = {}
            count = 0
            first = True
            for idx in range(window_dim):
                ## Get mandatory data inputs
                for key in mandatory_inputs:
                    X[key] = inputs[key][idx:idx + 1] if sampled else inputs[key][:,
                                                                      idx:idx + self._input_n_samples[key]]
                    if 'type' in json_inputs[key].keys():
                        X[key] = X[key].requires_grad_(True)
                ## reset states
                if count == 0 or prediction_samples == 'auto':
                    count = prediction_samples
                    for key in non_mandatory_inputs:  ## Get non mandatory data (from inputs, from states, or with zeros)
                        ## if prediction_samples is 'auto' and i have enough samples
                        ## if prediction_samples is NOT 'auto' but i have enough extended window (with zeros)
                        if (key in inputs.keys() and prediction_samples == 'auto' and idx < num_of_windows[key]) or (
                                key in inputs.keys() and prediction_samples != 'auto' and idx < inputs[key].shape[1]):
                            X[key] = inputs[key][idx:idx + 1] if sampled else inputs[key][:,
                                                                              idx:idx + self._input_n_samples[key]]
                        ## if im in the first reset
                        ## if i have a state in memory
                        ## if i have prediction_samples = 'auto' and not enough samples
                        elif (key in self._states.keys() and (first or prediction_samples == 'auto')) and (
                                prediction_samples == 'auto' or prediction_samples == None):
                            X[key] = self._states[key]
                        else:  ## if i have no samples and no states
                            window_size = self._input_n_samples[key]
                            dim = json_inputs[key]['dim']
                            X[key] = torch.zeros(size=(1, window_size, dim), dtype=TORCH_DTYPE, requires_grad=False)
                            self._states[key] = X[key]
                        if 'type' in json_inputs[key].keys():
                            X[key] = X[key].requires_grad_(True)
                    first = False
                else:
                    # Remove the gradient of the previous forward
                    for key in X.keys():
                        if 'type' in json_inputs[key].keys():
                            X[key] = X[key].detach().requires_grad_(True)
                    count -= 1
                ## Forward pass
                result, _, out_closed_loop, out_connect = self._model(X)

                ## Append the prediction of the current sample to the result dictionary
                for key in self._model_def['Outputs'].keys():
                    if result[key].shape[-1] == 1:
                        result[key] = result[key].squeeze(-1)
                        if result[key].shape[-1] == 1:
                            result[key] = result[key].squeeze(-1)
                    result_dict[key].append(result[key].detach().squeeze(dim=0).tolist())

                ## Update closed_loop and connect
                if prediction_samples:
                    self._updateState(X, out_closed_loop, out_connect)

        ## Remove virtual states
        self._removeVirtualStates(connect, closed_loop)

        return result_dict



1	import copy, torch	1✔
2
3	import numpy as np	1✔
4
5	from nnodely.basic.modeldef import ModelDef	1✔
6	from nnodely.basic.model import Model	1✔
7	from nnodely.support.utils import check, log, TORCH_DTYPE, NP_DTYPE, argmax_dict, argmin_dict, enforce_types	1✔
8	from nnodely.basic.relation import Stream	1✔
9	from nnodely.layers.input import State	1✔
10	from nnodely.layers.output import Output	1✔
11
12	class Network():	1✔
13	def __init__(self):	1✔
14	check(type(self) is not Network, TypeError, "Network class cannot be instantiated directly")	1✔
15
16	def __addInfo(self):	1✔
17	total_params = sum(p.numel() for p in self._model.parameters() if p.requires_grad)	1✔
18	self._model_def['Info']['num_parameters'] = total_params	1✔
19	from nnodely import __version__	1✔
20	self._model_def['Info']['nnodely_version'] = __version__	1✔
21
22	@enforce_types	1✔
23	def addModel(self, name:str, stream_list:list\|Output\|Stream) -> None:	1✔
24	"""
25	Adds a new model with the given name along with a list of Outputs.
26
27	Parameters
28	----------
29	name : str
30	The name of the model.
31	stream_list : list of Stream
32	The list of Outputs stream in the model.
33
34	Example
35	-------
36	Example usage:
37	>>> model = Modely()
38	>>> x = Input('x')
39	>>> out = Output('out', Fir(x.last()))
40	>>> model.addModel('example_model', [out])
41	"""
42	try:	1✔
43	self._model_def.addModel(name, stream_list)	1✔
44	except Exception as e:	1✔
45	self._model_def.removeModel(name)	1✔
46	raise e	1✔
47
48	@enforce_types	1✔
49	def removeModel(self, name_list:list) -> None:	1✔
50	"""
51	Removes models with the given list of names.
52
53	Parameters
54	----------
55	name_list : list of str
56	The list of model names to remove.
57
58	Example
59	-------
60	Example usage:
61	>>> model.removeModel(['sub_model1', 'sub_model2'])
62	"""
NEW 63	self._model_def.removeModel(name_list)	×
64
65	@enforce_types	1✔
66	def addConnect(self, stream_out:Output\|Stream, state_list_in:State) -> None:	1✔
67	"""
68	Adds a connection from a relation stream to an input state.
69
70	Parameters
71	----------
72	stream_out : Stream
73	The relation stream to connect from.
74	state_list_in : State
75	The states to connect to.
76
77	Examples
78	--------
79	.. image:: https://colab.research.google.com/assets/colab-badge.svg
80	:target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
81	:alt: Open in Colab
82
83	Example:
84	>>> model = Modely()
85	>>> x = Input('x')
86	>>> y = State('y')
87	>>> relation = Fir(x.last())
88	>>> model.addConnect(relation, y)
89	"""
90	self._model_def.addConnect(stream_out, state_list_in)	1✔
91
92	@enforce_types	1✔
93	def addClosedLoop(self, stream_out:Output\|Stream, state_list_in:State) -> None:	1✔
94	"""
95	Adds a closed loop connection from a relation stream to an input state.
96
97	Parameters
98	----------
99	stream_out : Stream
100	The relation stream to connect from.
101	state_list_in : list of State
102	The list of input states to connect to.
103
104	Examples
105	--------
106	.. image:: https://colab.research.google.com/assets/colab-badge.svg
107	:target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/states.ipynb
108	:alt: Open in Colab
109
110	Example:
111	>>> model = Modely()
112	>>> x = Input('x')
113	>>> y = State('y')
114	>>> relation = Fir(x.last())
115	>>> model.addClosedLoop(relation, y)
116	"""
117	self._model_def.addClosedLoop(stream_out, state_list_in)	1✔
118
119	@enforce_types	1✔
120	def neuralizeModel(self, sample_time:float\|int\|None = None, clear_model:bool = False, model_def:dict\|None = None) -> None:	1✔
121	"""
122	Neuralizes the model, preparing it for inference and training. This method creates a neural network model starting from the model definition.
123	It will also create all the time windows for the inputs and states.
124
125	Parameters
126	----------
127	sample_time : float or None, optional
128	The sample time for the model. Default is None.
129	clear_model : bool, optional
130	Whether to clear the existing model definition. Default is False.
131	model_def : dict or None, optional
132	A dictionary defining the model. If provided, it overrides the existing model definition. Default is None.
133
134	Raises
135	------
136	ValueError
137	If sample_time is not None and model_def is provided.
138	If clear_model is True and model_def is provided.
139
140	Example
141	-------
142	Example usage:
143	>>> model = Modely(name='example_model')
144	>>> model.neuralizeModel(sample_time=0.1, clear_model=True)
145	"""
146	if model_def is not None:	1✔
147	check(sample_time == None, ValueError, 'The sample_time must be None if a model_def is provided')	1✔
148	check(clear_model == False, ValueError, 'The clear_model must be False if a model_def is provided')	1✔
149	self._model_def = ModelDef(model_def)	1✔
150	else:
151	if clear_model:	1✔
152	self._model_def.update()	1✔
153	else:
154	self._model_def.updateParameters(self._model)	1✔
155
156	for key, state in self._model_def['States'].items():	1✔
157	check("connect" in state.keys() or 'closedLoop' in state.keys(), KeyError, f'The connect or closed loop missing for state "{key}"')	1✔
158
159	self._model_def.setBuildWindow(sample_time)	1✔
160	self._model = Model(self._model_def.getJson())	1✔
161	self.__addInfo()	1✔
162
163	self._input_ns_backward = {key:value['ns'][0] for key, value in (self._model_def['Inputs']\|self._model_def['States']).items()}	1✔
164	self._input_ns_forward = {key:value['ns'][1] for key, value in (self._model_def['Inputs']\|self._model_def['States']).items()}	1✔
165	self._max_samples_backward = max(self._input_ns_backward.values())	1✔
166	self._max_samples_forward = max(self._input_ns_forward.values())	1✔
167	self._input_n_samples = {}	1✔
168	for key, value in (self._model_def['Inputs'] \| self._model_def['States']).items():	1✔
169	self._input_n_samples[key] = self._input_ns_backward[key] + self._input_ns_forward[key]	1✔
170	self._max_n_samples = max(self._input_ns_backward.values()) + max(self._input_ns_forward.values())	1✔
171
172	## Initialize States
173	self.resetStates()	1✔
174
175	self._neuralized = True	1✔
176	self._traced = False	1✔
177	self.visualizer.showModel(self._model_def.getJson())	1✔
178	self.visualizer.showModelInputWindow()	1✔
179	self.visualizer.showBuiltModel()	1✔
180
181	@enforce_types	1✔
182	def __call__(self, inputs:dict={}, sampled:bool=False, closed_loop:dict={}, connect:dict={}, prediction_samples:str\|int\|None='auto',	1✔
183	num_of_samples:int\|None=None) -> dict: ##, align_input=False):
184	"""
185	Performs inference on the model.
186
187	Parameters
188	----------
189	inputs : dict, optional
190	A dictionary of input data. The keys are input names and the values are the corresponding data. Default is an empty dictionary.
191	sampled : bool, optional
192	A boolean indicating whether the inputs are already sampled. Default is False.
193	closed_loop : dict, optional
194	A dictionary specifying closed loop connections. The keys are input names and the values are output names. Default is an empty dictionary.
195	connect : dict, optional
196	A dictionary specifying connections. The keys are input names and the values are output names. Default is an empty dictionary.
197	prediction_samples : str or int, optional
198	The number of prediction samples. Can be 'auto', None or an integer. Default is 'auto'.
199	num_of_samples : str or int, optional
200	The number of samples. Can be 'auto', None or an integer. Default is 'auto'.
201
202	Returns
203	-------
204	dict
205	A dictionary containing the model's prediction outputs.
206
207	Raises
208	------
209	RuntimeError
210	If the network is not neuralized.
211	ValueError
212	If an input variable is not in the model definition or if an output variable is not in the model definition.
213
214	Examples
215	--------
216	.. image:: https://colab.research.google.com/assets/colab-badge.svg
217	:target: https://colab.research.google.com/github/tonegas/nnodely/blob/main/examples/inference.ipynb
218	:alt: Open in Colab
219
220	Example usage:
221	>>> model = Modely()
222	>>> x = Input('x')
223	>>> out = Output('out', Fir(x.last()))
224	>>> model.addModel('example_model', [out])
225	>>> model.neuralizeModel()
226	>>> predictions = model(inputs={'x': [1, 2, 3]})
227	"""
228
229	## Copy dict for avoid python bug
230	inputs = copy.deepcopy(inputs)	1✔
231	closed_loop = copy.deepcopy(closed_loop)	1✔
232	connect = copy.deepcopy(connect)	1✔
233
234	## Check neuralize
235	check(self.neuralized, RuntimeError, "The network is not neuralized.")	1✔
236
237	## Check closed loop integrity
238	for close_in, close_out in (closed_loop \| connect).items():	1✔
239	check(close_in in self._model_def['Inputs'], ValueError, f'the tag "{close_in}" is not an input variable.')	1✔
240	check(close_out in self._model_def['Outputs'], ValueError,	1✔
241	f'the tag "{close_out}" is not an output of the network')
242
243	## List of keys
244	model_inputs = list(self._model_def['Inputs'].keys())	1✔
245	model_states = list(self._model_def['States'].keys())	1✔
246	json_inputs = self._model_def['Inputs'] \| self._model_def['States']	1✔
247	state_closed_loop = [key for key, value in self._model_def['States'].items() if	1✔
248	'closedLoop' in value.keys()] + list(closed_loop.keys())
249	state_connect = [key for key, value in self._model_def['States'].items() if 'connect' in value.keys()] + list(	1✔
250	connect.keys())
251	extra_inputs = list(set(list(inputs.keys())) - set(model_inputs) - set(model_states))	1✔
252	non_mandatory_inputs = state_closed_loop + state_connect	1✔
253	mandatory_inputs = list(set(model_inputs) - set(non_mandatory_inputs))	1✔
254
255	## Remove extra inputs
256	for key in extra_inputs:	1✔
257	log.warning(	1✔
258	f'The provided input {key} is not used inside the network. the inference will continue without using it')
259	del inputs[key]	1✔
260
261	## Get the number of data windows for each input/state
262	num_of_windows = {key: len(value) for key, value in inputs.items()} if sampled else {	1✔
263	key: len(value) - self._input_n_samples[key] + 1 for key, value in inputs.items()}
264
265	## Get the maximum inference window
266	if num_of_samples:	1✔
267	window_dim = num_of_samples	1✔
268	for key in inputs.keys():	1✔
269	input_dim = self._model_def['Inputs'][key]['dim'] if key in model_inputs else \	1✔
270	self._model_def['States'][key]['dim']
271	new_samples = num_of_samples - (len(inputs[key]) - self._input_n_samples[key] + 1)	1✔
272	if input_dim > 1:	1✔
273	log.warning(f'The variable {key} is filled with {new_samples} samples equal to zeros.')	1✔
274	inputs[key] += [[0 for _ in range(input_dim)] for _ in range(new_samples)]	1✔
275	else:
276	log.warning(f'The variable {key} is filled with {new_samples} samples equal to zeros.')	1✔
277	inputs[key] += [0 for _ in range(new_samples)]	1✔
278	elif inputs:	1✔
279	windows = []	1✔
280	for key in inputs.keys():	1✔
281	if key in mandatory_inputs:	1✔
282	n_samples = len(inputs[key]) if sampled else len(inputs[key]) - self._model_def['Inputs'][key][	1✔
283	'ntot'] + 1
284	windows.append(n_samples)	1✔
285	if not windows:	1✔
286	for key in inputs.keys():	1✔
287	if key in non_mandatory_inputs:	1✔
288	if key in model_inputs:	1✔
289	n_samples = len(inputs[key]) if sampled else len(inputs[key]) - \	1✔
290	self._model_def['Inputs'][key]['ntot'] + 1
291	else:
292	n_samples = len(inputs[key]) if sampled else len(inputs[key]) - \	1✔
293	self._model_def['States'][key]['ntot'] + 1
294	windows.append(n_samples)	1✔
295	window_dim = min(windows) if windows else 0	1✔
296	else: ## No inputs
297	window_dim = 1 if non_mandatory_inputs else 0	1✔
298	check(window_dim > 0, StopIteration, f'Missing samples in the input window')	1✔
299
300	if len(set(num_of_windows.values())) > 1:	1✔
301	max_ind_key, max_dim = argmax_dict(num_of_windows)	1✔
302	min_ind_key, min_dim = argmin_dict(num_of_windows)	1✔
303	log.warning(	1✔
304	f'Different number of samples between inputs [MAX {num_of_windows[max_ind_key]} = {max_dim}; MIN {num_of_windows[min_ind_key]} = {min_dim}]')
305
306	## Autofill the missing inputs
307	provided_inputs = list(inputs.keys())	1✔
308	missing_inputs = list(set(mandatory_inputs) - set(provided_inputs))	1✔
309	if missing_inputs:	1✔
310	log.warning(f'Inputs not provided: {missing_inputs}. Autofilling with zeros..')	1✔
311	for key in missing_inputs:	1✔
312	inputs[key] = np.zeros(	1✔
313	shape=(self._input_n_samples[key] + window_dim - 1, self._model_def['Inputs'][key]['dim']),
314	dtype=NP_DTYPE).tolist()
315
316	## Transform inputs in 3D Tensors
317	for key in inputs.keys():	1✔
318	input_dim = json_inputs[key]['dim']	1✔
319	inputs[key] = torch.from_numpy(np.array(inputs[key])).to(TORCH_DTYPE)	1✔
320
321	if input_dim > 1:	1✔
322	correct_dim = 3 if sampled else 2	1✔
323	check(len(inputs[key].shape) == correct_dim, ValueError,	1✔
324	f'The input {key} must have {correct_dim} dimensions')
325	check(inputs[key].shape[correct_dim - 1] == input_dim, ValueError,	1✔
326	f'The second dimension of the input "{key}" must be equal to {input_dim}')
327
328	if input_dim == 1 and inputs[key].shape[-1] != 1: ## add the input dimension	1✔
329	inputs[key] = inputs[key].unsqueeze(-1)	1✔
330	if inputs[key].ndim <= 1: ## add the batch dimension	1✔
331	inputs[key] = inputs[key].unsqueeze(0)	1✔
332	if inputs[key].ndim <= 2: ## add the time dimension	1✔
333	inputs[key] = inputs[key].unsqueeze(0)	1✔
334
335	## initialize the resulting dictionary
336	result_dict = {}	1✔
337	for key in self._model_def['Outputs'].keys():	1✔
338	result_dict[key] = []	1✔
339
340	## Inference
341	calculate_grad = False	1✔
342	for key, value in json_inputs.items():	1✔
343	if 'type' in value.keys():	1✔
344	calculate_grad = True	1✔
345	break	1✔
346	with torch.enable_grad() if calculate_grad else torch.inference_mode():	1✔
347	## Update with virtual states
348	if prediction_samples is not None:	1✔
349	self._model.update(closed_loop=closed_loop, connect=connect)	1✔
350	else:
351	prediction_samples = 0	1✔
352	X = {}	1✔
353	count = 0	1✔
354	first = True	1✔
355	for idx in range(window_dim):	1✔
356	## Get mandatory data inputs
357	for key in mandatory_inputs:	1✔
358	X[key] = inputs[key][idx:idx + 1] if sampled else inputs[key][:,	1✔
359	idx:idx + self._input_n_samples[key]]
360	if 'type' in json_inputs[key].keys():	1✔
361	X[key] = X[key].requires_grad_(True)	1✔
362	## reset states
363	if count == 0 or prediction_samples == 'auto':	1✔
364	count = prediction_samples	1✔
365	for key in non_mandatory_inputs: ## Get non mandatory data (from inputs, from states, or with zeros)	1✔
366	## if prediction_samples is 'auto' and i have enough samples
367	## if prediction_samples is NOT 'auto' but i have enough extended window (with zeros)
368	if (key in inputs.keys() and prediction_samples == 'auto' and idx < num_of_windows[key]) or (	1✔
369	key in inputs.keys() and prediction_samples != 'auto' and idx < inputs[key].shape[1]):
370	X[key] = inputs[key][idx:idx + 1] if sampled else inputs[key][:,	1✔
371	idx:idx + self._input_n_samples[key]]
372	## if im in the first reset
373	## if i have a state in memory
374	## if i have prediction_samples = 'auto' and not enough samples
375	elif (key in self._states.keys() and (first or prediction_samples == 'auto')) and (	1✔
376	prediction_samples == 'auto' or prediction_samples == None):
377	X[key] = self._states[key]	1✔
378	else: ## if i have no samples and no states
379	window_size = self._input_n_samples[key]	1✔
380	dim = json_inputs[key]['dim']	1✔
381	X[key] = torch.zeros(size=(1, window_size, dim), dtype=TORCH_DTYPE, requires_grad=False)	1✔
382	self._states[key] = X[key]	1✔
383	if 'type' in json_inputs[key].keys():	1✔
384	X[key] = X[key].requires_grad_(True)	1✔
385	first = False	1✔
386	else:
387	# Remove the gradient of the previous forward
388	for key in X.keys():	1✔
389	if 'type' in json_inputs[key].keys():	1✔
390	X[key] = X[key].detach().requires_grad_(True)	1✔
391	count -= 1	1✔
392	## Forward pass
393	result, _, out_closed_loop, out_connect = self._model(X)	1✔
394
395	## Append the prediction of the current sample to the result dictionary
396	for key in self._model_def['Outputs'].keys():	1✔
397	if result[key].shape[-1] == 1:	1✔
398	result[key] = result[key].squeeze(-1)	1✔
399	if result[key].shape[-1] == 1:	1✔
400	result[key] = result[key].squeeze(-1)	1✔
401	result_dict[key].append(result[key].detach().squeeze(dim=0).tolist())	1✔
402
403	## Update closed_loop and connect
404	if prediction_samples:	1✔
405	self._updateState(X, out_closed_loop, out_connect)	1✔
406
407	## Remove virtual states
408	self._removeVirtualStates(connect, closed_loop)	1✔
409
410	return result_dict	1✔
411
412

tonegas / nnodely / 14359872492

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