update logging, process graph definition (bedc8945) · Commits · Blais, Chris / SOAR_integration

documenting_failures/soar_compare_bilinear.py

+210 −48

Original line number	Diff line number	Diff line
		@@ -2,6 +2,8 @@
		import os
		import sys
		import pickle
		import random
		import traceback
		import numpy as np

		soar_path = os.path.join(r"C:\Users\tjf\Documents\01_gitlab_repos\soar\python")
		@@ -28,20 +30,53 @@ import matplotlib.pyplot as plt
		from numerical_labelling.labelling import *

		# compare to soar
		from test_case_linear import constr_mem_lin
		from test_case_bilinear import constr_mem_bl, get_x0_bl_simple
		import soar.optimization.fEvaluate as feval
		import matplotlib.pyplot as plt

		def log_err(error_log, e, result, method=""):
		err_trace = traceback.format_exc()
		if result not in error_log.keys():
		error_log[result] = {}

		def int_to_bool_list(num, nSensorCandidate):
		error_log[result][method] = {
		"error":e, "method":method, "trace":err_trace,
		}

		return error_log

		def log_fail(result, failures, error_log, method, or_method, or_soar, layout):
		"""
		from Soar toolbox, creates unique id for each possible sensor layout.
		takes num (sensor layout number, from 0 to 2**nSensorCandidate)
		converts to binary, and uses that for measured/unmeasured determination.
		method: the value we've gotten, and the method used (eg obs_rref, red_lu)
		or_method: observability or redundancy for method
		or_soar: obs or red from soar
		"""
		form = format(int(nSensorCandidate), "d") + "b"
		bin_string = format(num, form)
		return [x == "1" for x in bin_string[::-1]], bin_string, form
		if result not in failures.keys():
		failures[result] = {}
		if method not in error_log[result].keys():
		failures[result][method] = {
		"method":or_method, "soar":or_soar, "layout":layout,
		}
		else:
		failures[result][method] = "error"

		return failures

		def log_success(result, successes, method, soar, lu, layout):
		"""
		method: results for method (could be tuple of obs and red)
		soar: results for soar
		lu: results for lu
		layout: bool array of measured vars
		"""
		successes[result]= {
		"method":method,
		"soar":soar,
		"lu":lu,
		"layout":layout,
		}

		return successes

		def check_incidence(iIncidence):
		"""
		@@ -50,7 +85,10 @@ def check_incidence(iIncidence):
		for row in iIncidence:
		assert sum(row) == 0, f"Incidence matrix row sums to {sum(row)}"

		def test_linear_soar(plot=False):
		def test_bilinear_soar(user_layouts=[], plot=False):
		"""
		user_layouts: supply list of integers for layouts to test
		"""
		# establish simple membrane system for testing
		iIncidence = np.array(
		[ # e P1 Mx Me1 Sp1 P2
		@@ -74,10 +112,28 @@ def test_linear_soar(plot=False):
		[2, -1], # sp1
		[1.5,-1]] # p2
		)
		pgraph = define_system(iIncidence=iIncidence, coordinates=coordinates, verbose=5)

		pgraph = define_system(
		iIncidence=iIncidence, coordinates = coordinates, verbose=5,
		arcSplitter =[[5 - 1, 6 - 1, 7 - 1]],
		arcEqualComposition = [[1-1, 2-1], [5 - 1, 6 - 1, 7 - 1, 8 - 1]],
		numberOfComponents = 3,
		)
		# pgraph = define_system(iIncidence=iIncidence, coordinates=coordinates, verbose=5)
		pgraph.add_graph_info(verbose=5)
		xMeasured_default = pgraph.xMeasured
		results = get_pareto_front(pgraph, objective="linear", verbose=5)
		# results = get_pareto_front(pgraph, objective="bilinear", verbose=5)

		# load pareto optimal layouts if they exist, run and save if they don't
		results_path = os.path.join(os.path.dirname(__file__), "bilinear_pareto_front.pkl")
		if os.path.exists(results_path):
		with open(results_path, "rb") as f:
		results = pickle.load(f)
		else:
		results = get_pareto_front(pgraph, objective="bilinear", verbose=5)
		results = [int(res) for res in results]
		with open(results_path, "wb") as f:
		pickle.dump(results, f)

		# plot
		if plot:
		@@ -86,23 +142,65 @@ def test_linear_soar(plot=False):
		fig.canvas.flush_events()
		plt.show(block=False)

		constr_lin, varbs_lin = constr_mem_lin()
		jac_lin = jacsym(constr_lin.values(), varbs_lin.values())
		constr_bl, varbs_bl = constr_mem_bl()
		jac_bl = jacsym(constr_bl.values(), varbs_bl.values())
		x0 = get_x0_bl_simple()

		failures = {}
		error_log = {}
		success_log = {}

		# results = random.sample(results, 20)
		# results = [
		# 150,
		# 3136, 33058, 16706, 9252, 8325, 4738, 33864, 8770, 17536, 33348, 16657,
		# 16426, 34836, 33041, 3104, 1668, 8218, 33416, 18498, 5140
		# ]
		if user_layouts:
		results = user_layouts

		for result in results:
		pgraph.xMeasured = xMeasured_default.copy()
		# sens, obs, red = evaluate_obs_red(int(result), pgraph)
		sens, obs, red = get_one_result(pgraph, result, objective="linear", verbose=5)

		# convert to list
		obs_soar = np.array([True if s==1 else False for s in obs[:,-1]])
		red_soar = np.array([True if s==1 else False for s in red[:,-1]])
		sens, obs, red = get_one_result(pgraph, result, objective="bilinear", verbose=5)

		# for concentrations we have it backwards.
		rev_sens = np.concatenate([sens[8:],sens[0:8]])
		# convert to list. in soar -1 means undetermined, right?
		obs_soar_mass = np.array([True if s==1 else False for s in obs[:,-1]])
		red_soar_mass = np.array([True if s==1 else False for s in red[:,-1]])

		# add the concentration sensors
		obs_soar_conc = np.array([True if s==1 else False for s in obs[:,0]])
		red_soar_conc = np.array([True if s==1 else False for s in red[:,0]])

		obs_soar = np.concatenate([obs_soar_mass, obs_soar_conc])
		red_soar = np.concatenate([red_soar_mass, red_soar_conc])
		layout = list(np.array(rev_sens).astype(bool))

		error_log[result] = {}
		try:
		obs_rref, red_rref = observability_redundancy_labelling_rref(jac_bl, layout, x0=x0)
		except Exception as e:
		err_log = log_err(error_log, e, result, method="rref")
		obs_rref = np.full(len(obs_soar), fill_value=np.nan)
		red_rref = np.full(len(red_soar), fill_value=np.nan)

		try:
		obs_lu, red_lu = observability_redundancy_labelling_lu(jac_bl, layout, x0=x0)
		except Exception as e:
		err_log = log_err(error_log, e, result, method="lu")
		# add in a nan array so we can run the other sections
		obs_lu = np.full(len(obs_soar), fill_value=np.nan)
		red_lu = np.full(len(red_soar), fill_value=np.nan)

		try:
		obs_qr, red_qr = observability_redundancy_labelling_qr(jac_bl, layout, x0=x0)
		except Exception as e:
		err_log = log_err(error_log, e, result, method="qr")
		obs_qr = np.full(len(obs_soar), fill_value=np.nan)
		red_qr = np.full(len(red_soar), fill_value=np.nan)

		layout = list(np.array(sens).astype(bool))
		obs_rref, red_rref = observability_redundancy_labelling_rref(jac_lin, layout)
		obs_lu, red_lu = observability_redundancy_labelling_lu(jac_lin, layout)
		obs_qr, red_qr = observability_redundancy_labelling_qr(jac_lin, layout)

		obs_rref_equ = np.array_equal(obs_rref, obs_soar)
		obs_lu_equ = np.array_equal(obs_lu, obs_soar)
		@@ -112,44 +210,108 @@ def test_linear_soar(plot=False):
		red_lu_equ = np.array_equal(red_lu, red_soar)
		red_qr_equ = np.array_equal(red_qr, red_soar)


		failures[result] = {}
		if not obs_rref_equ:
		failures[result] = {
		"obs_method":obs_rref, "obs_soar":obs_soar, "type":"obs", "method":"rref",
		}
		failures = log_fail(result, failures, error_log, "obs_rref", obs_rref, obs_soar, layout)
		if not red_rref_equ:
		failures = log_fail(result, failures, error_log, "red_rref", red_rref, red_soar, layout)

		if not obs_lu_equ:
		failures[result] = {
		"obs_method":obs_lu, "obs_soar":obs_soar, "type":"obs", "method":"lu",
		}
		failures = log_fail(result, failures, error_log, "obs_lu", obs_lu, obs_soar, layout)
		if not red_lu_equ:
		failures = log_fail(result, failures, error_log, "red_lu", red_lu, red_soar, layout)

		if not obs_qr_equ:
		failures[result] = {
		"obs_method":obs_qr, "obs_soar":obs_soar, "type":"obs", "method":"qr",
		}
		failures = log_fail(result, failures, error_log, "obs_qr", obs_qr, obs_soar, layout)
		if not red_qr_equ:
		failures = log_fail(result, failures, error_log, "red_qr", red_qr, red_soar, layout)

		if (obs_rref_equ and red_rref_equ) and not (obs_lu_equ or red_lu_equ):
		log_success(
		result=result,
		successes=success_log,
		method=(obs_rref, red_rref),
		soar=(obs_soar, red_soar),
		lu=(obs_lu, red_lu),
		layout = layout,
		)

		# try:
		# assert np.array_equal(red_rref,red_soar), "redundancies don't match"
		# except AssertionError as e:
		# failures[result] = {
		# "red_method":red_rref, "red_soar":red_soar, "type":"red", "method":"rref"
		# }
		if len(error_log) > 0:
		err_log = os.path.join(r".\documenting_failures\errors\bilinear_errors.pkl")
		if not os.path.exists(os.path.dirname(err_log)):
		os.mkdir(os.path.dirname(err_log))

		with open(err_log, "wb") as f:
		pickle.dump(obj=error_log, file=f)
		print(f"errors in {len(error_log)} layouts out of {len(results)}")
		else:
		print("no errors")

		if len(failures) > 0:
		fail_log = os.path.join(r".\documenting_failures\failures\linear_failures.pkl")
		fail_log = os.path.join(r".\documenting_failures\failures\bilinear_failures.pkl")
		if not os.path.exists(os.path.dirname(fail_log)):
		os.mkdir(os.path.dirname(fail_log))

		with open(fail_log, "wb") as f:
		pickle.dump(obj=fail_log, file=f)
		print(f"errors in {len(fail_log)} layouts")
		pickle.dump(obj=failures, file=f)
		print(f"failures in {len(fail_log)} layouts out of {len(results)}")

		else:
		print("all layouts pass for linear system")

		if len(success_log) > 0:
		success_path = os.path.join(r".\documenting_failures\successes\bilinear_successes.pkl")
		if not os.path.exists(os.path.dirname(success_path)):
		os.mkdir(os.path.dirname(success_path))

		with open(success_path, "wb") as f:
		pickle.dump(obj=success_log, file=f)
		print(f"successes in {len(success_log)} layouts out of {len(results)}")

		def run_single_soar_layout(layout = []):

		# establish simple membrane system for testing
		iIncidence = np.array(
		[ # e P1 Mx Me1 Sp1 P2
		[-1, +1, 0, 0, 0, 0], # 1
		[ 0, -1, +1, 0, 0, 0], # 2
		[ 0, 0, -1, +1, 0, 0], # 3
		[+1, 0, 0, -1, 0, 0], # 4
		[ 0, 0, 0, -1, +1, 0], # 5
		[+1, 0, 0, 0, -1, 0], # 6
		[ 0, 0, 0, 0, -1, +1], # 7
		[ 0, 0, +1, 0, 0, -1], # 8
		]
		)
		check_incidence(iIncidence)
		# define coordinates for plotting process graph
		coordinates = np.array([
		[2, 1], #env
		[0, 0], #p1
		[0.75, 0], #mx
		[1.5, 0], #me1
		[2, -1], # sp1
		[1.5,-1]] # p2
		)
		# translate layout to soar

		pgraph = define_system(
		iIncidence=iIncidence, coordinates = coordinates, verbose=5,
		arcSplitter =[[5 - 1, 6 - 1, 7 - 1]],
		arcEqualComposition = [[1-1, 2-1], [7-1, 8-1], [5 - 1, 6 - 1, 7 - 1]],
		numberOfComponents = 2,
		)

		pgraph.add_graph_info(verbose=5)
		xMeasured_default = pgraph.xMeasured
		sens, obs, red = get_one_result(pgraph, result, objective="bilinear", verbose=5)

		return sens, obs, red


		if __name__ == "__main__":
		test_linear_soar()
		# test_bilinear_soar(user_layouts=[34952])
		test_bilinear_soar()

documenting_failures/soar_compare_linear.py

+47 −55

Original line number	Diff line number	Diff line
		@@ -3,6 +3,7 @@ import os
		import sys
		import pickle
		import numpy as np
		import traceback

		soar_path = os.path.join(r"C:\Users\tjf\Documents\01_gitlab_repos\soar\python")
		if soar_path not in sys.path:
		@@ -32,16 +33,32 @@ from test_case_linear import constr_mem_lin
		import soar.optimization.fEvaluate as feval
		import matplotlib.pyplot as plt

		def log_err(error_log, e, result, method=""):
		err_trace = traceback.format_exc()
		if result not in error_log.keys():
		error_log[result] = {}
		error_log[result]['rref'] = {
		"error":e, "type":"obs", "method":method, "trace":err_trace,
		}

		return error_log

		def int_to_bool_list(num, nSensorCandidate):
		def log_fail(result, failures, error_log, method, or_method, or_soar, layout):
		"""
		from Soar toolbox, creates unique id for each possible sensor layout.
		takes num (sensor layout number, from 0 to 2**nSensorCandidate)
		converts to binary, and uses that for measured/unmeasured determination.
		method: the value we've gotten, and the method used (eg obs_rref, red_lu)
		or_method: observability or redundancy for method
		or_soar: obs or red from soar
		"""
		form = format(int(nSensorCandidate), "d") + "b"
		bin_string = format(num, form)
		return [x == "1" for x in bin_string[::-1]], bin_string, form
		if result not in failures.keys():
		failures[result] = {}
		if method not in error_log[result].keys():
		failures[result][method] = {
		"method":or_method, "soar":or_soar, "layout":layout,
		}
		else:
		failures[result]["rref"] = "error"

		return failures

		def check_incidence(iIncidence):
		"""
		@@ -74,7 +91,12 @@ def test_linear_soar(plot=False):
		[2, -1], # sp1
		[1.5,-1]] # p2
		)
		pgraph = define_system(iIncidence=iIncidence, coordinates=coordinates, verbose=5)
		pgraph = define_system(
		iIncidence=iIncidence, coordinates = coordinates, verbose=5,
		arcSplitter =[[5 - 1, 6 - 1, 7 - 1]],
		arcEqualComposition = [[1-1, 2-1], [5 - 1, 6 - 1, 7 - 1, 8 - 1]],
		numberOfComponents = 3,
		)
		pgraph.add_graph_info(verbose=5)
		xMeasured_default = pgraph.xMeasured

		@@ -96,6 +118,7 @@ def test_linear_soar(plot=False):
		constr_lin, varbs_lin = constr_mem_lin()
		jac_lin = jacsym(constr_lin.values(), varbs_lin.values())

		successes = {}
		failures = {}
		error_log = {}
		for result in results:
		@@ -113,35 +136,22 @@ def test_linear_soar(plot=False):
		try:
		obs_rref, red_rref = observability_redundancy_labelling_rref(jac_lin, layout)
		except Exception as e:
		if result not in error_log.keys():
		error_log[result] = {}
		error_log[result]['rref'] = {
		"error":e, "type":"obs", "method":"rref",
		}
		err_log = log_err(error_log, e, result, method="rref")
		obs_rref = np.full(len(obs_soar), fill_value=np.nan)
		red_rref = np.full(len(red_soar), fill_value=np.nan)

		try:
		obs_lu, red_lu = observability_redundancy_labelling_lu(jac_lin, layout)
		except Exception as e:
		if result not in error_log.keys():
		error_log[result] = {}
		error_log[result]['lu'] = {
		"error":e, "type":"obs", "method":"lu",
		}
		err_log = log_err(error_log, e, result, method="lu")
		# add in a nan array so we can run the other sections
		obs_lu = np.full(len(obs_soar), fill_value=np.nan)
		red_lu = np.full(len(red_soar), fill_value=np.nan)


		try:
		obs_qr, red_qr = observability_redundancy_labelling_qr(jac_lin, layout)
		except Exception as e:
		if result not in error_log.keys():
		error_log[result] = {}
		error_log[result]['qr'] = {
		"error":e, "type":"obs", "method":"qr",
		}
		err_log = log_err(error_log, e, result, method="qr")
		obs_qr = np.full(len(obs_soar), fill_value=np.nan)
		red_qr = np.full(len(red_soar), fill_value=np.nan)

		@@ -156,40 +166,22 @@ def test_linear_soar(plot=False):

		failures[result] = {}
		if not obs_rref_equ:
		if result not in failures.keys():
		failures[result] = {}
		if "rref" not in error_log[result].keys():
		failures[result]['rref'] = {
		"obs_method":obs_rref, "obs_soar":obs_soar, "type":"obs", "method":"rref",
		}
		else:
		failures[result]["rref"] = "error"
		failures = log_fail(result, failures, error_log, "obs_rref", obs_rref, obs_soar, layout)
		if not red_rref_equ:
		failures = log_fail(result, failures, error_log, "red_rref", red_rref, red_soar, layout)

		if not obs_lu_equ:
		if result not in failures.keys():
		failures[result] = {}
		if "lu" not in error_log[result].keys():
		failures[result]['lu'] = {
		"obs_method":obs_lu, "obs_soar":obs_soar, "type":"obs", "method":"lu",
		}
		else:
		failures[result]['lu'] = "error"
		failures = log_fail(result, failures, error_log, "obs_lu", obs_lu, obs_soar, layout)
		if not red_lu_equ:
		failures = log_fail(result, failures, error_log, "red_lu", red_lu, red_soar, layout)

		if not obs_qr_equ:
		if result not in failures.keys():
		failures[result] = {}
		if "qr" not in error_log[result].keys():
		failures[result]['qr'] = {
		"obs_method":obs_qr, "obs_soar":obs_soar, "type":"obs", "method":"qr",
		}
		else:
		failures[result]['qr'] = "error"
		# try:
		# assert np.array_equal(red_rref,red_soar), "redundancies don't match"
		# except AssertionError as e:
		# failures[result] = {
		# "red_method":red_rref, "red_soar":red_soar, "type":"red", "method":"rref"
		# }
		failures = log_fail(result, failures, error_log, "obs_qr", obs_qr, obs_soar, layout)
		if not red_qr_equ:
		failures = log_fail(result, failures, error_log, "red_qr", red_qr, red_soar, layout)

		# success = (obs_rref_equ and red_rref_equ) and not (obs_lu_equ or red_lu_eq)
		# if success:


		if len(error_log) > 0:

documenting_failures/soar_compare_nonlinear.py

+73 −32

File changed.

Preview size limit exceeded, changes collapsed.

Admin message