20071179148

Committed 09 Dec 2025 04:40PM UTC coverage: 96.588% (-1.2%) from 97.767%

Build # 20071179148

Build Type

Pull #109

github

Committed by

tonegas

Commit Message

Edits of the README

Pull Request Pull Request #109: New version of nnodely

Run Details

813 of 858 new or added lines in 37 files covered. (94.76%)

153 existing lines in 4 files now uncovered.

13021 of 13481 relevant lines covered (96.59%)

0.97 hits per line

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

96.5

/nnodely/operators/composer.py

import copy, torch

import numpy as np

from nnodely.operators.network import Network

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

from nnodely.support.logger import logging, nnLogger
log = nnLogger(__name__, logging.WARNING)

class Composer(Network):
    @enforce_types
    def __init__(self):
        check(type(self) is not Composer, TypeError, "Composer class cannot be instantiated directly")
        super().__init__()

    def __addInfo(self) -> None:
        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) -> 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])
        """
        self._model_def.addModel(name, stream_list)
        self._neuralized = False

    @enforce_types
    def removeModel(self, name_list:list|str) -> 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)
        self._neuralized = False

    @enforce_types
    def addConnect(self, stream_out:str|Output|Stream, input_in:str|Input, *, local:bool=False) -> None:
        """
        Adds a connection from a relation stream to an input.

        Parameters
        ----------
        stream_out : Stream
            The relation stream to connect from.
        input_in : Input or list of inputs
            The input or list of input 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 = Input('y')
            >>> relation = Fir(x.last())
            >>> model.addConnect(relation, y)
        """
        self._model_def.addConnection(stream_out, input_in,'connect', local)
        self._neuralized = False

    @enforce_types
    def addClosedLoop(self, stream_out:str|Output|Stream, input_in:str|Input, *, local:bool=False) -> None:
        """
        Adds a closed loop connection from a relation stream to an input.

        Parameters
        ----------
        stream_out : Stream
            The relation stream to connect from.
        input_in : Input or list of inputs
            The Input or the list of inputs 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 = Input('y')
            >>> relation = Fir(x.last())
            >>> model.addClosedLoop(relation, y)
        """
        self._model_def.addConnection(stream_out, input_in,'closedLoop', local)
        self._neuralized = False

    @enforce_types
    def removeConnection(self, input_in:str|Input) -> None:
        """
        Remove a closed loop or connect connection from an input.

        Parameters
        ----------
        input_in : Input or name of the input of inputs
            The Input to disconnect.

        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 = Input('y')
            >>> relation = Fir(x.last())
            >>> model.addConnect(relation, y)
            >>> model.removeConnection(y)
        """
        if isinstance(input_in, Input):
            input_name = input_in.name
        else:
            input_name = input_in
        self._model_def.removeConnection(input_name)
        self._neuralized = False

    @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 and correct slicing for all the inputs defined.

        Parameters
        ----------
        sample_time : float or None, optional
            The sample time for the model. Default is 1.0
        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:
            self._model_def.updateParameters(model = None, clear_model = clear_model)

        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'].items()}
        self._input_ns_forward = {key:value['ns'][1] for key, value in self._model_def['Inputs'].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'].items():
            if self._input_ns_forward[key] >= 0:
                if 'closedLoop' in value:
                    log.warning(f"Closed loop on {key} with sample in the future.")
                if 'connect' in value:
                    log.warning(f"Connect on {key} with sample in the future.")
            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._model_def.updateParameters(self._model)
        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='auto', num_of_samples:int|None=None, log_internal:bool=False) -> dict:
        """
        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 direct 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)
        all_closed_loop = copy.deepcopy(closed_loop) #| self._model_def._input_closed_loop
        all_connect = copy.deepcopy(connect) #| self._model_def._input_connect

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

        ## Check closed loop integrity
        prediction_samples = self._setup_recurrent_variables(prediction_samples, all_closed_loop, all_connect)

        ## List of keys
        model_inputs = list(self._model_def['Inputs'].keys())
        json_inputs = self._model_def['Inputs']
        extra_inputs = list(set(list(inputs.keys())) - set(model_inputs))
        non_mandatory_inputs = list(all_closed_loop.keys()) + list(all_connect.keys()) +  list(self._model_def.recurrentInputs().keys())
        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
        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()}

        if num_of_samples is not None and sampled == True:
            log.warning(f'num_of_samples is ignored if sampled is equal to True')

        ## Get the maximum inference window
        if num_of_samples and not sampled:
            window_dim = num_of_samples
            for key in inputs.keys():
                input_dim = self._model_def['Inputs'][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
                        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] = []
        if log_internal:
            internals_dict = {'ingress': [], 'state': [], 'closedLoop': [], 'connect': []}

        ## Inference
        with (torch.enable_grad() if self._get_gradient_on_inference() else torch.inference_mode()):
            ## Update with virtual states
            if prediction_samples == 'auto' or prediction_samples >= 0:
                self._model.update(closed_loop=all_closed_loop, connect=all_connect)
            else:
                self._model.update(disconnect=True)
                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':
                    init_states = []
                    count = prediction_samples
                    for key in non_mandatory_inputs:  ## Get non mandatory data (from inputs, from states, or with zeros)
                        ## If it is given as input AND
                        ## if prediction_samples is 'auto' and there are enough samples OR
                        ## if prediction_samples is NOT 'auto'
                        if key in inputs.keys() and (
                                (prediction_samples == 'auto' and idx < num_of_windows[key]) or \
                                (prediction_samples != 'auto')
                        ):
                            X[key] = inputs[key][idx:idx + 1] if sampled else inputs[key][:,idx:idx + self._input_n_samples[key]]
                            if 0 in X[key].shape:
                                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)
                        ## if it is a state AND
                        ## if prediction_samples = 'auto' and there are not enough samples OR
                        ## it is the first iteration with prediction_samples = None
                        elif key in self._states.keys() and (
                                prediction_samples == 'auto' or
                                (first and prediction_samples == None)
                        ):
                            X[key] = self._states[key]
                        else:
                        ## if there are no samples
                            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)

                        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)
                if log_internal:
                    internals_dict['ingress'].append(tensor_to_list(X)) 
                    internals_dict['closedLoop'].append(out_closed_loop)
                    internals_dict['connect'].append(out_connect)

                if init_states:
                    for key in init_states:
                        del self._model.connect_update[key]
                    init_states = []

                ## 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._update_state(X, out_closed_loop, out_connect)
                    
        ## Remove virtual states
        self._remove_virtual_states(connect, closed_loop)

        return result_dict if not log_internal else (result_dict, internals_dict)



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

tonegas / nnodely / 20071179148

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