18305449975

Committed 07 Oct 2025 07:32AM UTC coverage: 97.691% (-0.04%) from 97.727%

Build # 18305449975

Build Type

push

github

Committed by

tonegas

Commit Message

Modified the version

Run Details

1 of 1 new or added line in 1 file covered. (100.0%)

38 existing lines in 5 files now uncovered.

12733 of 13034 relevant lines covered (97.69%)

0.98 hits per line

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

87.72

/nnodely/support/utils.py

import torch, inspect
import types

from collections import OrderedDict

import numpy as np
from functools import wraps
from typing import get_type_hints
import keyword

TORCH_DTYPE = torch.float32
NP_DTYPE = np.float32

ForbiddenTags = keyword.kwlist

class ReadOnlyDict:
    def __init__(self, data):
        self._data = data

    def __getitem__(self, key):
        value = self._data[key]
        if isinstance(value, dict):
            return dict(ReadOnlyDict(value))
        return value

    def __len__(self):
        return len(self._data)

    def __iter__(self):
        return iter(self._data)

    def keys(self):
        return self._data.keys()

    def items(self):
        return self._data.items()

    def values(self):
        return self._data.values()

    def __repr__(self):
        from pprint import pformat
        return pformat(self._data)

    def __or__(self, other):
        if not isinstance(other, ReadOnlyDict):
            return NotImplemented
        combined_data = {**self._data, **other._data}
        return ReadOnlyDict(combined_data)

    def __str__(self):
        from nnodely.visualizer.emptyvisualizer import color, GREEN
        from pprint import pformat
        return color(pformat(self._data), GREEN)

    def __eq__(self, other):
        if not isinstance(other, ReadOnlyDict):
            return self._data == other
        return self._data == other._data

class ParamDict(ReadOnlyDict):
    def __init__(self, data, internal_data = None):
        super().__init__(data)
        self._internal_data = internal_data if internal_data is not None else {}

    def __setitem__(self, key, value):
        self._data[key]['values'] = value
        self._internal_data[key] = self._internal_data[key].new_tensor(value)

    def __getitem__(self, key):
        value = self._data[key]['values'] if 'values' in self._data[key] else None
        return value

def enforce_types(func):
    @wraps(func)
    def wrapper(*args, **kwargs):
        hints = get_type_hints(func)
        all_args = kwargs.copy()

        sig = OrderedDict(inspect.signature(func).parameters)
        if len(sig) != len(args):
            var_type = None
            for ind, arg in enumerate(args):
                if ind < len(list(sig.values())) and list(sig.values())[ind].kind == inspect.Parameter.VAR_POSITIONAL:
                    var_name = list(sig.keys())[ind]
                    var_type = sig.pop(var_name)
                if var_type:
                    sig[var_name+str(ind)] = var_type

        all_args.update(dict(zip(sig, args)))
        if 'self' in sig.keys():
            sig.pop('self')

        for arg_name, arg in all_args.items():
            if (arg_name in hints.keys() or arg_name in sig.keys()) and not isinstance(arg,sig[arg_name].annotation):
                class_name = func.__qualname__.split('.')[0]
                if isinstance(sig[arg_name].annotation, types.UnionType):
                    type_list = [val.__name__ for val in sig[arg_name].annotation.__args__]
                else:
                    type_list = sig[arg_name].annotation.__name__
                raise TypeError(
                    f"In Function or Class {class_name} the argument '{arg_name}' to be of type {type_list}, but got {type(arg).__name__}")

        # for arg, arg_type in hints.items():
        #     if arg in all_args and not isinstance(all_args[arg], arg_type):
        #         raise TypeError(
        #             f"In Function or Class {func} Expected argument '{arg}' to be of type {arg_type}, but got {type(all_args[arg]).__name__}")

        return func(*args, **kwargs)

    return wrapper

def is_notebook():
    try:
        from IPython import get_ipython
        if 'IPKernelApp' in get_ipython().config:
            return True  # È un notebook
    except Exception:
        pass
    return False  # È uno script

def tensor_to_list(data):
    if isinstance(data, torch.Tensor):
        # Converte il tensore in una lista
        return data.tolist()
    elif isinstance(data, dict):
        # Ricorsione per i dizionari
        return {key: tensor_to_list(value) for key, value in data.items()}
    elif isinstance(data, list):
        # Ricorsione per le liste
        return [tensor_to_list(item) for item in data]
    elif isinstance(data, tuple):
        # Ricorsione per tuple
        return tuple(tensor_to_list(item) for item in data)
    elif isinstance(data, torch.nn.modules.container.ParameterDict):
        # Ricorsione per parameter dict
        return {key: tensor_to_list(value) for key, value in data.items()}
    else:
        # Altri tipi di dati rimangono invariati
        return data

