13531366980

Committed 25 Feb 2025 09:32PM UTC coverage: 95.399% (-0.01%) from 95.411%

Build # 13531366980

Build Type

Pull #65

github

Committed by

web-flow

Commit Message

Merge 75d565ccc into 8d0789317

Pull Request Pull Request #65: 49 general bug fix

Run Details

46 of 53 new or added lines in 8 files covered. (86.79%)

3 existing lines in 3 files now uncovered.

10574 of 11084 relevant lines covered (95.4%)

0.95 hits per line

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

92.75

/nnodely/relation.py

import copy

import numpy as np

from nnodely.utils import check, merge, enforce_types, ForbiddenTags

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

MAIN_JSON = {
                'Info' : {},
                'Inputs' : {},
                'States' : {},
                'Constants': {},
                'Parameters' : {},
                'Functions' : {},
                'Relations': {},
                'Outputs': {}
            }

CHECK_NAMES = True

def toStream(obj):
    from nnodely.parameter import Parameter, Constant
    if type(obj) in (int,float,list,np.ndarray):
        obj = Constant('Constant'+str(NeuObj.count), obj)
        #obj = Stream(obj, MAIN_JSON, {'dim': 1}) if type(obj) in (int, float) else obj
    if type(obj) is Parameter or type(obj) is Constant:
        obj = Stream(obj.name, obj.json, obj.dim)
    return obj


class NeuObj():
    count = 0
    names = []
    @classmethod
    @enforce_types
    def clearNames(self, names:str|list|None=None):
        if names is None:
            NeuObj.count = 0
            NeuObj.names = []
        else:
            if type(names) is list:
                for name in names:
                    if name in NeuObj.names:
                        NeuObj.names.remove(name)
            else:
                if names in NeuObj.names:
                    NeuObj.names.remove(names)
    def __init__(self, name='', json={}, dim=0):
        NeuObj.count += 1
        if name == '':
            name = 'Auto'+str(NeuObj.count)
        if CHECK_NAMES == True:
            check(name not in NeuObj.names, NameError, f"The name {name} is already used change the name of NeuObj.")
            check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")
            NeuObj.names.append(name)
        self.name = name
        self.dim = dim
        if json:
            self.json = copy.deepcopy(json)
        else:
            self.json = copy.deepcopy(MAIN_JSON)

class Relation():
    def __add__(self, obj):
        from nnodely.arithmetic import Add
        return Add(self, obj)

    def __sub__(self, obj):
        from nnodely.arithmetic import Sub
        return Sub(self, obj)

    def __truediv__(self, obj):
        from nnodely.arithmetic import Div
        return Div(self, obj)

    def __mul__(self, obj):
        from nnodely.arithmetic import Mul
        return Mul(self, obj)

    def __pow__(self, obj):
        from nnodely.arithmetic import Pow
        return Pow(self, obj)

    def __neg__(self):
        from nnodely.arithmetic import Neg
        return Neg(self)

