13504505673

Committed 24 Feb 2025 05:53PM UTC coverage: 95.261% (+0.3%) from 94.961%

Build # 13504505673

Build Type

Pull #59

github

Committed by

web-flow

Commit Message

Merge c15288a9e into 0c108fc0d

Pull Request Pull Request #59: Features/56 dynamic parametric function

Run Details

567 of 582 new or added lines in 24 files covered. (97.42%)

3 existing lines in 3 files now uncovered.

10171 of 10677 relevant lines covered (95.26%)

0.95 hits per line

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

95.59

/nnodely/parametricfunction.py

import inspect, copy, textwrap, torch, math

import torch.nn as nn
import numpy as np

from typing import Union
from collections.abc import Callable

from nnodely.relation import NeuObj, Stream, toStream
from nnodely.model import Model
from nnodely.parameter import Parameter, Constant
from nnodely.utils import check, merge, enforce_types

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


paramfun_relation_name = 'ParamFun'

class ParamFun(NeuObj):
    """
    Represents a parametric function in the neural network model.

    Parameters
    ----------
    param_fun : Callable
        The parametric function to be used.
    constants : list or dict or None, optional
        A list or dictionary of constants to be used in the function. Default is None.
    parameters_dimensions : list or dict or None, optional
        A list or dictionary specifying the dimensions of the parameters. Default is None.
    parameters : list or dict or None, optional
        A list or dictionary of parameters to be used in the function. Default is None.
    map_over_batch : bool, optional
        A boolean indicating whether to map the function over the batch dimension. Default is False.

    Attributes
    ----------
    relation_name : str
        The name of the relation.
    param_fun : Callable
        The parametric function to be used.
    constants : list or dict or None
        A list or dictionary of constants to be used in the function.
    parameters_dimensions : list or dict or None
        A list or dictionary specifying the dimensions of the parameters.
    parameters : list or dict or None
        A list or dictionary of parameters to be used in the function.
    map_over_batch : bool
        A boolean indicating whether to map the function over the batch dimension.
    output_dimension : dict
        A dictionary containing the output dimensions of the function.
    json : dict
        A dictionary containing the configuration of the function.

    Example
    -------
        >>> input1 = Input('input1')
        >>> input2 = Input('input2')

        >>> def my_function(x, y, param1, const1):
        >>>     return param1 * x + const1 * y

        >>> param_fun = ParamFun(my_function, constants={'const1': 1.0}, parameters_dimensions={'param1': 1})
        >>> result = param_fun(input1, input2)
    """
    @enforce_types
    def __init__(self, param_fun:Callable,
                 constants:list|dict|None = None,
                 parameters_dimensions:list|dict|None = None,
                 parameters:list|dict|None = None,
                 map_over_batch:bool = False) -> Stream:

        self.relation_name = paramfun_relation_name

        # input parameters
        self.param_fun = param_fun
        self.constants = constants
        self.parameters_dimensions = parameters_dimensions
        self.parameters = parameters
        self.map_over_batch = map_over_batch

        self.output_dimension = {}
        super().__init__('F'+paramfun_relation_name + str(NeuObj.count))
        code = textwrap.dedent(inspect.getsource(param_fun)).replace('\"', '\'')
        self.json['Functions'][self.name] = {
            'code' : code,
            'name' : param_fun.__name__,
        }
        self.json['Functions'][self.name]['params_and_consts'] = []

        # Create the missing constants from list
        if type(self.constants) is list:
            for const in self.constants:
                if type(const) is Constant:
                    self.json['Functions'][self.name]['params_and_consts'].append(const.name)
                    self.json['Constants'][const.name] = copy.deepcopy(const.json['Constants'][const.name])
                elif type(const) is str:
                    self.json['Functions'][self.name]['params_and_consts'].append(const)
                    self.json['Constants'][const] = {'dim': 1, 'sw' : 1}
                else:
                    check(type(const) is Constant or type(const) is str, TypeError,
                          'The element inside the \"constants\" list must be a Constant or str')

        # Create the missing parameters from list
        if type(self.parameters) is list:
            check(self.parameters_dimensions is None, ValueError,
                  '\"parameters_dimensions\" must be None if \"parameters\" is set using list')
            for param in self.parameters:
                if type(param) is Parameter:
                    self.json['Functions'][self.name]['params_and_consts'].append(param.name)
                    self.json['Parameters'][param.name] = copy.deepcopy(param.json['Parameters'][param.name])
                elif type(param) is str:
                    self.json['Functions'][self.name]['params_and_consts'].append(param)
                    self.json['Parameters'][param] = {'dim': 1, 'sw' : 1}
                else:
                    check(type(param) is Parameter or type(param) is str, TypeError,
                          'The element inside the \"parameters\" list must be a Parameter or str')
        elif type(self.parameters_dimensions) is list:
            funinfo = inspect.getfullargspec(self.param_fun)
            for i, param_dim in enumerate(self.parameters_dimensions):
                idx = i + len(funinfo.args) - len(self.parameters_dimensions)
                param_name = self.name + str(idx)
                self.json['Functions'][self.name]['params_and_consts'].append(param_name)
                self.json['Parameters'][param_name] = {'dim': list(self.parameters_dimensions[i]),'sw' : 1}

        self.json_stream = {}

    @enforce_types
    def __call__(self, *obj:Union[Stream|Parameter|Constant]) -> Stream:
        stream_name = paramfun_relation_name + str(Stream.count)

        funinfo = inspect.getfullargspec(self.param_fun)
        n_function_input = len(funinfo.args)
        n_call_input = len(obj)
        n_new_constants_and_params = n_function_input - n_call_input

        if n_call_input not in self.json_stream:
            if len(self.json_stream) > 0:
                log.warning(f"The function {self.name} was called with a different number of inputs. If both functions enter in the model an error will be raised.")

            self.json_stream[n_call_input] = copy.deepcopy(self.json)
            self.json_stream[n_call_input]['Functions'][self.name]['n_input'] = n_call_input
            self.__create_missing_parameters(self.json_stream[n_call_input], n_new_constants_and_params)

            input_dimensions = []
            input_types = []
            for ind, o in enumerate(obj):
                if type(o) in (int, float, list):
                    obj_type = Constant
                else:
                    obj_type = type(o)
                o = toStream(o)
                check(type(o) is Stream, TypeError,
                      f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
                input_types.append(obj_type)
                input_dimensions.append(o.dim)

            # Create the missing parameters
            missing_params = n_new_constants_and_params - len(self.json_stream[n_call_input]['Functions'][self.name]['params_and_consts'])
            check(missing_params == 0, ValueError, f"The function is called with different number of inputs.")

            self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = copy.deepcopy(input_dimensions)
            self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'] = self.__map_over_batch(
                    n_call_input,
                    n_new_constants_and_params)
            output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types,
                                                                  input_dimensions)

        else:
            input_dimensions = []
            input_types = []
            for ind, o in enumerate(obj):
                if type(o) in (int, float, list):
                    obj_type = Constant
                else:
                    obj_type = type(o)
                o = toStream(o)
                check(type(o) is Stream, TypeError,
                      f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
                input_types.append(obj_type)
                input_dimensions.append(o.dim)
            output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)

            # Save the all the input dimension used for call the parametric function
            in_dim = self.json_stream[n_call_input]['Functions'][self.name]['in_dim']
            if type(in_dim[0]) is dict:
                if in_dim != input_dimensions:
                    in_dim = [in_dim, input_dimensions]
                    log.warning(f"The function {self.name} was called with inputs with different dimensions.")
            elif input_dimensions not in in_dim:
                in_dim.append(input_dimensions)
                log.warning(f"The function {self.name} was called with inputs with different dimensions.")
            self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = in_dim

        stream_json = copy.deepcopy(self.json_stream[n_call_input])
        input_names = []
        for ind, o in enumerate(obj):
            o = toStream(o)
            check(type(o) is Stream, TypeError,
                  f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
            stream_json = merge(stream_json, o.json)
            input_names.append(o.name)

        stream_json['Relations'][stream_name] = [paramfun_relation_name, input_names, self.name]
        return Stream(stream_name, stream_json, output_dimension)

    def __map_over_batch(self, n_call_input, n_constants_and_params):
        input_map_dim = ()

        for i in range(n_call_input):
            input_map_dim += (0,)
        for i in range(n_constants_and_params):
            input_map_dim += (None,)

        if self.map_over_batch:
            return list(input_map_dim)
        else:
            return False

    def __create_missing_parameters(self, stream_json, n_new_constants_and_params):
        funinfo = inspect.getfullargspec(self.param_fun)
        # Create the missing parameters and constants from dict
        missing_params = n_new_constants_and_params - len(stream_json['Functions'][self.name]['params_and_consts'])
        if missing_params or type(self.constants) is dict or type(self.parameters) is dict or type(self.parameters_dimensions) is dict:
            n_input = len(funinfo.args) - missing_params
            n_elem_dict = (len(self.constants if type(self.constants) is dict else [])
                           + len(self.parameters if type(self.parameters) is dict else [])
                           + len(self.parameters_dimensions if type(self.parameters_dimensions) is dict else []))
            for i, key in enumerate(funinfo.args):
                if i >= n_input:
                    if type(self.parameters) is dict and key in self.parameters:
                        if self.parameters_dimensions:
                            check(key in self.parameters_dimensions, TypeError,
                                  f'The parameter {key} must be removed from \"parameters_dimensions\".')
                        param = self.parameters[key]
                        if type(self.parameters[key]) is Parameter:
                            stream_json['Functions'][self.name]['params_and_consts'].append(param.name)
                            stream_json['Parameters'][param.name] = copy.deepcopy(param.json['Parameters'][param.name])
                        elif type(self.parameters[key]) is str:
                            stream_json['Functions'][self.name]['params_and_consts'].append(param)
                            stream_json['Parameters'][param] = {'dim' : 1,'sw' : 1}
                        else:
                            check(type(param) is Parameter or type(param) is str, TypeError,
                                  'The element inside the \"parameters\" dict must be a Parameter or str')
                        n_elem_dict -= 1
                    elif type(self.parameters_dimensions) is dict and key in self.parameters_dimensions:
                        param_name = self.name + key
                        dim = self.parameters_dimensions[key]
                        check(isinstance(dim,(list,tuple,int)), TypeError,
                              'The element inside the \"parameters_dimensions\" dict must be a tuple or int')
                        stream_json['Functions'][self.name]['params_and_consts'].append(param_name)
                        stream_json['Parameters'][param_name] = {'dim': list(dim) if type(dim) is tuple else dim, 'sw' : 1}
                        n_elem_dict -= 1
                    elif type(self.constants) is dict and key in self.constants:
                        const = self.constants[key]
                        if type(self.constants[key]) is Constant:
                            stream_json['Functions'][self.name]['params_and_consts'].append(const.name)
                            stream_json['Constants'][const.name] = copy.deepcopy(const.json['Constants'][const.name])
                        elif type(self.constants[key]) is str:
                            stream_json['Functions'][self.name]['params_and_consts'].append(const)
                            stream_json['Constants'][const] = {'dim': 1, 'sw' : 1}
                        else:
                            check(type(const) is Constant or type(const) is str, TypeError,
                                  'The element inside the \"constants\" dict must be a Constant or str')
                        n_elem_dict -= 1
                    else:
                        param_name = self.name + key
                        stream_json['Functions'][self.name]['params_and_consts'].append(param_name)
                        stream_json['Parameters'][param_name] = {'dim': 1, 'sw' : 1}
            check(n_elem_dict == 0, ValueError, 'Some of the input parameters are not used in the function.')

    def __infer_output_dimensions(self, stream_json, input_types, input_dimensions):
        import torch
        batch_dim = 5

        all_inputs_dim = copy.deepcopy(input_dimensions)
        all_inputs_type = copy.deepcopy(input_types)
        params_and_consts = stream_json['Constants'] | stream_json['Parameters']
        for name in stream_json['Functions'][self.name]['params_and_consts']:
            all_inputs_dim.append(params_and_consts[name])
            all_inputs_type.append(Constant)

        n_samples_sec = 0.1
        is_int = False
        while is_int == False:
            n_samples_sec *= 10
            vect_input_time = [math.isclose(d['tw']*n_samples_sec,round(d['tw']*n_samples_sec)) for d in all_inputs_dim if 'tw' in d]
            if len(vect_input_time) == 0:
                is_int = True
            else:
                is_int = sum(vect_input_time) == len(vect_input_time)

        # Build input with right dimensions
        inputs = []
        inputs_win_type = []
        inputs_win = []

        for t, dim in zip(all_inputs_type,all_inputs_dim):
            window = 'tw' if 'tw' in dim else ('sw' if 'sw' in dim else None)
            if window == 'tw':
                dim_win = round(dim[window] * n_samples_sec)
            elif window == 'sw':
                dim_win = dim[window]
            else:
                dim_win = 1
            if t in (Parameter, Constant):
                if type(dim['dim']) is list:
                    inputs.append(torch.rand(size=(dim_win,) + tuple(dim['dim'])))
                else:
                    inputs.append(torch.rand(size=(dim_win, dim['dim'])))
            else:
                inputs.append(torch.rand(size=(batch_dim, dim_win, dim['dim'])))

            inputs_win_type.append(window)
            inputs_win.append(dim_win)

        if self.map_over_batch:
            function_to_call = torch.func.vmap(self.param_fun,in_dims=tuple(stream_json['Functions'][self.name]['map_over_dim']))
        else:
            function_to_call = self.param_fun
        out = function_to_call(*inputs)
        out_shape = out.shape
        check(out_shape[0] == batch_dim, ValueError, "The batch output dimension it is not correct.")
        out_dim = list(out_shape[2:])
        check(len(out_dim) == 1, ValueError, "The output dimension of the function is bigger than a vector.")
        out_win_type = 'sw'
        out_win = out_shape[1]
        for idx, win in enumerate(inputs_win):
            if out_shape[1] == win and all_inputs_type[idx] not in (Parameter, Constant):
                out_win_type = inputs_win_type[idx]
                out_win = all_inputs_dim[idx][out_win_type]

        return { 'dim': out_dim[0], out_win_type : out_win }

def return_standard_inputs(json, model_def, xlim = None, num_points = 1000):
    check(json['n_input'] == 1 or json['n_input'] == 2, ValueError, "The function must have only one or two inputs.")
    fun_inputs = tuple()
    for i in range(json['n_input']):
        dim = json['in_dim'][i]
        check(dim['dim'] == 1, ValueError, "The input dimension must be 1.")
        if 'tw' in dim:
            check(dim['tw'] == model_def['Info']['SampleTime'], ValueError, "The input window must be 1.")
        elif 'sw' in dim:
            check(dim['sw'] == 1, ValueError, "The input window must be 1.")
        if xlim is not None:
            if json['n_input'] == 2:
                check(np.array(xlim).shape == (json['n_input'], 2), ValueError,
                      "The xlim must have the same shape as the number of inputs.")
                x_value = np.linspace(xlim[i][0], xlim[i][1], num=num_points)
            else:
                check(np.array(xlim).shape == (2,), ValueError,
                      "The xlim must have the same shape as the number of inputs.")
                x_value = np.linspace(xlim[0], xlim[1], num=num_points)
        else:
            x_value = np.linspace(0, 1, num=num_points)
        if i == 0:
            x0_value = torch.from_numpy(x_value)
        else:
            x1_value = torch.from_numpy(x_value)

    if json['n_input'] == 2:
        x0_value, x1_value = torch.meshgrid(x0_value,x1_value,indexing="xy")
        x0_value = x0_value.flatten().unsqueeze(1).unsqueeze(1)
        x1_value = x1_value.flatten().unsqueeze(1).unsqueeze(1)
        fun_inputs += (x0_value,x1_value,)
    else:
        x0_value = x0_value.unsqueeze(1).unsqueeze(1)
        fun_inputs += (x0_value,)

    for key in json['params_and_consts']:
        val = model_def['Parameters'][key] if key in model_def['Parameters'] else model_def['Constants'][key]
        fun_inputs += tuple([torch.from_numpy(np.array(val['values']))]) # The vector is transform in a tuple

    return fun_inputs

def return_function(json, fun_inputs):
    exec(json['code'], globals())
    function_to_call = globals()[json['name']]
    output = function_to_call(*fun_inputs)
    check(output.shape[1] == 1, ValueError, "The output dimension must be 1.")
    check(output.shape[2] == 1, ValueError, "The output window must be 1.")
    funinfo = inspect.getfullargspec(function_to_call)
    return output, funinfo.args

class Parametric_Layer(nn.Module):
    def __init__(self, func, params_and_consts, map_over_batch):
        super().__init__()
        self.name = func['name']
        self.params_and_consts = params_and_consts
        if type(map_over_batch) is list:
            self.map_over_batch = True
            self.input_map_dim = tuple(map_over_batch)
        else:
            self.map_over_batch = False
        ## Add the function to the globals
        try:
            code = 'import torch\n@torch.fx.wrap\n' + func['code']
            exec(code, globals())
        except Exception as e:
            print(f"An error occurred: {e}")

    def forward(self, *inputs):
        args = list(inputs) + self.params_and_consts
        # Retrieve the function object from the globals dictionary
        function_to_call = globals()[self.name]
        # Call the function using the retrieved function object
        if self.map_over_batch:
            function_to_call = torch.func.vmap(function_to_call,in_dims=self.input_map_dim)
        result = function_to_call(*args)
        return result

def createParamFun(self, *func_params):
    return Parametric_Layer(func=func_params[0], params_and_consts=func_params[1], map_over_batch=func_params[2])

setattr(Model, paramfun_relation_name, createParamFun)

1	import inspect, copy, textwrap, torch, math	1✔
2
3	import torch.nn as nn	1✔
4	import numpy as np	1✔
5
6	from typing import Union	1✔
7	from collections.abc import Callable	1✔
8
9	from nnodely.relation import NeuObj, Stream, toStream	1✔
10	from nnodely.model import Model	1✔
11	from nnodely.parameter import Parameter, Constant	1✔
12	from nnodely.utils import check, merge, enforce_types	1✔
13
14	from nnodely.logger import logging, nnLogger	1✔
15	log = nnLogger(__name__, logging.WARNING)	1✔
16
17
18	paramfun_relation_name = 'ParamFun'	1✔
19
20	class ParamFun(NeuObj):	1✔
21	"""
22	Represents a parametric function in the neural network model.
23
24	Parameters
25	----------
26	param_fun : Callable
27	The parametric function to be used.
28	constants : list or dict or None, optional
29	A list or dictionary of constants to be used in the function. Default is None.
30	parameters_dimensions : list or dict or None, optional
31	A list or dictionary specifying the dimensions of the parameters. Default is None.
32	parameters : list or dict or None, optional
33	A list or dictionary of parameters to be used in the function. Default is None.
34	map_over_batch : bool, optional
35	A boolean indicating whether to map the function over the batch dimension. Default is False.
36
37	Attributes
38	----------
39	relation_name : str
40	The name of the relation.
41	param_fun : Callable
42	The parametric function to be used.
43	constants : list or dict or None
44	A list or dictionary of constants to be used in the function.
45	parameters_dimensions : list or dict or None
46	A list or dictionary specifying the dimensions of the parameters.
47	parameters : list or dict or None
48	A list or dictionary of parameters to be used in the function.
49	map_over_batch : bool
50	A boolean indicating whether to map the function over the batch dimension.
51	output_dimension : dict
52	A dictionary containing the output dimensions of the function.
53	json : dict
54	A dictionary containing the configuration of the function.
55
56	Example
57	-------
58	>>> input1 = Input('input1')
59	>>> input2 = Input('input2')
60
61	>>> def my_function(x, y, param1, const1):
62	>>> return param1 * x + const1 * y
63
64	>>> param_fun = ParamFun(my_function, constants={'const1': 1.0}, parameters_dimensions={'param1': 1})
65	>>> result = param_fun(input1, input2)
66	"""
67	@enforce_types	1✔
68	def __init__(self, param_fun:Callable,	1✔
69	constants:list\|dict\|None = None,
70	parameters_dimensions:list\|dict\|None = None,
71	parameters:list\|dict\|None = None,
72	map_over_batch:bool = False) -> Stream:
73
74	self.relation_name = paramfun_relation_name	1✔
75
76	# input parameters
77	self.param_fun = param_fun	1✔
78	self.constants = constants	1✔
79	self.parameters_dimensions = parameters_dimensions	1✔
80	self.parameters = parameters	1✔
81	self.map_over_batch = map_over_batch	1✔
82
83	self.output_dimension = {}	1✔
84	super().__init__('F'+paramfun_relation_name + str(NeuObj.count))	1✔
85	code = textwrap.dedent(inspect.getsource(param_fun)).replace('\"', '\'')	1✔
86	self.json['Functions'][self.name] = {	1✔
87	'code' : code,
88	'name' : param_fun.__name__,
89	}
90	self.json['Functions'][self.name]['params_and_consts'] = []	1✔
91
92	# Create the missing constants from list
93	if type(self.constants) is list:	1✔
94	for const in self.constants:	1✔
95	if type(const) is Constant:	1✔
96	self.json['Functions'][self.name]['params_and_consts'].append(const.name)	1✔
97	self.json['Constants'][const.name] = copy.deepcopy(const.json['Constants'][const.name])	1✔
98	elif type(const) is str:	1✔
99	self.json['Functions'][self.name]['params_and_consts'].append(const)	1✔
100	self.json['Constants'][const] = {'dim': 1, 'sw' : 1}	1✔
101	else:
NEW 102	check(type(const) is Constant or type(const) is str, TypeError,	×
103	'The element inside the \"constants\" list must be a Constant or str')
104
105	# Create the missing parameters from list
106	if type(self.parameters) is list:	1✔
107	check(self.parameters_dimensions is None, ValueError,	1✔
108	'\"parameters_dimensions\" must be None if \"parameters\" is set using list')
109	for param in self.parameters:	1✔
110	if type(param) is Parameter:	1✔
111	self.json['Functions'][self.name]['params_and_consts'].append(param.name)	1✔
112	self.json['Parameters'][param.name] = copy.deepcopy(param.json['Parameters'][param.name])	1✔
113	elif type(param) is str:	1✔
114	self.json['Functions'][self.name]['params_and_consts'].append(param)	1✔
115	self.json['Parameters'][param] = {'dim': 1, 'sw' : 1}	1✔
116	else:
NEW 117	check(type(param) is Parameter or type(param) is str, TypeError,	×
118	'The element inside the \"parameters\" list must be a Parameter or str')
119	elif type(self.parameters_dimensions) is list:	1✔
120	funinfo = inspect.getfullargspec(self.param_fun)	1✔
121	for i, param_dim in enumerate(self.parameters_dimensions):	1✔
122	idx = i + len(funinfo.args) - len(self.parameters_dimensions)	1✔
123	param_name = self.name + str(idx)	1✔
124	self.json['Functions'][self.name]['params_and_consts'].append(param_name)	1✔
125	self.json['Parameters'][param_name] = {'dim': list(self.parameters_dimensions[i]),'sw' : 1}	1✔
126
127	self.json_stream = {}	1✔
128
129	@enforce_types	1✔
130	def __call__(self, *obj:Union[Stream\|Parameter\|Constant]) -> Stream:	1✔
131	stream_name = paramfun_relation_name + str(Stream.count)	1✔
132
133	funinfo = inspect.getfullargspec(self.param_fun)	1✔
134	n_function_input = len(funinfo.args)	1✔
135	n_call_input = len(obj)	1✔
136	n_new_constants_and_params = n_function_input - n_call_input	1✔
137
138	if n_call_input not in self.json_stream:	1✔
139	if len(self.json_stream) > 0:	1✔
140	log.warning(f"The function {self.name} was called with a different number of inputs. If both functions enter in the model an error will be raised.")	1✔
141
142	self.json_stream[n_call_input] = copy.deepcopy(self.json)	1✔
143	self.json_stream[n_call_input]['Functions'][self.name]['n_input'] = n_call_input	1✔
144	self.__create_missing_parameters(self.json_stream[n_call_input], n_new_constants_and_params)	1✔
145
146	input_dimensions = []	1✔
147	input_types = []	1✔
148	for ind, o in enumerate(obj):	1✔
149	if type(o) in (int, float, list):	1✔
150	obj_type = Constant	×
151	else:
152	obj_type = type(o)	1✔
153	o = toStream(o)	1✔
154	check(type(o) is Stream, TypeError,	1✔
155	f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
156	input_types.append(obj_type)	1✔
157	input_dimensions.append(o.dim)	1✔
158
159	# Create the missing parameters
160	missing_params = n_new_constants_and_params - len(self.json_stream[n_call_input]['Functions'][self.name]['params_and_consts'])	1✔
161	check(missing_params == 0, ValueError, f"The function is called with different number of inputs.")	1✔
162
163	self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = copy.deepcopy(input_dimensions)	1✔
164	self.json_stream[n_call_input]['Functions'][self.name]['map_over_dim'] = self.__map_over_batch(	1✔
165	n_call_input,
166	n_new_constants_and_params)
167	output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types,	1✔
168	input_dimensions)
169
170	else:
171	input_dimensions = []	1✔
172	input_types = []	1✔
173	for ind, o in enumerate(obj):	1✔
174	if type(o) in (int, float, list):	1✔
NEW 175	obj_type = Constant	×
176	else:
177	obj_type = type(o)	1✔
178	o = toStream(o)	1✔
179	check(type(o) is Stream, TypeError,	1✔
180	f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
181	input_types.append(obj_type)	1✔
182	input_dimensions.append(o.dim)	1✔
183	output_dimension = self.__infer_output_dimensions(self.json_stream[n_call_input], input_types, input_dimensions)	1✔
184
185	# Save the all the input dimension used for call the parametric function
186	in_dim = self.json_stream[n_call_input]['Functions'][self.name]['in_dim']	1✔
187	if type(in_dim[0]) is dict:	1✔
188	if in_dim != input_dimensions:	1✔
189	in_dim = [in_dim, input_dimensions]	1✔
190	log.warning(f"The function {self.name} was called with inputs with different dimensions.")	1✔
191	elif input_dimensions not in in_dim:	1✔
192	in_dim.append(input_dimensions)	1✔
193	log.warning(f"The function {self.name} was called with inputs with different dimensions.")	1✔
194	self.json_stream[n_call_input]['Functions'][self.name]['in_dim'] = in_dim	1✔
195
196	stream_json = copy.deepcopy(self.json_stream[n_call_input])	1✔
197	input_names = []	1✔
198	for ind, o in enumerate(obj):	1✔
199	o = toStream(o)	1✔
200	check(type(o) is Stream, TypeError,	1✔
201	f"The type of {o} is {type(o)} and is not supported for ParamFun operation.")
202	stream_json = merge(stream_json, o.json)	1✔
203	input_names.append(o.name)	1✔
204
205	stream_json['Relations'][stream_name] = [paramfun_relation_name, input_names, self.name]	1✔
206	return Stream(stream_name, stream_json, output_dimension)	1✔
207
208	def __map_over_batch(self, n_call_input, n_constants_and_params):	1✔
209	input_map_dim = ()	1✔
210
211	for i in range(n_call_input):	1✔
212	input_map_dim += (0,)	1✔
213	for i in range(n_constants_and_params):	1✔
214	input_map_dim += (None,)	1✔
215
216	if self.map_over_batch:	1✔
217	return list(input_map_dim)	1✔
218	else:
219	return False	1✔
220
221	def __create_missing_parameters(self, stream_json, n_new_constants_and_params):	1✔
222	funinfo = inspect.getfullargspec(self.param_fun)	1✔
223	# Create the missing parameters and constants from dict
224	missing_params = n_new_constants_and_params - len(stream_json['Functions'][self.name]['params_and_consts'])	1✔
225	if missing_params or type(self.constants) is dict or type(self.parameters) is dict or type(self.parameters_dimensions) is dict:	1✔
226	n_input = len(funinfo.args) - missing_params	1✔
227	n_elem_dict = (len(self.constants if type(self.constants) is dict else [])	1✔
228	+ len(self.parameters if type(self.parameters) is dict else [])
229	+ len(self.parameters_dimensions if type(self.parameters_dimensions) is dict else []))
230	for i, key in enumerate(funinfo.args):	1✔
231	if i >= n_input:	1✔
232	if type(self.parameters) is dict and key in self.parameters:	1✔
233	if self.parameters_dimensions:	1✔
234	check(key in self.parameters_dimensions, TypeError,	×
235	f'The parameter {key} must be removed from \"parameters_dimensions\".')
236	param = self.parameters[key]	1✔
237	if type(self.parameters[key]) is Parameter:	1✔
238	stream_json['Functions'][self.name]['params_and_consts'].append(param.name)	1✔
239	stream_json['Parameters'][param.name] = copy.deepcopy(param.json['Parameters'][param.name])	1✔
240	elif type(self.parameters[key]) is str:	1✔
241	stream_json['Functions'][self.name]['params_and_consts'].append(param)	1✔
242	stream_json['Parameters'][param] = {'dim' : 1,'sw' : 1}	1✔
243	else:
244	check(type(param) is Parameter or type(param) is str, TypeError,	×
245	'The element inside the \"parameters\" dict must be a Parameter or str')
246	n_elem_dict -= 1	1✔
247	elif type(self.parameters_dimensions) is dict and key in self.parameters_dimensions:	1✔
248	param_name = self.name + key	1✔
249	dim = self.parameters_dimensions[key]	1✔
250	check(isinstance(dim,(list,tuple,int)), TypeError,	1✔
251	'The element inside the \"parameters_dimensions\" dict must be a tuple or int')
252	stream_json['Functions'][self.name]['params_and_consts'].append(param_name)	1✔
253	stream_json['Parameters'][param_name] = {'dim': list(dim) if type(dim) is tuple else dim, 'sw' : 1}	1✔
254	n_elem_dict -= 1	1✔
255	elif type(self.constants) is dict and key in self.constants:	1✔
256	const = self.constants[key]	1✔
257	if type(self.constants[key]) is Constant:	1✔
258	stream_json['Functions'][self.name]['params_and_consts'].append(const.name)	1✔
259	stream_json['Constants'][const.name] = copy.deepcopy(const.json['Constants'][const.name])	1✔
260	elif type(self.constants[key]) is str:	1✔
261	stream_json['Functions'][self.name]['params_and_consts'].append(const)	1✔
262	stream_json['Constants'][const] = {'dim': 1, 'sw' : 1}	1✔
263	else:
264	check(type(const) is Constant or type(const) is str, TypeError,	×
265	'The element inside the \"constants\" dict must be a Constant or str')
266	n_elem_dict -= 1	1✔
267	else:
268	param_name = self.name + key	1✔
269	stream_json['Functions'][self.name]['params_and_consts'].append(param_name)	1✔
270	stream_json['Parameters'][param_name] = {'dim': 1, 'sw' : 1}	1✔
271	check(n_elem_dict == 0, ValueError, 'Some of the input parameters are not used in the function.')	1✔
272
273	def __infer_output_dimensions(self, stream_json, input_types, input_dimensions):	1✔
274	import torch	1✔
275	batch_dim = 5	1✔
276
277	all_inputs_dim = copy.deepcopy(input_dimensions)	1✔
278	all_inputs_type = copy.deepcopy(input_types)	1✔
279	params_and_consts = stream_json['Constants'] \| stream_json['Parameters']	1✔
280	for name in stream_json['Functions'][self.name]['params_and_consts']:	1✔
281	all_inputs_dim.append(params_and_consts[name])	1✔
282	all_inputs_type.append(Constant)	1✔
283
284	n_samples_sec = 0.1	1✔
285	is_int = False	1✔
286	while is_int == False:	1✔
287	n_samples_sec *= 10	1✔
288	vect_input_time = [math.isclose(d['tw']n_samples_sec,round(d['tw']n_samples_sec)) for d in all_inputs_dim if 'tw' in d]	1✔
289	if len(vect_input_time) == 0:	1✔
290	is_int = True	1✔
291	else:
292	is_int = sum(vect_input_time) == len(vect_input_time)	1✔
293
294	# Build input with right dimensions
295	inputs = []	1✔
296	inputs_win_type = []	1✔
297	inputs_win = []	1✔
298
299	for t, dim in zip(all_inputs_type,all_inputs_dim):	1✔
300	window = 'tw' if 'tw' in dim else ('sw' if 'sw' in dim else None)	1✔
301	if window == 'tw':	1✔
302	dim_win = round(dim[window] * n_samples_sec)	1✔
303	elif window == 'sw':	1✔
304	dim_win = dim[window]	1✔
305	else:
UNCOV 306	dim_win = 1	×
307	if t in (Parameter, Constant):	1✔
308	if type(dim['dim']) is list:	1✔
309	inputs.append(torch.rand(size=(dim_win,) + tuple(dim['dim'])))	1✔
310	else:
311	inputs.append(torch.rand(size=(dim_win, dim['dim'])))	1✔
312	else:
313	inputs.append(torch.rand(size=(batch_dim, dim_win, dim['dim'])))	1✔
314
315	inputs_win_type.append(window)	1✔
316	inputs_win.append(dim_win)	1✔
317
318	if self.map_over_batch:	1✔
319	function_to_call = torch.func.vmap(self.param_fun,in_dims=tuple(stream_json['Functions'][self.name]['map_over_dim']))	1✔
320	else:
321	function_to_call = self.param_fun	1✔
322	out = function_to_call(*inputs)	1✔
323	out_shape = out.shape	1✔
324	check(out_shape[0] == batch_dim, ValueError, "The batch output dimension it is not correct.")	1✔
325	out_dim = list(out_shape[2:])	1✔
326	check(len(out_dim) == 1, ValueError, "The output dimension of the function is bigger than a vector.")	1✔
327	out_win_type = 'sw'	1✔
328	out_win = out_shape[1]	1✔
329	for idx, win in enumerate(inputs_win):	1✔
330	if out_shape[1] == win and all_inputs_type[idx] not in (Parameter, Constant):	1✔
331	out_win_type = inputs_win_type[idx]	1✔
332	out_win = all_inputs_dim[idx][out_win_type]	1✔
333
334	return { 'dim': out_dim[0], out_win_type : out_win }	1✔
335
336	def return_standard_inputs(json, model_def, xlim = None, num_points = 1000):	1✔
337	check(json['n_input'] == 1 or json['n_input'] == 2, ValueError, "The function must have only one or two inputs.")	1✔
338	fun_inputs = tuple()	1✔
339	for i in range(json['n_input']):	1✔
340	dim = json['in_dim'][i]	1✔
341	check(dim['dim'] == 1, ValueError, "The input dimension must be 1.")	1✔
342	if 'tw' in dim:	1✔
343	check(dim['tw'] == model_def['Info']['SampleTime'], ValueError, "The input window must be 1.")	1✔
344	elif 'sw' in dim:	1✔
345	check(dim['sw'] == 1, ValueError, "The input window must be 1.")	1✔
346	if xlim is not None:	1✔
347	if json['n_input'] == 2:	1✔
348	check(np.array(xlim).shape == (json['n_input'], 2), ValueError,	1✔
349	"The xlim must have the same shape as the number of inputs.")
350	x_value = np.linspace(xlim[i][0], xlim[i][1], num=num_points)	1✔
351	else:
352	check(np.array(xlim).shape == (2,), ValueError,	×
353	"The xlim must have the same shape as the number of inputs.")
354	x_value = np.linspace(xlim[0], xlim[1], num=num_points)	×
355	else:
356	x_value = np.linspace(0, 1, num=num_points)	1✔
357	if i == 0:	1✔
358	x0_value = torch.from_numpy(x_value)	1✔
359	else:
360	x1_value = torch.from_numpy(x_value)	1✔
361
362	if json['n_input'] == 2:	1✔
363	x0_value, x1_value = torch.meshgrid(x0_value,x1_value,indexing="xy")	1✔
364	x0_value = x0_value.flatten().unsqueeze(1).unsqueeze(1)	1✔
365	x1_value = x1_value.flatten().unsqueeze(1).unsqueeze(1)	1✔
366	fun_inputs += (x0_value,x1_value,)	1✔
367	else:
368	x0_value = x0_value.unsqueeze(1).unsqueeze(1)	1✔
369	fun_inputs += (x0_value,)	1✔
370
371	for key in json['params_and_consts']:	1✔
372	val = model_def['Parameters'][key] if key in model_def['Parameters'] else model_def['Constants'][key]	1✔
373	fun_inputs += tuple([torch.from_numpy(np.array(val['values']))]) # The vector is transform in a tuple	1✔
374
375	return fun_inputs	1✔
376
377	def return_function(json, fun_inputs):	1✔
378	exec(json['code'], globals())	1✔
379	function_to_call = globals()[json['name']]	1✔
380	output = function_to_call(*fun_inputs)	1✔
381	check(output.shape[1] == 1, ValueError, "The output dimension must be 1.")	1✔
382	check(output.shape[2] == 1, ValueError, "The output window must be 1.")	1✔
383	funinfo = inspect.getfullargspec(function_to_call)	1✔
384	return output, funinfo.args	1✔
385
386	class Parametric_Layer(nn.Module):	1✔
387	def __init__(self, func, params_and_consts, map_over_batch):	1✔
388	super().__init__()	1✔
389	self.name = func['name']	1✔
390	self.params_and_consts = params_and_consts	1✔
391	if type(map_over_batch) is list:	1✔
392	self.map_over_batch = True	1✔
393	self.input_map_dim = tuple(map_over_batch)	1✔
394	else:
395	self.map_over_batch = False	1✔
396	## Add the function to the globals
397	try:	1✔
398	code = 'import torch\n@torch.fx.wrap\n' + func['code']	1✔
399	exec(code, globals())	1✔
400	except Exception as e:	×
401	print(f"An error occurred: {e}")	×
402
403	def forward(self, *inputs):	1✔
404	args = list(inputs) + self.params_and_consts	1✔
405	# Retrieve the function object from the globals dictionary
406	function_to_call = globals()[self.name]	1✔
407	# Call the function using the retrieved function object
408	if self.map_over_batch:	1✔
409	function_to_call = torch.func.vmap(function_to_call,in_dims=self.input_map_dim)	1✔
410	result = function_to_call(*args)	1✔
411	return result	1✔
412
413	def createParamFun(self, *func_params):	1✔
414	return Parametric_Layer(func=func_params[0], params_and_consts=func_params[1], map_over_batch=func_params[2])	1✔
415
416	setattr(Model, paramfun_relation_name, createParamFun)	1✔

tonegas / nnodely / 13504505673

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