def get_batch_size(n_samples, batch_size = None, predicion_samples = 0):
    batch_size = batch_size if batch_size is not None else n_samples
    predicion_samples = 0 if predicion_samples == -1 else predicion_samples #This value is used to disconnect the connect
    return batch_size if batch_size <= n_samples - predicion_samples else max(0, n_samples - predicion_samples)

def check_and_get_list(name_list, available_names, error_fun):
    if type(name_list) is str:
        name_list = [name_list]
    if type(name_list) is list:
        for name in name_list:
            check(name in available_names, IndexError,  error_fun(name))
    return name_list

def check(condition, exception, string):
    if not condition:
        raise exception(string)

# Function used to verified the number of gradient operations in the graph
# def count_gradient_operations(grad_fn):
#     count = 0
#     if grad_fn is None:
#         return count
#     nodes = [grad_fn]
#     while nodes:
#         node = nodes.pop()
#         count += 1
#         nodes.extend(next_fn[0] for next_fn in node.next_functions if next_fn[0] is not None)
#     return count

# def check_gradient_operations(X:dict):
#     count = 0
#     for key in X.keys():
#         count += count_gradient_operations(X[key].grad_fn)
#     return count

1	import torch, inspect	1✔
2	import types	1✔
3
4	from collections import OrderedDict	1✔
5
6	import numpy as np	1✔
7	from functools import wraps	1✔
8	from typing import get_type_hints	1✔
9	import keyword	1✔
10
11	TORCH_DTYPE = torch.float32	1✔
12	NP_DTYPE = np.float32	1✔
13
14	ForbiddenTags = keyword.kwlist	1✔
15
16	class ReadOnlyDict:	1✔
17	def __init__(self, data):	1✔
18	self._data = data	1✔
19
20	def __getitem__(self, key):	1✔
21	value = self._data[key]	1✔
22	if isinstance(value, dict):	1✔
23	return dict(ReadOnlyDict(value))	1✔
24	return value	1✔
25
26	def __len__(self):	1✔
27	return len(self._data)	1✔
28
29	def __iter__(self):	1✔
30	return iter(self._data)	1✔
31
32	def keys(self):	1✔
33	return self._data.keys()	1✔
34
35	def items(self):	1✔
36	return self._data.items()	1✔
37
38	def values(self):	1✔
UNCOV 39	return self._data.values()	×
40
41	def __repr__(self):	1✔
42	from pprint import pformat	1✔
43	return pformat(self._data)	1✔
44
45	def __or__(self, other):	1✔
UNCOV 46	if not isinstance(other, ReadOnlyDict):	×
UNCOV 47	return NotImplemented	×
UNCOV 48	combined_data = {self._data, other._data}	×
49	return ReadOnlyDict(combined_data)	×
50
51	def __str__(self):	1✔
52	from nnodely.visualizer.emptyvisualizer import color, GREEN	×
UNCOV 53	from pprint import pformat	×
UNCOV 54	return color(pformat(self._data), GREEN)	×
55
56	def __eq__(self, other):	1✔
57	if not isinstance(other, ReadOnlyDict):	×
UNCOV 58	return self._data == other	×
UNCOV 59	return self._data == other._data	×
60
61	class ParamDict(ReadOnlyDict):	1✔
62	def __init__(self, data, internal_data = None):	1✔
63	super().__init__(data)	1✔
64	self._internal_data = internal_data if internal_data is not None else {}	1✔
65
66	def __setitem__(self, key, value):	1✔
67	self._data[key]['values'] = value	1✔
68	self._internal_data[key] = self._internal_data[key].new_tensor(value)	1✔
69
70	def __getitem__(self, key):	1✔
71	value = self._data[key]['values'] if 'values' in self._data[key] else None	1✔
72	return value	1✔
73
74	def enforce_types(func):	1✔
75	@wraps(func)	1✔
76	def wrapper(args, *kwargs):	1✔
77	hints = get_type_hints(func)	1✔
78	all_args = kwargs.copy()	1✔
79
80	sig = OrderedDict(inspect.signature(func).parameters)	1✔
81	if len(sig) != len(args):	1✔
82	var_type = None	1✔
83	for ind, arg in enumerate(args):	1✔
84	if ind < len(list(sig.values())) and list(sig.values())[ind].kind == inspect.Parameter.VAR_POSITIONAL:	1✔
85	var_name = list(sig.keys())[ind]	1✔
86	var_type = sig.pop(var_name)	1✔
87	if var_type:	1✔
88	sig[var_name+str(ind)] = var_type	1✔
89
90	all_args.update(dict(zip(sig, args)))	1✔
91	if 'self' in sig.keys():	1✔
92	sig.pop('self')	1✔
93
94	for arg_name, arg in all_args.items():	1✔
95	if (arg_name in hints.keys() or arg_name in sig.keys()) and not isinstance(arg,sig[arg_name].annotation):	1✔
96	class_name = func.__qualname__.split('.')[0]	1✔
97	if isinstance(sig[arg_name].annotation, types.UnionType):	1✔
98	type_list = [val.__name__ for val in sig[arg_name].annotation.__args__]	1✔
99	else:
100	type_list = sig[arg_name].annotation.__name__	1✔
101	raise TypeError(	1✔
102	f"In Function or Class {class_name} the argument '{arg_name}' to be of type {type_list}, but got {type(arg).__name__}")
103
104	# for arg, arg_type in hints.items():
105	# if arg in all_args and not isinstance(all_args[arg], arg_type):
106	# raise TypeError(
107	# f"In Function or Class {func} Expected argument '{arg}' to be of type {arg_type}, but got {type(all_args[arg]).__name__}")
108
109	return func(args, *kwargs)	1✔
110
111	return wrapper	1✔
112
113	def is_notebook():	1✔
114	try:	1✔
115	from IPython import get_ipython	1✔
UNCOV 116	if 'IPKernelApp' in get_ipython().config:	×
UNCOV 117	return True # È un notebook	×
118	except Exception:	1✔
119	pass	1✔
120	return False # È uno script	1✔
121
122	def tensor_to_list(data):	1✔
123	if isinstance(data, torch.Tensor):	1✔
124	# Converte il tensore in una lista
125	return data.tolist()	1✔
126	elif isinstance(data, dict):	1✔
127	# Ricorsione per i dizionari
128	return {key: tensor_to_list(value) for key, value in data.items()}	1✔
129	elif isinstance(data, list):	1✔
130	# Ricorsione per le liste
131	return [tensor_to_list(item) for item in data]	1✔
132	elif isinstance(data, tuple):	1✔
133	# Ricorsione per tuple
UNCOV 134	return tuple(tensor_to_list(item) for item in data)	×
135	elif isinstance(data, torch.nn.modules.container.ParameterDict):	1✔
136	# Ricorsione per parameter dict
137	return {key: tensor_to_list(value) for key, value in data.items()}	1✔
138	else:
139	# Altri tipi di dati rimangono invariati
140	return data	1✔
141
142	def get_batch_size(n_samples, batch_size = None, predicion_samples = 0):	1✔
143	batch_size = batch_size if batch_size is not None else n_samples	1✔
144	predicion_samples = 0 if predicion_samples == -1 else predicion_samples #This value is used to disconnect the connect	1✔
145	return batch_size if batch_size <= n_samples - predicion_samples else max(0, n_samples - predicion_samples)	1✔
146
147	def check_and_get_list(name_list, available_names, error_fun):	1✔
148	if type(name_list) is str:	1✔
149	name_list = [name_list]	1✔
150	if type(name_list) is list:	1✔
151	for name in name_list:	1✔
152	check(name in available_names, IndexError, error_fun(name))	1✔
153	return name_list	1✔
154
155	def check(condition, exception, string):	1✔
156	if not condition:	1✔
157	raise exception(string)	1✔
158
159	# Function used to verified the number of gradient operations in the graph
160	# def count_gradient_operations(grad_fn):
161	# count = 0
162	# if grad_fn is None:
163	# return count
164	# nodes = [grad_fn]
165	# while nodes:
166	# node = nodes.pop()
167	# count += 1
168	# nodes.extend(next_fn[0] for next_fn in node.next_functions if next_fn[0] is not None)
169	# return count
170
171	# def check_gradient_operations(X:dict):
172	# count = 0
173	# for key in X.keys():
174	# count += count_gradient_operations(X[key].grad_fn)
175	# return count

tonegas / nnodely / 18305449975

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