From 61cecc52dd7a50ccb2c5f767d82ada4142181b86 Mon Sep 17 00:00:00 2001 From: Dmytro Bogovych Date: Sun, 20 Aug 2023 14:07:03 +0300 Subject: [PATCH] - fixes --- src/agent_gsm.py | 2 +- src/bt_phone.py | 17 ++++++++++++----- src/utils_cache.py | 20 +++++++++++++++----- src/utils_qualtest.py | 14 ++++++++------ src/utils_sevana.py | 2 +- 5 files changed, 37 insertions(+), 18 deletions(-) diff --git a/src/agent_gsm.py b/src/agent_gsm.py index c6fa62b..8e7fe2b 100644 --- a/src/agent_gsm.py +++ b/src/agent_gsm.py @@ -470,7 +470,7 @@ if config['log']['adb']: if 'cache_dir' in config: DIR_CACHE = Path(config['cache_dir']) if not DIR_CACHE.is_absolute(): - DIR_CACHE = DIR_CACHE / config['cache_dir'] + DIR_CACHE = DIR_THIS / config['cache_dir'] CACHE = utils_cache.InfoCache(dir=DIR_CACHE) diff --git a/src/bt_phone.py b/src/bt_phone.py index e56a2f9..0afe7ce 100644 --- a/src/bt_phone.py +++ b/src/bt_phone.py @@ -67,14 +67,17 @@ class Phone(Observable): # Wait for online modem and return this - def wait_for_online_modem(self): - while True: + def wait_for_online_modem(self, timeout_seconds: int = 1000000): + timestamp = time.time() + while True and timestamp + timeout_seconds > time.time(): modem = self.get_online_modem() if modem != None: return modem # Sleep another 10 seconds and check again - time.sleep(10.0) + time.sleep(1.0) + + return None def get_incoming_call(self): @@ -108,8 +111,12 @@ class Phone(Observable): # Wait for online modem utils.log('Waiting for BT modem (phone must be paired and connected before)...') - self.modem = self.wait_for_online_modem() - + self.modem = self.wait_for_online_modem(timeout_seconds=10) # 10 seconds timeout + + if self.modem is None: + utils.log_error(f'No BT modem found. Please reconnect the phone.') + raise RuntimeError('No BT modem found.') + # Log about found modem utils.log(f'BT modem found. Modem: {self.modem}') diff --git a/src/utils_cache.py b/src/utils_cache.py index 735b36c..55c519c 100644 --- a/src/utils_cache.py +++ b/src/utils_cache.py @@ -5,6 +5,7 @@ import os import shutil import json import uuid +import utils class InfoCache: dir: Path @@ -12,7 +13,12 @@ class InfoCache: def __init__(self, dir: Path) -> None: self.dir = dir if dir is not None and not dir.exists(): - os.mkdir(dir) + try: + os.mkdir(dir) + except Exception as e: + utils.log_error(str(e)) + self.dir = None + def is_active(self) -> bool: @@ -66,10 +72,13 @@ class InfoCache: f.write(phone.dump()) def get_phone(self, name: str) -> Phone: - p = self.dir / f'phone_{name}.json', 'wt' - if p.exists(): - with open(p, 'rt') as f: - return Phone.make(f.read()) + p = self.dir / f'phone_{name}.json' + if not p.exists(): + utils.log(f'Phone definition at path {p} not found.') + return None + + with open(p, 'rt') as f: + return Phone.make(f.read()) def put_tasks(self, name: str, tasks: TaskList): p = self.dir / f'tasks_{name}.json' @@ -79,6 +88,7 @@ class InfoCache: def get_tasks(self, name: str) -> TaskList: p = self.dir / f'tasks_{name}.json' + # ToDo def add_report(self, report: dict) -> str: diff --git a/src/utils_qualtest.py b/src/utils_qualtest.py index 5a28d14..1adc2dc 100644 --- a/src/utils_qualtest.py +++ b/src/utils_qualtest.py @@ -154,11 +154,13 @@ class QualtestBackend: try: response = urllib.request.urlopen(url, timeout=utils.NETWORK_TIMEOUT) if response.getcode() != 200: - utils.log_error("Failed to get task list. Error code: %s" % response.getcode()) - return None + raise RuntimeError(f'Failed to load phone definition. Error code: {response.getcode()}') except Exception as e: - utils.log_error(f'Problem when loading the phone definition.') - return cache.get_phone(self.instance) + utils.log_error(f'Problem when loading the phone definition from backend. Error: {str(e)}') + r = cache.get_phone(self.instance) + if r is None: + raise RuntimeError(f'No cached phone definition.') + return r # Get possible list of phones phones = json.loads(response.read().decode()) @@ -199,8 +201,8 @@ class QualtestBackend: return result except Exception as err: - utils.log_error("Exception when fetching task list: {0}".format(err)) - return dict() + utils.log_error(f"Exception when fetching task list: {str(err)}") + return None def load_audio(self, audio_id: int, output_path: Path): diff --git a/src/utils_sevana.py b/src/utils_sevana.py index 9af0635..7e9b8ca 100644 --- a/src/utils_sevana.py +++ b/src/utils_sevana.py @@ -95,7 +95,7 @@ def find_binaries(bin_directory: Path, license_server: str = None): SILER_PATH = bin_directory / platform_prefix / SILER_PATH SPEECH_DETECTOR_PATH = bin_directory / platform_prefix / SPEECH_DETECTOR_PATH - print(f'Looking for binaries/licenses/configs at {directory}...', end=' ') + print(f'Looking for binaries/licenses/configs at {bin_directory}...', end=' ') # Check if binaries exist if not PVQA_PATH.exists():