class Stream(Relation):
    count = 0
    @classmethod
    def resetCount(self):
        Stream.count = 0

    def __init__(self, name, json, dim, count = 1):
        Stream.count += count
        check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")
        self.name = name
        self.json = copy.deepcopy(json)
        self.dim = dim

    @enforce_types
    def tw(self, tw:float|int|list, offset:float|int|None = None) -> "Stream":
        """
        Selects a time window on Stream. It is possible to create a smaller or bigger time window on the stream.
        The Time Window must be in the past not in the future.

        Parameters
        ----------
        tw : float, int, list
            The time window represents the time in the past. If a list, it should contain the start and end times, both indexes must be in the past.
        offset : float, int, optional
            The offset for the sample window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the TimePart object with the selected time window.

        """
        from nnodely.input import State, Connect
        if type(tw) is list:
            check(0 >= tw[1] > tw[0] and tw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")
        s = State(self.name+"_state",dimensions=self.dim['dim'])
        out_connect = Connect(self, s)
        win_state = s.tw(tw, offset)
        return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )

    @enforce_types
    def sw(self, sw:int|list, offset:int|None = None) -> "Stream":
        """
        Selects a sample window on Stream. It is possible to create a smaller or bigger window on the stream.
        The Sample Window must be in the past not in the future.

        Parameters
        ----------
        sw : int, list
            The sample window represents the number of steps in the past. If a list, it should contain the start and end indices, both indexes must be in the past.
        offset : int, optional
            The offset for the sample window. Default is None.

        Returns
        -------
        Stream
            A Stream representing the SamplePart object with the selected samples.

        """
        from nnodely.input import State, Connect
        if type(sw) is list:
            check(0 >= sw[1] > sw[0] and sw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")
        s = State(self.name+"_state",dimensions=self.dim['dim'])
        out_connect = Connect(self, s)
        win_state = s.sw(sw, offset)
        return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )

    @enforce_types
    def z(self, delay:int|float) -> "Stream":
        """
        Considering the Zeta transform notation. The function is used to delay a Stream.
        The value of the delay can be only positive.

        Parameters
        ----------
        delay : int
            The delay value.

        Returns
        -------
        Stream
            A Stream representing the delayed Stream
        """
        check(delay > 0, ValueError, "The delay must be a positive integer")
        check('sw' in self.dim, TypeError, "The stream is not defined in samples but in time")
        return self.sw([-self.dim['sw']-delay,-delay])

    @enforce_types
    def delay(self, delay:int|float) -> "Stream":
        """
        The function is used to delay a Stream.
        The value of the delay can be only positive.

        Parameters
        ----------
        delay : int, float
            The delay value.

        Returns
        -------
        Stream
            A Stream representing the delayed Stream
        """
        check(delay > 0, ValueError, "The delay must be a positive integer")
        check('tw' in self.dim, TypeError, "The stream is not defined in time but in sample")
        return self.tw([-self.dim['tw']-delay,-delay])

    @enforce_types
    def s(self, order:int, method:str = 'ForwardEuler') -> "Stream":
        """
        Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on a Stream.
        The order of the integral or the derivative operation is indicated by the order parameter.

        Parameters
        ----------
        order : int
            Order of the Laplace transform
        method : str, optional
            Integration or derivation method

        Returns
        -------
        Stream
            A Stream of the signal represents the integral or derivation operation.
        """
        from nnodely.timeoperation import Derivate, Integrate
        check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")
        if order > 0:
            for i in range(order):
                o = Derivate(self, method = method)
        elif order < 0:
            for i in range(-order):
                o = Integrate(self, method = method)
        return o

    def connect(self, obj) -> "Stream":
        """
        Connects the current stream to a given state object.

        Parameters
        ----------
        obj : State
            The state object to connect to.

        Returns
        -------
        Stream
            A new Stream object representing the connected state.

        Raises
        ------
        TypeError
            If the provided object is not of type State.
        KeyError
            If the state variable is already connected.
        """
        from nnodely.input import State
        check(type(obj) is State, TypeError,
              f"The {obj} must be a State and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name], KeyError,
              f"The state variable {obj.name} is already connected.")
        self.json['States'][obj.name]['connect'] = self.name
        return Stream(self.name, self.json, self.dim,0 )

    def closedLoop(self, obj) -> "Stream":
        """
        Creates a closed loop connection with a given state object.

        Parameters
        ----------
        obj : State
            The state object to create a closed loop with.

        Returns
        -------
        Stream
            A new Stream object representing the closed loop state.

        Raises
        ------
        TypeError
            If the provided object is not of type State.
        KeyError
            If the state variable is already connected.
        """
        from nnodely.input import State
        check(type(obj) is State, TypeError,
              f"The {obj} must be a State and not a {type(obj)}.")
        self.json = merge(self.json, obj.json)
        check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name],
              KeyError,
              f"The state variable {obj.name} is already connected.")
        self.json['States'][obj.name]['closedLoop'] = self.name
        return Stream(self.name, self.json, self.dim,0 )

