2024-03-12 02:23:44 -03:00
|
|
|
#!/usr/bin/python3
|
|
|
|
|
|
|
|
'''
|
|
|
|
Test parameter upgrade, master vs branch
|
|
|
|
|
2024-03-15 01:54:39 -03:00
|
|
|
./Tools/autotest/test_param_upgrade.py --vehicle=arduplane --param "GPS_TYPE=17->GPS1_TYPE=17" --param "GPS_TYPE2=37->GPS2_TYPE=37" --param "GPS_GNSS_MODE=21->GPS1_GNSS_MODE=21" --param "GPS_GNSS_MODE2=45->GPS2_GNSS_MODE=45" --param "GPS_RATE_MS=186->GPS1_RATE_MS=186" --param "GPS_RATE_MS2=123->GPS2_RATE_MS=123" --param "GPS_POS1_X=3.75->GPS1_POS_X=3.75" --param "GPS_POS2_X=6.9->GPS2_POS_X=6.9" --param "GPS_POS1_Y=2.75->GPS1_POS_Y=2.75" --param "GPS_POS2_Y=5.9->GPS2_POS_Y=5.9" --param "GPS_POS1_Z=12.1->GPS1_POS_Z=12.1" --param "GPS_POS2_Z=-6.9->GPS2_POS_Z=-6.9" --param "GPS_DELAY_MS=987->GPS1_DELAY_MS=987" --param "GPS_DELAY_MS2=2345->GPS2_DELAY_MS=2345" --param "GPS_COM_PORT=19->GPS1_COM_PORT=19" --param "GPS_COM_PORT2=100->GPS2_COM_PORT=100" --param "GPS_CAN_NODEID1=109->GPS1_CAN_NODEID=109" --param "GPS_CAN_NODEID2=102->GPS2_CAN_NODEID=102" --param "GPS1_CAN_OVRIDE=34->GPS1_CAN_OVRIDE=34" --param "GPS2_CAN_OVRIDE=67" --param "GPS_MB1_TYPE=1->GPS1_MB_TYPE=1" --param "GPS_MB1_OFS_X=3.14->GPS1_MB_OFS_X=3.14" --param "GPS_MB1_OFS_Y=2.18->GPS1_MB_OFS_Y=2.18" --param "GPS_MB1_OFS_Z=17.6->GPS1_MB_OFS_Z=17.6" --param "GPS_MB2_TYPE=13->GPS2_MB_TYPE=13" --param "GPS_MB2_OFS_X=17.14->GPS2_MB_OFS_X=17.14" --param "GPS_MB2_OFS_Y=12.18->GPS2_MB_OFS_Y=12.18" --param "GPS_MB2_OFS_Z=27.6->GPS2_MB_OFS_Z=27.6" # noqa
|
2024-03-12 02:23:44 -03:00
|
|
|
|
|
|
|
AP_FLAKE8_CLEAN
|
|
|
|
'''
|
|
|
|
|
|
|
|
import vehicle_test_suite
|
|
|
|
import os
|
|
|
|
import sys
|
|
|
|
import argparse
|
|
|
|
import subprocess
|
|
|
|
import time
|
|
|
|
import string
|
|
|
|
import pathlib
|
|
|
|
|
|
|
|
from pysim import util
|
|
|
|
|
|
|
|
|
|
|
|
class ParamChange():
|
|
|
|
def __init__(self, old_name, old_value, new_name, new_value):
|
|
|
|
self.old_name = old_name
|
|
|
|
self.old_value = old_value
|
|
|
|
self.new_name = new_name
|
|
|
|
self.new_value = new_value
|
|
|
|
|
|
|
|
def __str__(self):
|
|
|
|
return f"{self.old_name}={self.old_value}->{self.new_name}={self.new_value}"
|
|
|
|
|
|
|
|
|
|
|
|
class TestParamUpgradeTestSuiteSetParameters(vehicle_test_suite.TestSuite):
|
|
|
|
def __init__(self, binary, param_changes):
|
|
|
|
super(TestParamUpgradeTestSuiteSetParameters, self).__init__(binary)
|
|
|
|
self.param_changes = param_changes
|
|
|
|
|
|
|
|
def sysid_thismav(self):
|
|
|
|
if "antennatracker" in self.binary:
|
|
|
|
return 2
|
|
|
|
return super(TestParamUpgradeTestSuiteSetParameters, self).sysid_thismav()
|
|
|
|
|
|
|
|
def vehicleinfo_key(self):
|
|
|
|
'''magically guess vehicleinfo_key from filepath'''
|
|
|
|
path = self.binary.lower()
|
|
|
|
if "plane" in path:
|
|
|
|
return "ArduPlane"
|
|
|
|
if "copter" in path:
|
|
|
|
return "ArduCopter"
|
|
|
|
raise ValueError("Can't determine vehicleinfo_key from binary path")
|
|
|
|
|
|
|
|
def run(self):
|
|
|
|
self.start_SITL(
|
|
|
|
binary=self.binary,
|
|
|
|
model="novehicle",
|
|
|
|
wipe=True,
|
|
|
|
sitl_home="1,1,1,1",
|
|
|
|
)
|
|
|
|
self.get_mavlink_connection_going()
|
|
|
|
|
|
|
|
sets = {}
|
|
|
|
for param_change in param_changes:
|
|
|
|
sets[param_change.old_name] = param_change.old_value
|
|
|
|
self.set_parameters(sets)
|
|
|
|
|
|
|
|
# stopping SITL too soon can leave eeprom.bin corrupt!
|
|
|
|
self.delay_sim_time(2) # FIXTHAT, should not be needed!
|
|
|
|
|
|
|
|
self.stop_SITL()
|
|
|
|
|
|
|
|
|
|
|
|
class TestParamUpgradeTestSuiteCheckParameters(vehicle_test_suite.TestSuite):
|
|
|
|
def __init__(self, binary, param_changes, epsilon=0.0001):
|
|
|
|
super(TestParamUpgradeTestSuiteCheckParameters, self).__init__(binary)
|
|
|
|
self.param_changes = param_changes
|
|
|
|
self.epsilon = epsilon
|
|
|
|
|
|
|
|
def sysid_thismav(self):
|
|
|
|
if "antennatracker" in self.binary:
|
|
|
|
return 2
|
|
|
|
return super(TestParamUpgradeTestSuiteCheckParameters, self).sysid_thismav()
|
|
|
|
|
|
|
|
def vehicleinfo_key(self):
|
|
|
|
'''magically guess vehicleinfo_key from filepath'''
|
|
|
|
path = self.binary.lower()
|
|
|
|
if "plane" in path:
|
|
|
|
return "ArduPlane"
|
|
|
|
if "copter" in path:
|
|
|
|
return "ArduCopter"
|
|
|
|
raise ValueError("Can't determine vehicleinfo_key from binary path")
|
|
|
|
|
|
|
|
def run(self):
|
|
|
|
self.start_SITL(
|
|
|
|
binary=self.binary,
|
|
|
|
model="novehicle",
|
|
|
|
sitl_home="1,1,1,1",
|
|
|
|
wipe=False,
|
|
|
|
)
|
|
|
|
self.get_mavlink_connection_going()
|
|
|
|
|
|
|
|
params_to_check = [x.new_name for x in self.param_changes]
|
|
|
|
fetched_params = self.get_parameters(params_to_check)
|
|
|
|
|
|
|
|
for p in self.param_changes:
|
|
|
|
if abs(fetched_params[p.new_name] - p.new_value) > self.epsilon:
|
|
|
|
raise ValueError(f"{p.old_name}={p.old_value} did not convert into {p.new_name}={p.new_value}, got {p.new_name}={fetched_params[p.new_name]}") # noqa
|
|
|
|
|
|
|
|
print(f"OK: {p.old_name}={p.old_value} converted to {p.new_name}={p.new_value}")
|
|
|
|
|
|
|
|
self.stop_SITL()
|
|
|
|
|
|
|
|
|
|
|
|
class TestParamUpgradeForVehicle():
|
|
|
|
def __init__(self,
|
|
|
|
vehicle,
|
|
|
|
param_changes,
|
|
|
|
branch=None,
|
|
|
|
master_branch="master",
|
|
|
|
run_eedump_before=False,
|
|
|
|
run_eedump_after=False,
|
|
|
|
):
|
|
|
|
self.vehicle = vehicle
|
|
|
|
self.master_branch = master_branch
|
|
|
|
self.branch = branch
|
|
|
|
self.param_changes = param_changes
|
|
|
|
self.run_eedump_before = run_eedump_before
|
|
|
|
self.run_eedump_after = run_eedump_after
|
|
|
|
|
|
|
|
def run_program(self, prefix, cmd_list, show_output=True, env=None, show_output_on_error=True, show_command=None, cwd="."):
|
|
|
|
if show_command is None:
|
|
|
|
show_command = True
|
|
|
|
if show_command:
|
|
|
|
cmd = " ".join(cmd_list)
|
|
|
|
if cwd is None:
|
|
|
|
cwd = "."
|
|
|
|
self.progress(f"Running ({cmd}) in ({cwd})")
|
|
|
|
p = subprocess.Popen(
|
|
|
|
cmd_list,
|
|
|
|
stdin=None,
|
|
|
|
stdout=subprocess.PIPE,
|
|
|
|
close_fds=True,
|
|
|
|
stderr=subprocess.STDOUT,
|
|
|
|
cwd=cwd,
|
|
|
|
env=env)
|
|
|
|
output = ""
|
|
|
|
while True:
|
|
|
|
x = p.stdout.readline()
|
|
|
|
if len(x) == 0:
|
|
|
|
returncode = os.waitpid(p.pid, 0)
|
|
|
|
if returncode:
|
|
|
|
break
|
|
|
|
# select not available on Windows... probably...
|
|
|
|
time.sleep(0.1)
|
|
|
|
continue
|
|
|
|
x = bytearray(x)
|
|
|
|
x = filter(lambda x : chr(x) in string.printable, x)
|
|
|
|
x = "".join([chr(c) for c in x])
|
|
|
|
output += x
|
|
|
|
x = x.rstrip()
|
|
|
|
some_output = "%s: %s" % (prefix, x)
|
|
|
|
if show_output:
|
|
|
|
print(some_output)
|
|
|
|
else:
|
|
|
|
output += some_output
|
|
|
|
(_, status) = returncode
|
|
|
|
if status != 0:
|
|
|
|
if not show_output and show_output_on_error:
|
|
|
|
# we were told not to show output, but we just
|
|
|
|
# failed... so show output...
|
|
|
|
print(output)
|
|
|
|
self.progress("Process failed (%s)" %
|
|
|
|
str(returncode))
|
|
|
|
try:
|
|
|
|
path = pathlib.Path(self.tmpdir, f"process-failure-{int(time.time())}")
|
|
|
|
path.write_text(output)
|
|
|
|
self.progress("Wrote process failure file (%s)" % path)
|
|
|
|
except Exception:
|
|
|
|
self.progress("Writing process failure file failed")
|
|
|
|
raise subprocess.CalledProcessError(
|
|
|
|
returncode, cmd_list)
|
|
|
|
return output
|
|
|
|
|
|
|
|
def find_current_git_branch_or_sha1(self):
|
|
|
|
try:
|
|
|
|
output = self.run_git(["symbolic-ref", "--short", "HEAD"])
|
|
|
|
output = output.strip()
|
|
|
|
return output
|
|
|
|
except subprocess.CalledProcessError:
|
|
|
|
pass
|
|
|
|
|
|
|
|
# probably in a detached-head state. Get a sha1 instead:
|
|
|
|
output = self.run_git(["rev-parse", "--short", "HEAD"])
|
|
|
|
output = output.strip()
|
|
|
|
return output
|
|
|
|
|
|
|
|
def find_git_branch_merge_base(self, branch, master_branch):
|
|
|
|
output = self.run_git(["merge-base", branch, master_branch])
|
|
|
|
output = output.strip()
|
|
|
|
return output
|
|
|
|
|
|
|
|
def run_git(self, args, show_output=True, source_dir=None):
|
|
|
|
'''run git with args git_args; returns git's output'''
|
|
|
|
cmd_list = ["git"]
|
|
|
|
cmd_list.extend(args)
|
|
|
|
return self.run_program("TPU-GIT", cmd_list, show_output=show_output, cwd=source_dir)
|
|
|
|
|
|
|
|
def progress(self, string):
|
|
|
|
'''pretty-print progress'''
|
|
|
|
print("TPU: %s" % string)
|
|
|
|
|
|
|
|
def binary_path(self, vehicle):
|
|
|
|
build_subdir = "sitl"
|
|
|
|
if 'AP_Periph' in vehicle:
|
|
|
|
build_subdir = "sitl_periph_universal"
|
|
|
|
return f"build/{build_subdir}/{self.build_target_name(vehicle)}"
|
|
|
|
|
|
|
|
def build_target_name(self, vehicle):
|
|
|
|
binary_name = vehicle
|
|
|
|
if binary_name == 'heli':
|
|
|
|
binary_name = 'arducopter-heli'
|
|
|
|
if binary_name == 'rover':
|
|
|
|
binary_name = 'ardurover'
|
|
|
|
return f"bin/{binary_name}"
|
|
|
|
|
|
|
|
def run_eedump(self):
|
|
|
|
self.run_program("TPU-EED", [
|
|
|
|
"libraries/AP_Param/tools/eedump_apparam",
|
|
|
|
"eeprom.bin"
|
|
|
|
])
|
|
|
|
|
|
|
|
def run(self):
|
|
|
|
branch = self.branch
|
|
|
|
if branch is None:
|
|
|
|
branch = self.find_current_git_branch_or_sha1()
|
|
|
|
|
|
|
|
master_commit = self.master_branch
|
|
|
|
|
|
|
|
self.use_merge_base = True
|
|
|
|
if self.use_merge_base:
|
|
|
|
master_commit = self.find_git_branch_merge_base(branch, self.master_branch)
|
|
|
|
self.progress("Using merge base (%s)" % master_commit)
|
|
|
|
|
|
|
|
self.run_git(["checkout", master_commit], show_output=False)
|
|
|
|
self.run_git(["submodule", "update", "--recursive"], show_output=False)
|
|
|
|
board = "sitl"
|
|
|
|
if "AP_Periph" in self.vehicle:
|
|
|
|
board = "sitl_periph_universal"
|
|
|
|
util.build_SITL(
|
|
|
|
self.build_target_name(self.vehicle),
|
|
|
|
board=board,
|
|
|
|
clean=False,
|
|
|
|
configure=True,
|
|
|
|
)
|
|
|
|
suite = TestParamUpgradeTestSuiteSetParameters(self.binary_path(self.vehicle), self.param_changes)
|
|
|
|
suite.run()
|
|
|
|
|
|
|
|
self.run_git(["checkout", branch], show_output=False)
|
|
|
|
self.run_git(["submodule", "update", "--recursive"], show_output=False)
|
|
|
|
util.build_SITL(
|
|
|
|
self.build_target_name(self.vehicle),
|
|
|
|
board=board,
|
|
|
|
clean=False,
|
|
|
|
configure=True,
|
|
|
|
)
|
|
|
|
suite = TestParamUpgradeTestSuiteCheckParameters(self.binary_path(self.vehicle), self.param_changes)
|
|
|
|
|
|
|
|
if self.run_eedump_before:
|
|
|
|
self.run_eedump()
|
|
|
|
|
|
|
|
# this call starts SITL which will do the upgrade:
|
|
|
|
suite.run()
|
|
|
|
|
|
|
|
if self.run_eedump_after:
|
|
|
|
self.run_eedump()
|
|
|
|
|
|
|
|
|
|
|
|
class TestParamUpgrade():
|
|
|
|
def __init__(self,
|
|
|
|
param_changes,
|
|
|
|
vehicles=None,
|
|
|
|
run_eedump_before=False,
|
|
|
|
run_eedump_after=False,
|
|
|
|
):
|
|
|
|
self.vehicles = vehicles
|
|
|
|
self.param_changes = param_changes
|
|
|
|
self.vehicles = vehicles
|
|
|
|
self.run_eedump_before = run_eedump_before
|
|
|
|
self.run_eedump_after = run_eedump_after
|
|
|
|
|
|
|
|
if self.vehicles is None:
|
|
|
|
self.vehicles = self.all_vehicles()
|
|
|
|
|
|
|
|
def all_vehicles(self):
|
|
|
|
return [
|
|
|
|
# 'AP_Periph',
|
|
|
|
'arducopter',
|
|
|
|
'arduplane',
|
|
|
|
'antennatracker',
|
|
|
|
'heli',
|
|
|
|
'rover',
|
|
|
|
'blimp',
|
|
|
|
'ardusub',
|
|
|
|
]
|
|
|
|
|
|
|
|
def run(self):
|
|
|
|
for vehicle in self.vehicles:
|
|
|
|
s = TestParamUpgradeForVehicle(
|
|
|
|
vehicle,
|
|
|
|
self.param_changes,
|
|
|
|
run_eedump_before=self.run_eedump_before,
|
|
|
|
run_eedump_after=self.run_eedump_after,
|
|
|
|
)
|
|
|
|
s.run()
|
|
|
|
|
|
|
|
|
|
|
|
if __name__ == "__main__":
|
|
|
|
''' main program '''
|
|
|
|
os.environ['PYTHONUNBUFFERED'] = '1'
|
|
|
|
|
|
|
|
if sys.platform != "darwin":
|
|
|
|
os.putenv('TMPDIR', util.reltopdir('tmp'))
|
|
|
|
|
|
|
|
parser = argparse.ArgumentParser("test_param_upgrade.py")
|
|
|
|
parser.add_argument(
|
|
|
|
"--param",
|
|
|
|
action='append',
|
|
|
|
default=[],
|
|
|
|
help="PARAM=VALUE pair to test upgrade for, or PARAM=VALUE->NEWPARAM=NEWVALUE",
|
|
|
|
)
|
|
|
|
parser.add_argument(
|
|
|
|
"--vehicle",
|
|
|
|
action='append',
|
|
|
|
default=[],
|
|
|
|
help="vehicle to test",
|
|
|
|
)
|
|
|
|
parser.add_argument(
|
|
|
|
"--run-eedump-before",
|
|
|
|
action='store_true',
|
|
|
|
default=False,
|
|
|
|
help="run the (already-compiled) eedump tool on eeprom.bin before doing conversion",
|
|
|
|
)
|
|
|
|
parser.add_argument(
|
|
|
|
"--run-eedump-after",
|
|
|
|
action='store_true',
|
|
|
|
default=False,
|
|
|
|
help="run the (already-compiled) eedump tool on eeprom.bin after doing conversion",
|
|
|
|
)
|
|
|
|
|
|
|
|
args = parser.parse_args()
|
|
|
|
|
|
|
|
param_changes = []
|
|
|
|
for x in args.param:
|
|
|
|
if "->" in x:
|
|
|
|
(old, new) = x.split("->")
|
|
|
|
(old_name, old_value) = old.split("=")
|
|
|
|
(new_name, new_value) = new.split("=")
|
|
|
|
param_changes.append(ParamChange(old_name, float(old_value), new_name, float(new_value)))
|
|
|
|
|
|
|
|
else:
|
|
|
|
(name, value) = x.split("=")
|
|
|
|
param_changes.append(ParamChange(name, float(value), name, float(value)))
|
|
|
|
|
|
|
|
vehicles = args.vehicle
|
|
|
|
|
|
|
|
if 'AP_Periph' in vehicles:
|
|
|
|
raise ValueError("AP_Periph not supported yet")
|
|
|
|
|
|
|
|
if len(vehicles) == 0:
|
|
|
|
vehicles = None
|
|
|
|
|
|
|
|
x = TestParamUpgrade(
|
|
|
|
param_changes,
|
|
|
|
vehicles=vehicles,
|
|
|
|
run_eedump_before=args.run_eedump_before,
|
|
|
|
run_eedump_after=args.run_eedump_after,
|
|
|
|
)
|
|
|
|
x.run()
|