12354928461

Committed 16 Dec 2024 02:33PM UTC coverage: 46.141% (-18.9%) from 65.01%

Build # 12354928461

Build Type

Pull #30

github

Committed by

NicoRenaud

Commit Message

clean up

Pull Request Pull Request #30: Dev

Run Details

531 of 1346 new or added lines in 14 files covered. (39.45%)

843 of 1827 relevant lines covered (46.14%)

0.46 hits per line

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

71.67

/wntr_quantum/sampler/step/base_step.py

import numpy as np


class BaseStep:  # noqa: D101
    def __init__(  # noqa: D417
        self,
        var_names,
        single_var_names,
        single_var_index,
        step_size=1,
        optimize_values=None,
    ):
        """Propose a new solution vector.

        Args:
            var_names (list): names of the variables in the problem
            single_var_names (_type_): list of the single variables names e.g. x_001_002
            single_var_index (_type_): index of the single variables
            step_size (int, optional): size of the steps
            optimize_values (list, optional): index of the values to optimize
        """
        self.var_names = var_names
        self.single_var_names = single_var_names
        self.single_var_index = single_var_index
        self.num_single_var = len(self.single_var_names)
        self.high_order_terms_mapping = self.define_mapping()

        self.value_names = np.unique(
            [self._get_variable_root_name(n) for n in single_var_names]
        )
        self.index_values = {v: [] for v in self.value_names}
        for n, idx in zip(self.single_var_names, self.single_var_index):
            val = self._get_variable_root_name(n)
            self.index_values[val].append(idx)

        self.step_size = step_size
        self.optimize_values = optimize_values
        if self.optimize_values is None:
            self.optimize_values = list(np.arange(len(self.value_names)))

    @staticmethod
    def _get_variable_root_name(var_name) -> str:
        """Extract the root name of the variables.

        Args:
            var_name (str): variable name

        Returns:
            str: root name
        """
        return "_".join(var_name.split("_")[:2])

    def define_mapping(self):
        """Define the mapping of the higher order terms.

        Returns:
            list: mapping of the higher order terms
        """
        high_order_terms_mapping = []

        # loop over all the variables
        for iv, v in enumerate(self.var_names):

            # if we have a cmomposite variables e.g. x_001 * x_002 we ignore it
            if v not in self.single_var_names:
                high_order_terms_mapping.append(None)

            # if the variables is a unique one e.g. x_011
            else:
                high_order_terms_mapping.append({})
                # we loop over all the variables
                for iiv, vv in enumerate(self.var_names):
                    if v != vv:
                        if v in vv:

                            var_tmp = vv.split("*")
                            idx_terms = []
                            for vtmp in var_tmp:
                                idx = self.single_var_index[
                                    self.single_var_names.index(vtmp)
                                ]
                                idx_terms.append(idx)
                            high_order_terms_mapping[-1][iiv] = idx_terms

        return high_order_terms_mapping

    def fix_constraint(self, x, idx):
        """Ensure that the solution vectors respect quadratization.

        Args:
            x (list): sample
            idx (int): index of the element that has changed

        Returns:
            list: new sampel that respects quadratization constraints
        """
        fix_var = self.high_order_terms_mapping[idx]
        for idx_fix, idx_prods in fix_var.items():
            x[idx_fix] = np.array([x[i] for i in idx_prods]).prod()
        return x

    def verify_quadratic_constraints(self, data):
        """Check if quadratic constraints are respected or not.

        Args:
            data (list): sample
        """
        for v, d in zip(self.var_names, data):
            if v not in self.single_var_names:
                var_tmp = v.split("*")
                itmp = 0
                for vtmp in var_tmp:
                    idx = self.single_var_index[self.single_var_names.index(vtmp)]
                    if itmp == 0:
                        dcomposite = data[idx]
                        itmp = 1
                    else:
                        dcomposite *= data[idx]
                if d != dcomposite:
                    print("Error in the quadratic contraints")
                    print("%s = %d" % (v, d))
                    for vtmp in var_tmp:
                        idx = self.single_var_index[self.single_var_names.index(vtmp)]
                        print("%s = %d" % (vtmp, data[idx]))

    def __call__(self, x, verbose=False):
        """Call function of the method.

        Args:
            x (list): sample
            verbose (bool): print stuff

        Returns:
            list: new sample
        """
        raise NotImplementedError("Implement a __call__ method")

