(self)
| 18 | |
| 19 | class ExampleRunner: |
| 20 | def __init__(self): |
| 21 | # initial operations |
| 22 | now = datetime.now().strftime("%Y-%m-%d %H:%M:%S") |
| 23 | print(f"\n{YELLOW}{BOLD}Running all examples on {now}...{RESET}\n") |
| 24 | |
| 25 | # initial member variables |
| 26 | self.args = self.parse_args() # parse command line arguments |
| 27 | self.test_start_time = time.time() |
| 28 | self.max_threads = self.args.parallel_num # maximum number of threads to use |
| 29 | self.item_results = {} # store test results for each item |
| 30 | self.lock_dict = {} # store lock for some limisted items |
| 31 | |
| 32 | if self.args.save is not None: |
| 33 | self.check_and_create_directory(self.args.save) # todo ... |
| 34 | |
| 35 | # Read version from VERSION file |
| 36 | version_file = os.path.join(default_build_path, "..", "VERSION") |
| 37 | with open(version_file, "r") as f: |
| 38 | version = f.read().strip() |
| 39 | |
| 40 | subprocess.run(["pip3", |
| 41 | "install", |
| 42 | f"./aimrt_py_pkg/dist/aimrt_py-{version}-cp310-cp310-linux_x86_64.whl", |
| 43 | "--force-reinstall"], |
| 44 | cwd=default_build_path, |
| 45 | ) |
| 46 | subprocess.run( |
| 47 | ["bash", os.path.join("build_examples_py_pb_rpc.sh")], |
| 48 | cwd=os.path.join(py_cwd, "pb_rpc"), |
| 49 | ) |
| 50 | subprocess.run( |
| 51 | ["bash", os.path.join("build_examples_py_pb_chn.sh")], |
| 52 | cwd=os.path.join(py_cwd, "pb_chn"), |
| 53 | ) |
| 54 | subprocess.run( |
| 55 | ["bash", os.path.join("build_examples_py_ros2_rpc.sh")], |
| 56 | cwd=os.path.join(py_cwd, "ros2_rpc"), |
| 57 | ) |
| 58 | |
| 59 | def check_and_create_directory(self, test_log_save_path: str) -> None: |
| 60 | if test_log_save_path and not os.path.exists(test_log_save_path): |
nothing calls this directly
no test coverage detected