13521235900

Committed 25 Feb 2025 12:29PM UTC coverage: 95.415% (+0.02%) from 95.395%

Build # 13521235900

Build Type

Pull #63

github

Committed by

tonegas

Commit Message

merge minor fix

Pull Request Pull Request #63: 49 general bug fix

Run Details

222 of 224 new or added lines in 18 files covered. (99.11%)

12 existing lines in 2 files now uncovered.

10550 of 11057 relevant lines covered (95.41%)

0.95 hits per line

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

96.92

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

tonegas / nnodely / 13521235900

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