1	import numpy as np	1✔
2
3
4	class BaseStep: # noqa: D101	1✔
5	def __init__( # noqa: D417	1✔
6	self,
7	var_names,
8	single_var_names,
9	single_var_index,
10	step_size=1,
11	optimize_values=None,
12	):
13	"""Propose a new solution vector.
14
15	Args:
16	var_names (list): names of the variables in the problem
17	single_var_names (_type_): list of the single variables names e.g. x_001_002
18	single_var_index (_type_): index of the single variables
19	step_size (int, optional): size of the steps
20	optimize_values (list, optional): index of the values to optimize
21	"""
22	self.var_names = var_names	1✔
23	self.single_var_names = single_var_names	1✔
24	self.single_var_index = single_var_index	1✔
25	self.num_single_var = len(self.single_var_names)	1✔
26	self.high_order_terms_mapping = self.define_mapping()	1✔
27
28	self.value_names = np.unique(	1✔
29	[self._get_variable_root_name(n) for n in single_var_names]
30	)
31	self.index_values = {v: [] for v in self.value_names}	1✔
32	for n, idx in zip(self.single_var_names, self.single_var_index):	1✔
33	val = self._get_variable_root_name(n)	1✔
34	self.index_values[val].append(idx)	1✔
35
36	self.step_size = step_size	1✔
37	self.optimize_values = optimize_values	1✔
38	if self.optimize_values is None:	1✔
39	self.optimize_values = list(np.arange(len(self.value_names)))	1✔
40
41	@staticmethod	1✔
42	def _get_variable_root_name(var_name) -> str:	1✔
43	"""Extract the root name of the variables.
44
45	Args:
46	var_name (str): variable name
47
48	Returns:
49	str: root name
50	"""
51	return "_".join(var_name.split("_")[:2])	1✔
52
53	def define_mapping(self):	1✔
54	"""Define the mapping of the higher order terms.
55
56	Returns:
57	list: mapping of the higher order terms
58	"""
59	high_order_terms_mapping = []	1✔
60
61	# loop over all the variables
62	for iv, v in enumerate(self.var_names):	1✔
63
64	# if we have a cmomposite variables e.g. x_001 * x_002 we ignore it
65	if v not in self.single_var_names:	1✔
66	high_order_terms_mapping.append(None)	1✔
67
68	# if the variables is a unique one e.g. x_011
69	else:
70	high_order_terms_mapping.append({})	1✔
71	# we loop over all the variables
72	for iiv, vv in enumerate(self.var_names):	1✔
73	if v != vv:	1✔
74	if v in vv:	1✔
75
76	var_tmp = vv.split("*")	1✔
77	idx_terms = []	1✔
78	for vtmp in var_tmp:	1✔
79	idx = self.single_var_index[	1✔
80	self.single_var_names.index(vtmp)
81	]
82	idx_terms.append(idx)	1✔
83	high_order_terms_mapping[-1][iiv] = idx_terms	1✔
84
85	return high_order_terms_mapping	1✔
86
87	def fix_constraint(self, x, idx):	1✔
88	"""Ensure that the solution vectors respect quadratization.
89
90	Args:
91	x (list): sample
92	idx (int): index of the element that has changed
93
94	Returns:
95	list: new sampel that respects quadratization constraints
96	"""
97	fix_var = self.high_order_terms_mapping[idx]	1✔
98	for idx_fix, idx_prods in fix_var.items():	1✔
99	x[idx_fix] = np.array([x[i] for i in idx_prods]).prod()	1✔
100	return x	1✔
101
102	def verify_quadratic_constraints(self, data):	1✔
103	"""Check if quadratic constraints are respected or not.
104
105	Args:
106	data (list): sample
107	"""
NEW 108	for v, d in zip(self.var_names, data):	×
NEW 109	if v not in self.single_var_names:	×
NEW 110	var_tmp = v.split("*")	×
NEW 111	itmp = 0	×
NEW 112	for vtmp in var_tmp:	×
NEW 113	idx = self.single_var_index[self.single_var_names.index(vtmp)]	×
NEW 114	if itmp == 0:	×
NEW 115	dcomposite = data[idx]	×
NEW 116	itmp = 1	×
117	else:
NEW 118	dcomposite *= data[idx]	×
NEW 119	if d != dcomposite:	×
NEW 120	print("Error in the quadratic contraints")	×
NEW 121	print("%s = %d" % (v, d))	×
NEW 122	for vtmp in var_tmp:	×
NEW 123	idx = self.single_var_index[self.single_var_names.index(vtmp)]	×
NEW 124	print("%s = %d" % (vtmp, data[idx]))	×
125
126	def __call__(self, x, verbose=False):	1✔
127	"""Call function of the method.
128
129	Args:
130	x (list): sample
131	verbose (bool): print stuff
132
133	Returns:
134	list: new sample
135	"""
NEW 136	raise NotImplementedError("Implement a __call__ method")	×

QuantumApplicationLab / wntr-quantum / 12354928461

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