class ToStream():
    def __new__(cls, *args, **kwargs):
        out = super(ToStream,cls).__new__(cls)
        out.__init__(*args, **kwargs)
        return Stream(out.name,out.json,out.dim,0)

class AutoToStream():
    def __new__(cls, *args,  **kwargs):
        if len(args) > 0 and (issubclass(type(args[0]),NeuObj) or type(args[0]) is Stream):
            instance = super().__new__(cls)
            #instance.__init__(**kwargs)
            instance.__init__()
            return instance(args[0])
        instance = super().__new__(cls)
        return instance

1	import copy	1✔
2
3	import numpy as np	1✔
4
5	from nnodely.utils import check, merge, enforce_types, ForbiddenTags	1✔
6
7	from nnodely.logger import logging, nnLogger	1✔
8	log = nnLogger(__name__, logging.CRITICAL)	1✔
9
10	MAIN_JSON = {	1✔
11	'Info' : {},
12	'Inputs' : {},
13	'States' : {},
14	'Constants': {},
15	'Parameters' : {},
16	'Functions' : {},
17	'Relations': {},
18	'Outputs': {}
19	}
20
21	CHECK_NAMES = True	1✔
22
23	def toStream(obj):	1✔
24	from nnodely.parameter import Parameter, Constant	1✔
25	if type(obj) in (int,float,list,np.ndarray):	1✔
26	obj = Constant('Constant'+str(NeuObj.count), obj)	1✔
27	#obj = Stream(obj, MAIN_JSON, {'dim': 1}) if type(obj) in (int, float) else obj
28	if type(obj) is Parameter or type(obj) is Constant:	1✔
29	obj = Stream(obj.name, obj.json, obj.dim)	1✔
30	return obj	1✔
31
32
33	class NeuObj():	1✔
34	count = 0	1✔
35	names = []	1✔
36	@classmethod	1✔
37	@enforce_types	1✔
38	def clearNames(self, names:str\|list\|None=None):	1✔
39	if names is None:	1✔
40	NeuObj.count = 0	1✔
41	NeuObj.names = []	1✔
42	else:
NEW 43	if type(names) is list:	×
NEW 44	for name in names:	×
NEW 45	if name in NeuObj.names:	×
NEW 46	NeuObj.names.remove(name)	×
47	else:
NEW 48	if names in NeuObj.names:	×
NEW 49	NeuObj.names.remove(names)	×
50	def __init__(self, name='', json={}, dim=0):	1✔
51	NeuObj.count += 1	1✔
52	if name == '':	1✔
53	name = 'Auto'+str(NeuObj.count)	×
54	if CHECK_NAMES == True:	1✔
55	check(name not in NeuObj.names, NameError, f"The name {name} is already used change the name of NeuObj.")	×
56	check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")	×
57	NeuObj.names.append(name)	×
58	self.name = name	1✔
59	self.dim = dim	1✔
60	if json:	1✔
61	self.json = copy.deepcopy(json)	1✔
62	else:
63	self.json = copy.deepcopy(MAIN_JSON)	1✔
64
65	class Relation():	1✔
66	def __add__(self, obj):	1✔
67	from nnodely.arithmetic import Add	1✔
68	return Add(self, obj)	1✔
69
70	def __sub__(self, obj):	1✔
71	from nnodely.arithmetic import Sub	1✔
72	return Sub(self, obj)	1✔
73
74	def __truediv__(self, obj):	1✔
75	from nnodely.arithmetic import Div	1✔
76	return Div(self, obj)	1✔
77
78	def __mul__(self, obj):	1✔
79	from nnodely.arithmetic import Mul	1✔
80	return Mul(self, obj)	1✔
81
82	def __pow__(self, obj):	1✔
83	from nnodely.arithmetic import Pow	1✔
84	return Pow(self, obj)	1✔
85
86	def __neg__(self):	1✔
87	from nnodely.arithmetic import Neg	1✔
88	return Neg(self)	1✔
89
90	class Stream(Relation):	1✔
91	count = 0	1✔
92	@classmethod	1✔
93	def resetCount(self):	1✔
94	Stream.count = 0	1✔
95
96	def __init__(self, name, json, dim, count = 1):	1✔
97	Stream.count += count	1✔
98	check(name not in ForbiddenTags, NameError, f"The name '{name}' is a forbidden tag.")	1✔
99	self.name = name	1✔
100	self.json = copy.deepcopy(json)	1✔
101	self.dim = dim	1✔
102
103	@enforce_types	1✔
104	def tw(self, tw:float\|int\|list, offset:float\|int\|None = None) -> "Stream":	1✔
105	"""
106	Selects a time window on Stream. It is possible to create a smaller or bigger time window on the stream.
107	The Time Window must be in the past not in the future.
108
109	Parameters
110	----------
111	tw : float, int, list
112	The time window represents the time in the past. If a list, it should contain the start and end times, both indexes must be in the past.
113	offset : float, int, optional
114	The offset for the sample window. Default is None.
115
116	Returns
117	-------
118	Stream
119	A Stream representing the TimePart object with the selected time window.
120
121	"""
122	from nnodely.input import State, Connect	1✔
123	if type(tw) is list:	1✔
124	check(0 >= tw[1] > tw[0] and tw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
125	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
126	out_connect = Connect(self, s)	1✔
127	win_state = s.tw(tw, offset)	1✔
128	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
129
130	@enforce_types	1✔
131	def sw(self, sw:int\|list, offset:int\|None = None) -> "Stream":	1✔
132	"""
133	Selects a sample window on Stream. It is possible to create a smaller or bigger window on the stream.
134	The Sample Window must be in the past not in the future.
135
136	Parameters
137	----------
138	sw : int, list
139	The sample window represents the number of steps in the past. If a list, it should contain the start and end indices, both indexes must be in the past.
140	offset : int, optional
141	The offset for the sample window. Default is None.
142
143	Returns
144	-------
145	Stream
146	A Stream representing the SamplePart object with the selected samples.
147
148	"""
149	from nnodely.input import State, Connect	1✔
150	if type(sw) is list:	1✔
151	check(0 >= sw[1] > sw[0] and sw[0] < 0, ValueError, "The dimension of the sample window must be in the past.")	1✔
152	s = State(self.name+"_state",dimensions=self.dim['dim'])	1✔
153	out_connect = Connect(self, s)	1✔
154	win_state = s.sw(sw, offset)	1✔
155	return Stream(win_state.name, merge(win_state.json, out_connect.json), win_state.dim,0 )	1✔
156
157	@enforce_types	1✔
158	def z(self, delay:int\|float) -> "Stream":	1✔
159	"""
160	Considering the Zeta transform notation. The function is used to delay a Stream.
161	The value of the delay can be only positive.
162
163	Parameters
164	----------
165	delay : int
166	The delay value.
167
168	Returns
169	-------
170	Stream
171	A Stream representing the delayed Stream
172	"""
173	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
174	check('sw' in self.dim, TypeError, "The stream is not defined in samples but in time")	1✔
175	return self.sw([-self.dim['sw']-delay,-delay])	1✔
176
177	@enforce_types	1✔
178	def delay(self, delay:int\|float) -> "Stream":	1✔
179	"""
180	The function is used to delay a Stream.
181	The value of the delay can be only positive.
182
183	Parameters
184	----------
185	delay : int, float
186	The delay value.
187
188	Returns
189	-------
190	Stream
191	A Stream representing the delayed Stream
192	"""
193	check(delay > 0, ValueError, "The delay must be a positive integer")	1✔
194	check('tw' in self.dim, TypeError, "The stream is not defined in time but in sample")	1✔
195	return self.tw([-self.dim['tw']-delay,-delay])	1✔
196
197	@enforce_types	1✔
198	def s(self, order:int, method:str = 'ForwardEuler') -> "Stream":	1✔
199	"""
200	Considering the Laplace transform notation. The function is used to operate an integral or derivate operation on a Stream.
201	The order of the integral or the derivative operation is indicated by the order parameter.
202
203	Parameters
204	----------
205	order : int
206	Order of the Laplace transform
207	method : str, optional
208	Integration or derivation method
209
210	Returns
211	-------
212	Stream
213	A Stream of the signal represents the integral or derivation operation.
214	"""
215	from nnodely.timeoperation import Derivate, Integrate	1✔
216	check(order != 0, ValueError, "The order must be a positive or negative integer not a zero")	1✔
217	if order > 0:	1✔
218	for i in range(order):	1✔
219	o = Derivate(self, method = method)	1✔
220	elif order < 0:	1✔
221	for i in range(-order):	1✔
222	o = Integrate(self, method = method)	1✔
223	return o	1✔
224
225	def connect(self, obj) -> "Stream":	1✔
226	"""
227	Connects the current stream to a given state object.
228
229	Parameters
230	----------
231	obj : State
232	The state object to connect to.
233
234	Returns
235	-------
236	Stream
237	A new Stream object representing the connected state.
238
239	Raises
240	------
241	TypeError
242	If the provided object is not of type State.
243	KeyError
244	If the state variable is already connected.
245	"""
246	from nnodely.input import State	1✔
247	check(type(obj) is State, TypeError,	1✔
248	f"The {obj} must be a State and not a {type(obj)}.")
249	self.json = merge(self.json, obj.json)	1✔
250	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name], KeyError,	1✔
251	f"The state variable {obj.name} is already connected.")
252	self.json['States'][obj.name]['connect'] = self.name	1✔
253	return Stream(self.name, self.json, self.dim,0 )	1✔
254
255	def closedLoop(self, obj) -> "Stream":	1✔
256	"""
257	Creates a closed loop connection with a given state object.
258
259	Parameters
260	----------
261	obj : State
262	The state object to create a closed loop with.
263
264	Returns
265	-------
266	Stream
267	A new Stream object representing the closed loop state.
268
269	Raises
270	------
271	TypeError
272	If the provided object is not of type State.
273	KeyError
274	If the state variable is already connected.
275	"""
276	from nnodely.input import State	1✔
277	check(type(obj) is State, TypeError,	1✔
278	f"The {obj} must be a State and not a {type(obj)}.")
279	self.json = merge(self.json, obj.json)	1✔
280	check('closedLoop' not in self.json['States'][obj.name] or 'connect' not in self.json['States'][obj.name],	1✔
281	KeyError,
282	f"The state variable {obj.name} is already connected.")
283	self.json['States'][obj.name]['closedLoop'] = self.name	1✔
284	return Stream(self.name, self.json, self.dim,0 )	1✔
285
286	class ToStream():	1✔
287	def __new__(cls, args, *kwargs):	1✔
288	out = super(ToStream,cls).__new__(cls)	1✔
289	out.__init__(args, *kwargs)	1✔
290	return Stream(out.name,out.json,out.dim,0)	1✔
291
292	class AutoToStream():	1✔
293	def __new__(cls, args, *kwargs):	1✔
294	if len(args) > 0 and (issubclass(type(args[0]),NeuObj) or type(args[0]) is Stream):	1✔
295	instance = super().__new__(cls)	1✔
296	#instance.__init__(**kwargs)
297	instance.__init__()	1✔
298	return instance(args[0])	1✔
299	instance = super().__new__(cls)	1✔
300	return instance	1✔

tonegas / nnodely / 13531366980

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