Skip to content
Snippets Groups Projects
Commit 4bb9966d authored by chsieh16's avatar chsieh16
Browse files

Rename feature variables and parse tree

parent d84105f4
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
Showing
with 107 additions and 77 deletions
dtree_learner.py
+ 107
− 77
View file @ 4bb9966d
from typing import List, Tuple
import numpy as np
import os
from learner_base import LearnerBase
from collections import OrderedDict
import csv
import grammar_generation_utils
import itertools
import json
import os
from typing import Any, List, Optional, Tuple
import numpy as np
from learner_base import LearnerBase
class DTreeLearner(LearnerBase):
......@@ -15,7 +18,7 @@ class DTreeLearner(LearnerBase):
self._state_dim: int = state_dim
self._perc_dim: int = perc_dim
self._s2f_func_list: List = []
self._s2f_func = lambda x: x
# check directory name exists, if not create it.
self.dir_name = "out"
......@@ -27,9 +30,8 @@ class DTreeLearner(LearnerBase):
self.names_file = path_prefix + ".names"
self.tree_out = path_prefix + ".json"
# create empty files or truncate existing contents in files
# create empty data files or truncate existing data in files
open(self.data_file, 'w').close()
open(self.names_file, 'w').close()
self.exec = f'c50exact/c5.0dbg -I 1 -m 1 -f {path_prefix}'
......@@ -42,42 +44,69 @@ class DTreeLearner(LearnerBase):
return self._perc_dim
def set_grammar(self, grammar) -> None:
numerical_vars: List[str] = []
for count, (a_mat, b_vec) in enumerate(grammar):
self._s2f_func_list.append(
construct_sample_to_feature_func(a_mat, b_vec))
numerical_vars.extend(f"fvar{i}_A{count}B{count}"
for i in range(self.perc_dim))
base_features: List[str] = []
derived_feature_map = OrderedDict()
grammar_generation_utils.generateInputLanguageFile(self.names_file, numerical_vars, [])
s2f_func_list = []
for i, (a_mat, b_vec) in enumerate(grammar):
s2f_func_list.append(
construct_sample_to_feature_func(a_mat, b_vec))
ith_vars = [f"fvar{j}_A{i}" for j in range(self.perc_dim)]
base_features.extend(ith_vars)
derived_feature_map.update(
self._generate_derived_features(ith_vars))
self._s2f_func = self._compose_s2f_functions(s2f_func_list)
file_lines = ["precondition."] + \
[f"{var}: continuous." for var in base_features] + \
[f"{var} := {expr}." for var, (_, expr) in derived_feature_map.items()] + \
["precondition: true, false."]
with open(self.names_file, "w") as f:
f.write('\n'.join(file_lines))
@staticmethod
def _compose_s2f_functions(s2f_func_list):
def composed_func(sample):
return sum((list(f(sample)) for f in s2f_func_list),
start=[])
return composed_func
@staticmethod
def _generate_derived_features(
base_vars: List[str], k: int = 2) -> List[Tuple[str, Tuple[Any, str]]]:
if len(base_vars) < k:
return []
res = []
coeff_combinations = list(itertools.product([1, -1], repeat=k))
var_id_iter = range(len(base_vars))
for selected_var_ids in itertools.combinations(var_id_iter, k):
for coeff in coeff_combinations:
var_coeff_map = {base_vars[i]: c
for c, i in zip(coeff, selected_var_ids)}
expr = " + ".join(f"({c}*{base_vars[i]})"
for c, i in zip(coeff, selected_var_ids))
name = f"({expr})"
res.append((name, (var_coeff_map, expr)))
return res
def add_implication_examples(self, *args) -> None:
return super().add_implication_examples(*args)
def add_positive_examples(self, *args) -> None:
feature_vec_list = []
for sample in args:
feature_vec = []
for sample_to_feature_vec in self._s2f_func_list:
feature_vec.extend(sample_to_feature_vec(sample))
feature_vec_list.append(feature_vec)
feature_vec_list = [self._s2f_func(sample) for sample in args]
print("Positive feature vectors (dbar, psibar):", feature_vec_list)
self.write_to_file(self.data_file, feature_vec_list, "true")
print("Positive feature vectors:", feature_vec_list)
self._append_to_data_file(feature_vec_list, "true")
def add_negative_examples(self, *args) -> None:
feature_vec_list = []
for sample in args:
feature_vec = []
for sample_to_feature_vec in self._s2f_func_list:
feature_vec.extend(sample_to_feature_vec(sample))
feature_vec_list.append(feature_vec)
print("Negative feature vectors (dbar, psibar):", feature_vec_list)
self.write_to_file(self.data_file, feature_vec_list, "false")
def write_to_file(self, file: str, feature_vec_list, label: str):
with open(file, 'a') as d_file:
feature_vec_list = [self._s2f_func(sample) for sample in args]
print("Negative feature vectors:", feature_vec_list)
self._append_to_data_file(feature_vec_list, "false")
def _append_to_data_file(self, feature_vec_list, label: str):
with open(self.data_file, 'a') as d_file:
data_out = csv.writer(d_file)
for f in feature_vec_list:
print(f)
......@@ -86,50 +115,49 @@ class DTreeLearner(LearnerBase):
def learn(self) -> Tuple:
res = os.popen(self.exec).read()
print(res)
assert os.path.exists(self.tree_out) , "if learned successfully there should be a json file in "+ self.dir_name
assert os.path.exists(self.tree_out), "if learned successfully" \
f"there should be a json file in {self.dir_name}"
result = self.get_pre_from_json(self.tree_out)
print(result)
#if simplify:
# result = z3simplify.simplify(self.intVariables, self.boolVariables, result)
return ()
def get_pre_from_json(self, path):
try:
precondition = "true"
with open(path) as json_file:
tree = json.load(json_file)
precondition = self.parse_tree(tree)
return precondition
except Exception as e:
raise ValueError("cannot open Json File")
def parse_tree(self, tree):
try:
if (tree['children'] is None):
return ' ' + str(tree['classification']).strip().lower() + ' '
elif (len(tree['children']) == 2):
# for parsing bools
if 'partition' in tree:
node = tree['attribute']
else:
node = '( <= ' + tree['attribute'] + ' ' + str(tree['cut']) + ' )'
left = self.parse_tree(tree['children'][0]).strip()
right = self.parse_tree(tree['children'][1]).strip()
return '(or ( and ' + node + ' ' + left + ' ) ( and ( not ' + node + ') ' + right + ' ))'
# " (" + node + " && (" + left.strip() + ")) || ((!" + node + ") && (" + right.strip() + ")) "
return self.parse_tree(tree)
except json.JSONDecodeError:
raise ValueError(f"cannot parse {path} as a json file")
def parse_tree(self, tree) -> Optional[List[List]]:
if tree['children'] is None:
# At a leaf node, return the clause
if tree['classification']:
return [[]] # Non-none value represtns a True leaf node
else:
raise ValueError("error parsing Json(Tree)")
return None
elif len(tree['children']) == 2:
# Post-order traversal
left = self.parse_tree(tree['children'][0])
right = self.parse_tree(tree['children'][1])
if left is None and right is None:
return None
res_left = []
if left is not None:
res_left = [[(tree['attribute'], "<=", tree['cut'])] + conjunct
for conjunct in left]
res_right = []
if right is not None:
res_right = [[(tree['attribute'], ">", tree['cut'])] + conjunct
for conjunct in right]
assert res_left or res_right
return res_left + res_right
else:
raise ValueError("error parsing the json object as a binary decision tree)")
except Exception as e:
raise ValueError("error parsing Json(Tree)")
def construct_sample_to_feature_func(a_mat: np.ndarray, b_vec: np.ndarray):
perc_dim, state_dim = a_mat.shape
......@@ -142,6 +170,7 @@ def construct_sample_to_feature_func(a_mat: np.ndarray, b_vec: np.ndarray):
return perc_bar
return sample_to_feature_vec
def test_dtree_learner():
a_mat = np.array([[0., -1., 0.],
[0., 0., -1]])
......@@ -155,7 +184,7 @@ def test_dtree_learner():
(1., 2., 3., -1., -2.)
]
learner.add_positive_examples(*pos_examples)
neg_examples = [
(10., 1.0, 1.0, 0.5, 0.5),
(10., 1.0, 1.0, 1.5, 1.5),
......@@ -165,29 +194,30 @@ def test_dtree_learner():
learner.learn()
def test_sample_to_feature():
# tuple
a_mat = np.array([[0., -1., 0.],
[0., 0., -1]])
b_vec = np.zeros(2)
#construct_sample_to_feature_func: returns a function
#map: lin_trans(a_mat and b_vec pair) -> func
# construct_sample_to_feature_func: returns a function
# map: lin_trans(a_mat and b_vec pair) -> func
sample_to_feature_func = construct_sample_to_feature_func(a_mat, b_vec)
#map = {name1:sample_to_feature_func}
# map = {name1:sample_to_feature_func}
sample = np.array([1., 2., 3., -2., -3.])
# sample_to_feature_func will compute dBar and psiBar
feature_vec = sample_to_feature_func(sample)
print("sample: "+ str(feature_vec))
print("sample: " + str(feature_vec))
assert np.array_equal(feature_vec, np.array([0., 0.]))
sample = np.array([1., 2., 3., -1., -2.])
feature_vec = sample_to_feature_func(sample)
print("sample: "+ str(feature_vec))
print("sample: " + str(feature_vec))
assert np.array_equal(feature_vec, np.array([1., 1.]))
if __name__ == "__main__":
#test_sample_to_feature()
# test_sample_to_feature()
test_dtree_learner()
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment