diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..8071342 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,13 @@ +name: CI +on: pull_request +jobs: + ruff: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/ruff-action@v3 + with: + args: "check --output-format=github" + - uses: astral-sh/ruff-action@v3 + with: + args: "format --check --diff --preview" diff --git a/.gitignore b/.gitignore index 943a903..51965e4 100644 --- a/.gitignore +++ b/.gitignore @@ -16,4 +16,7 @@ config/*.txt # Generated files *.ndjson -.ruff_cache/ \ No newline at end of file +.ruff_cache/ + +# Virtual environment +.venv/ \ No newline at end of file diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml new file mode 100644 index 0000000..436e530 --- /dev/null +++ b/.pre-commit-config.yaml @@ -0,0 +1,6 @@ +repos: + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.15.6 + hooks: + - id: ruff-check + - id: ruff-format diff --git a/README.md b/README.md index ecb2790..2ee6f45 100644 --- a/README.md +++ b/README.md @@ -1,52 +1,84 @@ # ISMU + Information system master unit ## How to Run the Project on Raspberry Pi Pico W Using VSCode + ### Prerequisites + 1. **Hardware:** - - Raspberry Pi Pico W. - - USB cable with data transfer capability. + - Raspberry Pi Pico W. + - USB cable with data transfer capability. -2. **Software:** - - [Visual Studio Code (VSCode)](https://code.visualstudio.com/) installed. - - VSCode Extension: MicroPico. - - Python 3.x (preferably version 3.11 or newer). - - MicroPython UF2 file installed on the Raspberry Pi Pico W. -> Follow the [official MicroPython setup guide](https://www.raspberrypi.com/documentation/microcontrollers/micropython.html) for installing it on your Pico. +2. **Software:** - [Visual Studio Code (VSCode)](https://code.visualstudio.com/) installed. - VSCode Extension: MicroPico. - Python 3.x (preferably version 3.11 or newer). - MicroPython UF2 file installed on the Raspberry Pi Pico W. + > Follow the [official MicroPython setup guide](https://www.raspberrypi.com/documentation/microcontrollers/micropython.html) for installing it on your Pico. ### Steps to Configure and Run the Project + #### 1. Install and Configure **MicroPico** Extension + - Open Visual Studio Code. - Go to the **Extensions** panel and install the **MicroPico** extension. - After installing, ensure your Raspberry Pi Pico is connected to your computer via USB. #### 2. Clone the Project Repository + - Copy the repository to your local machine: -``` bash + +```bash git clone https://github.com/publictransitdata/ISMU.git cd ISMU ``` + #### 3. Make Sure Required Files Are in Place + - Ensure Python files for your project are located in a directory that will be synced to the Pico. Typically, this is the project’s root directory. #### 4. Open the Project in VSCode + - Launch Visual Studio Code and open the project directory: -``` bash + +```bash code . ``` -#### 5. Initialize MicroPico project + +#### 5. Initialize virtual environment in project directory + +``` +python -m venv .venv +source .venv/bin/activate +``` + +#### 6. Install all required dependencies(if you want to develop project install also dev dependencies) + +``` +pip install -r requirements.txt +opptionally: +pip install -r requirements-dev.txt +``` + +#### 7. Set up the git hook scripts + +``` +pre-commit install +``` + +#### 8. Initialize MicroPico project + - **Right-click** on area in folder/project view. - In the context menu that appears, select **Initialize MicroPico project** -#### 6. Toggle virtual MicroPico workspace +#### 9. Toggle virtual MicroPico workspace + - At the bottom of vs studio you will see a button with the same name and you have to click it -#### 7. Upload Code to Pico +#### 10. Upload Code to Pico + - To upload your code: - **Right-click** on the file you want to upload in the side panel (or folder/project view). - In the context menu that appears, select **Upload File to Pico** -#### 8. Run the Script - - **Right-click** on the file you want to run In Mpy Remote Workspace. - - In the context menu that appears, select **run current file on Pico** +#### 11. Run the Script +- **Right-click** on the file you want to run In Mpy Remote Workspace. +- In the context menu that appears, select **run current file on Pico** diff --git a/app/config_management/__init__.py b/app/config_management/__init__.py index 2e8d272..6ef99e9 100644 --- a/app/config_management/__init__.py +++ b/app/config_management/__init__.py @@ -1,4 +1,4 @@ -from .config_info import CurrentRouteTripSelection, SystemConfig +from .config_info import SystemConfig from .config_manager import ConfigManager -__all__ = ["SystemConfig", "ConfigManager", "CurrentRouteTripSelection"] +__all__ = ["SystemConfig", "ConfigManager"] diff --git a/app/config_management/config_info.py b/app/config_management/config_info.py index 24be639..42e87bd 100644 --- a/app/config_management/config_info.py +++ b/app/config_management/config_info.py @@ -1,230 +1,26 @@ -from app.routes_management import RoutesManager -from app.state_management import StateManager from utils.singleton_decorator import singleton AP_NAME = "ismu-hotspot" AP_PASSWORD = "12345678" AP_IP = "192.168.4.1" +VERSION = "1.0.0" @singleton class SystemConfig: def __init__(self): - self._line_telegram: str = "" - self._destination_number_telegram: str = "" - self._destination_telegram: str = "" - self._show_start_and_end_stops: bool = False - self._force_short_names: bool = False - self._stop_board_telegram: str = "" - self._show_info_on_stop_board: bool = False - self._ap_name: str = AP_NAME - self._ap_password: str = AP_PASSWORD - self._ap_ip: str = AP_IP - self._baudrate: int = 1200 - self._bits: int = 7 - self._parity: int = 2 - self._stop: int = 2 - self._version: str = "1.0.0" - - @property - def line_telegram(self): - return self._line_telegram - - @line_telegram.setter - def line_telegram(self, value): - self._line_telegram = value - - @property - def destination_number_telegram(self): - return self._destination_number_telegram - - @destination_number_telegram.setter - def destination_number_telegram(self, value): - self._destination_number_telegram = value - - @property - def destination_telegram(self): - return self._destination_telegram - - @destination_telegram.setter - def destination_telegram(self, value): - self._destination_telegram = value - - @property - def show_start_and_end_stops(self): - return self._show_start_and_end_stops - - @show_start_and_end_stops.setter - def show_start_and_end_stops(self, value): - self._show_start_and_end_stops = value - - @property - def force_short_names(self): - return self._force_short_names - - @force_short_names.setter - def force_short_names(self, value): - self._force_short_names = value - - @property - def stop_board_telegram(self): - return self._stop_board_telegram - - @stop_board_telegram.setter - def stop_board_telegram(self, value): - self._stop_board_telegram = value - - @property - def show_info_on_stop_board(self): - return self._show_info_on_stop_board - - @show_info_on_stop_board.setter - def show_info_on_stop_board(self, value): - self._show_info_on_stop_board = value - - @property - def ap_name(self): - return self._ap_name - - @ap_name.setter - def ap_name(self, value): - self._ap_name = value - - @property - def ap_password(self): - return self._ap_password - - @ap_password.setter - def ap_password(self, value): - self._ap_password = value - - @property - def ap_ip(self): - return self._ap_ip - - @ap_ip.setter - def ap_ip(self, value): - self._ap_ip = value - - @property - def version(self): - return self._version - - @version.setter - def version(self, value): - self._version = value - - @property - def baudrate(self): - return self._baudrate - - @baudrate.setter - def baudrate(self, value): - self._baudrate = value - - @property - def bits(self): - return self._bits - - @bits.setter - def bits(self, value): - self._bits = value - - @property - def parity(self): - return self._parity - - @parity.setter - def parity(self, value): - self._parity = value - - @property - def stop(self): - return self._stop - - @stop.setter - def stop(self, value): - self._stop = value - - -class TripInfo: - def __init__( - self, - group_id: str, - point_id: str, - full_name: list[str], - short_name: list[str] | None, - ): - self.group_id = group_id - self.point_id = point_id - self.full_name = full_name - self.short_name = short_name - - @staticmethod - def trip_from_dict(d: dict | None): - if d is not None: - group_id = d.get("trip_id", "") - point_id = d.get("point_id", "") - full_name = d.get("full_name") or [] - short_name = d.get("short_name") or [] - return TripInfo(group_id, point_id, full_name, short_name) - else: - return None - - def get_proper_trip_name(self) -> list[str]: - if SystemConfig().force_short_names: - if self.short_name: - return self.short_name - else: - return self.full_name - else: - return self.full_name - - -class CurrentRouteTripSelection: - def __init__( - self, - route_number: str | None = None, - trip: dict | None = None, - no_line_telegram: bool = False, - ): - self.route_number = route_number - self.trip = TripInfo.trip_from_dict(trip) - self.no_line_telegram = no_line_telegram - self.is_updated = False - - def load_from_saved_state(self): - state = StateManager().get_state() - route = RoutesManager().get_route_by_index(state["route_id"]) - - if route and route.get("dirs"): - self._route_number = route["route_number"] - dirs = route["dirs"] - trip_id = state.get("trip_id", 0) - if trip_id < len(dirs): - self._trip = TripInfo.trip_from_dict(dirs[trip_id]) - self._no_line_telegram = route.get("no_line_telegram", False) - - @property - def route_number(self): - return self._route_number - - @route_number.setter - def route_number(self, value): - self._route_number = value - - @property - def trip(self): - return self._trip - - @trip.setter - def trip(self, value): - self._trip = value - - @property - def no_line_telegram(self): - return self._no_line_telegram - - @no_line_telegram.setter - def no_line_telegram(self, value): - self._no_line_telegram = value + self.line_telegram: str = "" + self.destination_number_telegram: str = "" + self.destination_telegram: str = "" + self.show_start_and_end_stops: bool = False + self.force_short_names: bool = False + self.stop_board_telegram: str = "" + self.show_info_on_stop_board: bool = False + self.ap_name: str = AP_NAME + self.ap_password: str = AP_PASSWORD + self.ap_ip: str = AP_IP + self.baudrate: int = 1200 + self.bits: int = 7 + self.parity: int = 2 + self.stop: int = 2 + self.version: str = VERSION diff --git a/app/config_management/config_manager.py b/app/config_management/config_manager.py index 4b5a66a..48ad1c6 100644 --- a/app/config_management/config_manager.py +++ b/app/config_management/config_manager.py @@ -1,15 +1,17 @@ +import gc + from app.error_codes import ErrorCodes +from utils.custom_error import CustomError from utils.error_handler import set_error_and_raise from utils.singleton_decorator import singleton -from .config_info import CurrentRouteTripSelection, SystemConfig, TripInfo +from .config_info import SystemConfig @singleton class ConfigManager: def __init__(self): - self._config = SystemConfig() - self._current_selection = CurrentRouteTripSelection() + self.config = SystemConfig() def _convert_value(self, key: str, value: str): if key in { @@ -18,46 +20,42 @@ def _convert_value(self, key: str, value: str): "show_info_on_stop_board", }: if value.lower() not in ("true", "false"): - set_error_and_raise( + raise CustomError( ErrorCodes.CONFIG_INVALID_VALUE, - ValueError(f"Expected 'true' or 'false' for {key}, got '{value}'"), - show_message=True, + f"Expected 'true' or 'false' for {key}, got '{value}'", ) return value.lower() == "true" if key in {"baudrate", "bits", "parity", "stop"}: try: return int(value) - except ValueError: - set_error_and_raise( + except ValueError as err: + raise CustomError( ErrorCodes.CONFIG_INVALID_VALUE, - ValueError(f"Could not convert {key}={value} to int"), - show_message=True, - ) - + f"Could not convert {key}={value} to int", + ) from err return value def load_config(self, config_path: str) -> None: try: - with open(config_path, "r") as file: + with open(config_path) as file: content = file.read() if not content.strip(): - set_error_and_raise( - ErrorCodes.CONFIG_FILE_EMPTY, - Exception("Config file is empty"), - show_message=True, - ) + raise CustomError(ErrorCodes.CONFIG_FILE_EMPTY, "Config file is empty") lines = content.splitlines() self._parse_config(lines) print("Config was loaded.") - except OSError as e: + gc.collect() + except CustomError as err: + set_error_and_raise(err.error_code, err, show_message=True, raise_exception=False) + except OSError as err: # errno 2 = ENOENT (file not found) - if e.args[0] == 2: - set_error_and_raise(ErrorCodes.CONFIG_FILE_NOT_FOUND, e) + if err.args[0] == 2: + set_error_and_raise(ErrorCodes.CONFIG_FILE_NOT_FOUND, err, raise_exception=False) else: - set_error_and_raise(ErrorCodes.CONFIG_IO_ERROR, e) + set_error_and_raise(ErrorCodes.CONFIG_IO_ERROR, err, raise_exception=False) def _parse_config(self, lines: list[str]) -> None: for line in lines: @@ -66,10 +64,10 @@ def _parse_config(self, lines: list[str]) -> None: continue if "=" not in line: - set_error_and_raise(ErrorCodes.CONFIG_NO_EQUALS_SIGN) + raise CustomError(ErrorCodes.CONFIG_NO_EQUALS_SIGN, f"Missing '=' in line: {line}") key, value = map(str.strip, line.split("=", 1)) - if hasattr(self._config, key): + if hasattr(self.config, key): if key in { "line_telegram", "destination_number_telegram", @@ -79,22 +77,9 @@ def _parse_config(self, lines: list[str]) -> None: converted = value if value else None else: converted = self._convert_value(key, value) - setattr(self._config, key, converted) + setattr(self.config, key, converted) else: - set_error_and_raise(ErrorCodes.CONFIG_UNKNOWN_KEY) - - @property - def config(self): - return self._config - - def update_current_selection(self, route_number, trip, no_line_telegram=False): - self._current_selection.route_number = route_number - self._current_selection.trip = TripInfo.trip_from_dict(trip) - self._current_selection.no_line_telegram = no_line_telegram - self._current_selection.is_updated = True - - def get_current_selection(self): - return self._current_selection + raise CustomError(ErrorCodes.CONFIG_UNKNOWN_KEY, f"Unknown key: {key}") def get_telegram_types(self): keys = [ @@ -103,4 +88,4 @@ def get_telegram_types(self): "destination_telegram", "stop_board_telegram", ] - return {getattr(self._config, k) for k in keys if getattr(self._config, k)} + return {getattr(self.config, k) for k in keys if getattr(self.config, k)} diff --git a/app/error_codes.py b/app/error_codes.py index 1ce2e0a..ae3b0bd 100644 --- a/app/error_codes.py +++ b/app/error_codes.py @@ -1,9 +1,10 @@ class ErrorCodes: + NONE = 0 # Config errors (1XX) # File errors (10X) CONFIG_FILE_NOT_FOUND = 100 CONFIG_IO_ERROR = 101 - TEMP_STATE_WRITE_ERROR = 102 + TEMP_SELECTION_WRITE_ERROR = 102 CONFIG_EXAMPLE_EXIST = 103 CONFIG_FILE_LOAD_ERROR = 104 CONFIG_FILE_EMPTY = 105 @@ -24,6 +25,7 @@ class ErrorCodes: ROUTES_DB_WRITE_FAILED = 203 ROUTES_FILE_OPEN_FAILED = 204 ROUTES_FILE_LOAD_ERROR = 205 + ROUTES_DB_DELETE_FAILED = 206 # Route number errors (21X) ROUTES_EMPTY_ROUTE_NUMBER = 210 @@ -50,6 +52,11 @@ class ErrorCodes: TRIP_INFO_IS_NONE = 312 CHAR_MAP_LOAD_ERROR = 313 POINT_ID_IS_NONE = 314 + ROUTE_VALUE_IS_WRONG = 315 + POINT_ID_VALUE_IS_WRONG = 316 + TRIP_NAME_IS_WRONG = 317 + TRIP_NAME_IS_NONE = 318 + TRIP_NAME_OR_ROUTE_NUMBER_IS_WRONG = 319 # Files errors (4XX) # File not found (40X) @@ -63,60 +70,79 @@ class ErrorCodes: WEB_SERVER_ERROR = 601 REFRESH_ROUTES_DB_ERROR = 602 - MESSAGES = { - # Config - 100: "E100: Config not found", - 101: "E101: Config IO error", - 102: "E102: Temp state write error", - 103: "E103: Rename config.example to config.txt and fill keys to start", - 104: "E104: Config file load error", - 105: "E105: Config file empty", - # Config - parse - 110: "E110: Config parse fail", - 111: "E111: Missing '=' in line", - # Config - validation - 120: "E120: Unknown config key", - 121: "E121: Invalid config val", - 122: "E122: Expected integer", - 123: "E123: Expected true/false", - # Routes - file - 200: "E200: Routes not found", - 201: "E201: Routes file empty", - 202: "E202: Routes DB open fail", - 203: "E203: Routes DB write fail", - 204: "E204: Routes file open fail", - 205: "E205: Routes file load error", - # Routes - route number - 210: "E210: Empty route number", - 211: "E211: No routes in file", - # Routes - direction - 220: "E220: Dir without route", - 221: "E221: Empty dir/point ID", - 222: "E222: Wrong field count", - # Routes - short name - 230: "E230: Short name no '^'", - 231: "E231: Short name <2 parts", - # IBIS - codes - 300: "E300: DS001 error", - 301: "E301: DS001NEU error", - 302: "E302: DS003 error, routes numbers should have only numbers(no letters or symbols)", - 303: "E303: DS003A error", - # IBIS - data - 310: "E310: Unknown telegram type", - 311: "E311: Route number is None", - 312: "E312: Trip info is None", - 313: "E313: Char map load error. There is should be correct char_map.json file in config directory. Look readme for details.", - 314: "E314: Point ID is None", - # Files - 400: "E400: Missing language file. There is should be correct lang.py file in config directory. Look readme for details.", - # GUI - 500: "E500: Unknown menu type", - # Web server - 600: "E600: Web server shutdown error", - 601: "E601: Web server error", - 602: "E602: Refresh routes DB error", - } + # Main loop errors (7XX) + MAIN_LOOP_ERROR = 700 + GUI_LOOP_ERROR = 701 + + MESSAGES = None + + @classmethod + def _load_messages(cls): + cls.MESSAGES = { + # Config + 100: "Файл конфігурації не знайдено", + 101: "Помилка IO конфігурації", + 102: "Помилка запису тим. вибору", + 103: "Перейменуйте config.example на config.txt та заповніть параметри", + 104: "Помилка завантаження конфігурації", + 105: "Файл конфігурації порожній", + # Config - parse + 110: "Помилка парсування конфігурації", + 111: "Відсутній символ '=' у рядку", + # Config - validation + 120: "Невідомий ключ конфігурації", + 121: "Невірне значення конфігурації", + 122: "Очікується ціле число", + 123: "Очікується true/false", + # Routes - file + 200: "Файл маршрутів не знайдено", + 201: "Файл маршрутів порожній", + 202: "Помилка відкриття БД маршрутів", + 203: "Помилка запису БД маршрутів", + 204: "Помилка відкриття файлу маршрутів", + 205: "Помилка завантаження маршрутів", + 206: "Помилка видалення БД маршрутів", + # Routes - route number + 210: "Порожній номер маршруту", + 211: "Маршрути у файлі відсутні", + # Routes - direction + 220: "Напрямок без маршруту", + 221: "Порожній ID напрямку або зупинки", + 222: "Неправильна кількість полей", + # Routes - short name + 230: "Відсутній символ '^' у короткій назві", + 231: "Коротка назва має менше 2 частин", + # IBIS - codes + 300: "DS001", + 301: "DS001NEU", + 302: "DS003,в номері маршруту лише цифри", + 303: "DS003A", + # IBIS - data + 310: "Невідомий тип телеграми", + 311: "Номер маршруту відсутній", + 312: "Інформація про рейс відсутня", + 313: "Помилка завантаження таблиці символів. Файл char_map.json має бути у папці config. Дивіться readme.", + 314: "ID зупинки відсутній", + 315: "Невірне значення маршруту", + 316: "Невірне значення ID зупинки", + 317: "Невірна назва рейсу", + 318: "Назва рейсу відсутня", + 319: "Невірна код рейсу/номер маршруту", + # Files + 400: "Відсутній файл мови. Файл lang.py має бути у папці config. Дивіться readme.", + # GUI + 500: "Невідомий тип меню", + # Web server + 600: "Помилка зупинки веб-сервера", + 601: "Помилка веб-сервера", + 602: "Помилка оновлення БД маршрутів", + # Main loop + 700: "Помилка в головному циклі", + 701: "Помилка в циклі GUI", + } @classmethod def get_message(cls, code: int) -> str: - return cls.MESSAGES.get(code, f"E{code}: Unknown error") + if cls.MESSAGES is None: + cls._load_messages() + return cls.MESSAGES.get(code, f"E{code}: Невідома помилка") diff --git a/app/gui_management/__init__.py b/app/gui_management/__init__.py index 84baedd..a918664 100644 --- a/app/gui_management/__init__.py +++ b/app/gui_management/__init__.py @@ -1,17 +1,17 @@ from .gui_config import ( - RouteMenuState, + RouteMenuData, ScreenConfig, - ScreenStates, - TripMenuState, + TripMenuData, ) from .gui_drawer import GuiDrawer -from .gui_manager import GuiManager +from .gui_manager import ErrorState, GuiManager, InitialState __all__ = [ "GuiManager", "ScreenConfig", - "RouteMenuState", - "TripMenuState", - "ScreenStates", + "RouteMenuData", + "TripMenuData", "GuiDrawer", + "InitialState", + "ErrorState", ] diff --git a/app/gui_management/gui_config.py b/app/gui_management/gui_config.py index 8cf827d..f07c3cb 100644 --- a/app/gui_management/gui_config.py +++ b/app/gui_management/gui_config.py @@ -1,42 +1,25 @@ -from app.state_management import StateManager +from app.selection_management import SelectionManager from utils.singleton_decorator import singleton -class ScreenStates: - STATUS_SCREEN = "status" - ROUTE_MENU = "route" - TRIP_MENU = "trip" - ERROR_SCREEN = "error" - SETTINGS_SCREEN = "settings" - UPDATE_SCREEN = "update" - INITIAL_SCREEN = "initial" - START_SCREEN = "start" - MESSAGE_SCREEN = "message" - - @singleton class ScreenConfig: """ Attributes: - _screen_width (int): in pixels. - _screen_height (int): in pixels. - _font_size (int): Font size used for rendering text. - _max_menu_items (int): Maximum number of items visible at once on screen. - _current_screen (str): Identifier of the current active screen. - _error_code (int): Error code to display on error screen. + screen_width (int): in pixels. + screen_height (int): in pixels. + font_size (int): Font size used for rendering text. + arrow_size (int): Size of the arrow used for navigation. + max_menu_items (int): Maximum number of items visible at once on screens of menu(route/trip). """ def __init__(self): - self._screen_width = 0 - self._screen_height = 0 - self._font_size = 0 - self._arrow_size = 0 - self._max_menu_items = 0 - self._max_number_of_characters_in_line = 0 - self._current_screen = ScreenStates.STATUS_SCREEN - self._error_code = 0 - self._message_to_display = None - self._dirty = True + self.screen_width = 0 + self.screen_height = 0 + self.font_size = 0 + self.arrow_size = 0 + self.max_menu_items = 0 + self.max_number_of_characters_in_line = 0 def set_screen_config( self, @@ -47,158 +30,56 @@ def set_screen_config( max_menu_items: int = 2, max_number_of_characters_in_line: int = 18, ): - self._screen_width = screen_width - self._screen_height = screen_height - self._font_size = font_size - self._arrow_size = arrow_size - self._max_menu_items = max_menu_items - self._max_number_of_characters_in_line = max_number_of_characters_in_line - - @property - def screen_width(self): - return self._screen_width - - @screen_width.setter - def screen_width(self, value: int): - self._screen_width = value - - @property - def screen_height(self): - return self._screen_height - - @screen_height.setter - def screen_height(self, value: int): - self._screen_height = value - - @property - def font_size(self): - return self._font_size - - @font_size.setter - def font_size(self, value: int): - self._font_size = value - - @property - def arrow_size(self): - return self._arrow_size + self.screen_width = screen_width + self.screen_height = screen_height + self.font_size = font_size + self.arrow_size = arrow_size + self.max_menu_items = max_menu_items + self.max_number_of_characters_in_line = max_number_of_characters_in_line - @arrow_size.setter - def arrow_size(self, value: int): - self._arrow_size = value - @property - def max_menu_items(self): - return self._max_menu_items - - @max_menu_items.setter - def max_menu_items(self, value: int): - self._max_menu_items = value - - @property - def current_screen(self): - return self._current_screen - - @current_screen.setter - def current_screen(self, value: str): - self._current_screen = value - - @property - def max_number_of_characters_in_line(self): - return self._max_number_of_characters_in_line - - @max_number_of_characters_in_line.setter - def max_number_of_characters_in_line(self, value: int): - self._max_number_of_characters_in_line = value - - @property - def error_code(self): - return self._error_code - - @error_code.setter - def error_code(self, value: int): - self._error_code = value - - @property - def message_to_display(self): - return self._message_to_display +class MenuData: + def __init__(self, selection_key: str): + self._selection_key = selection_key + self._selected_item_index: int | None = None + self._highlighted_item_index: int | None = None - @message_to_display.setter - def message_to_display(self, value: str | None): - self._message_to_display = value + def _load_from_selection(self) -> int: + selection = SelectionManager().get_selection_ids() + return selection[self._selection_key] @property - def dirty(self): - return self._dirty - - @dirty.setter - def dirty(self, value: bool): - self._dirty = value - - def mark_dirty(self): - self._dirty = True - - def mark_clean(self): - self._dirty = False - - def is_dirty(self): - return self._dirty - - -@singleton -class RouteMenuState: - def __init__(self): - self._selected_item_index = 0 - self._highlighted_item_index = 0 - - def load_from_saved_state(self): - state = StateManager().get_state() - self._selected_item_index = state["route_id"] - self._highlighted_item_index = state["route_id"] - - @property - def selected_item_index(self): + def selected_item_index(self) -> int: + if self._selected_item_index is None: + value = self._load_from_selection() + self._selected_item_index = value + self._highlighted_item_index = value return self._selected_item_index @selected_item_index.setter - def selected_item_index(self, value): + def selected_item_index(self, value: int): self._selected_item_index = value @property - def highlighted_item_index(self): + def highlighted_item_index(self) -> int: + if self._highlighted_item_index is None: + value = self._load_from_selection() + self._selected_item_index = value + self._highlighted_item_index = value return self._highlighted_item_index @highlighted_item_index.setter - def highlighted_item_index(self, value): + def highlighted_item_index(self, value: int): self._highlighted_item_index = value @singleton -class TripMenuState: +class RouteMenuData(MenuData): def __init__(self): - self._selected_item_index = 0 - self._highlighted_item_index = 0 - - def set_trip_state(self, trip_selected_item_index: int = 0): - self._selected_item_index = trip_selected_item_index - self._highlighted_item_index = trip_selected_item_index + super().__init__(selection_key="route_id") - def load_from_saved_state(self): - state = StateManager().get_state() - self._selected_item_index = state["trip_id"] - self._highlighted_item_index = state["trip_id"] - @property - def selected_item_index(self): - return self._selected_item_index - - @selected_item_index.setter - def selected_item_index(self, value): - self._selected_item_index = value - - @property - def highlighted_item_index(self): - return self._highlighted_item_index - - @highlighted_item_index.setter - def highlighted_item_index(self, value): - self._highlighted_item_index = value +@singleton +class TripMenuData(MenuData): + def __init__(self): + super().__init__(selection_key="trip_id") diff --git a/app/gui_management/gui_drawer.py b/app/gui_management/gui_drawer.py index 6b675d9..9d4ecfc 100644 --- a/app/gui_management/gui_drawer.py +++ b/app/gui_management/gui_drawer.py @@ -14,7 +14,6 @@ def __init__( self, display: SH1106_I2C, writer: Writer, - screen_config: ScreenConfig, ): """ Initializes the GuiDrawer with the necessary configurations and display components. @@ -22,11 +21,10 @@ def __init__( Args: display: The display object used for rendering content on the screen. writers: The writer objects used for rendering string_line on the screen. - screen_config: Configuration for the screen dimensions and properties. """ self._display = display self._writer = writer - self._screen_config = screen_config + self._screen_config = ScreenConfig() def _draw_menu( self, @@ -67,15 +65,10 @@ def _draw_menu( self._writer.set_textpos(self._display, 0, suffix_x) self._writer.printstring(header_suffix, False) - self._display.vline(separator_x, 0, line_height + 2, 1) + self._display.vline(separator_x, 0, line_height, 1) + self._display.fill_rect(separator_x, line_height - 1, self._screen_config.screen_width, 1, 1) - self._display.fill_rect( - separator_x, line_height + 1, self._screen_config.screen_width, 1, 1 - ) - - first_visible_menu_item_idx = ( - highlighted_item_index // max_menu_items - ) * max_menu_items + first_visible_menu_item_idx = (highlighted_item_index // max_menu_items) * max_menu_items last_visible_menu_item_idx = min( first_visible_menu_item_idx + max_menu_items, len(menu_items), @@ -102,9 +95,7 @@ def _draw_menu( self._display.show() - def trim_text_to_fit( - self, string_line: str, max_number_of_characters_in_line: int = 18 - ) -> str: + def trim_text_to_fit(self, string_line: str, max_number_of_characters_in_line: int = 18) -> str: return string_line[0:max_number_of_characters_in_line] def draw_status_screen( @@ -133,9 +124,7 @@ def draw_status_screen( route_text_width = self._writer.stringlen(f"М:{selected_route_id}") trip_text_width = self._writer.stringlen(f"Н:{selected_trip_id:02d}") - self._writer.set_textpos( - self._display, bottom_y, left_offset + route_text_width + 3 - ) + self._writer.set_textpos(self._display, bottom_y, left_offset + route_text_width + 3) self._writer.printstring( f"Н:{selected_trip_id:02d}", @@ -191,7 +180,7 @@ def draw_error_screen(self, error_code: str, message: str | None = None) -> None self._display.show() - def draw_message_screen(self, message: str | None) -> None: + def draw_message_screen(self, message: str, error_code: int | None) -> None: self._display.fill(0) line_height = self._screen_config.font_size + 2 @@ -199,16 +188,28 @@ def draw_message_screen(self, message: str | None) -> None: screen_height = self._screen_config.screen_height bottom_y = screen_height - line_height - note_for_user = ">Натисни OK<" - message_width = self._writer.stringlen(note_for_user) - message_offset = (screen_width - message_width) // 2 + note_width = self._writer.stringlen(note_for_user) + note_offset = (screen_width - note_width) // 2 - self._writer.set_textpos(self._display, 0, 0) - self._writer.printstring(message, False) + if error_code is not None: + line1 = f"E:{error_code}" + line1_width = self._writer.stringlen(line1) + + self._writer.set_textpos(self._display, 0, 0) + self._writer.printstring(line1, False) - self._writer.set_textpos(self._display, bottom_y, message_offset) + self._writer.set_textpos(self._display, 0, line1_width + 5) + self._writer.printstring(message, False) + + self._display.vline(line1_width + 2, 0, line_height, 1) + self._display.fill_rect(0, line_height - 1, line1_width + 2, 1, 1) + else: + self._writer.set_textpos(self._display, 0, 0) + self._writer.printstring(message, False) + + self._writer.set_textpos(self._display, bottom_y, note_offset) self._writer.printstring(note_for_user, False) self._display.show() @@ -216,9 +217,7 @@ def draw_message_screen(self, message: str | None) -> None: def draw_initial_screen(self) -> None: self._display.fill(0) self._writer.set_textpos(self._display, 0, 0) - self._writer.printstring( - "Потрібно завантажити файли конфігурації та маршрутів", False - ) + self._writer.printstring("Потрібно завантажити файли конфігурації та маршрутів", False) self._display.show() def draw_update_mode_screen(self, ip_address: str, ap_name: str) -> None: @@ -249,9 +248,7 @@ def draw_update_mode_screen(self, ip_address: str, ap_name: str) -> None: self._writer.set_textpos(self._display, top_y + line_height + 2, line2_offset) self._writer.printstring(line2, False) - self._writer.set_textpos( - self._display, top_y + line_height * 2 + 2, line3_offset - ) + self._writer.set_textpos(self._display, top_y + line_height * 2 + 2, line3_offset) self._writer.printstring(line3, False) self._display.show() @@ -275,13 +272,13 @@ def draw_active_settings_screen(self, config) -> None: telegrams_text = ", ".join(filtered_telegrams) self._writer.printstring( - f"Telegrams: {telegrams_text}", + f"Телеграми: {telegrams_text}", False, ) bottom_y = screen_height - line_height self._writer.set_textpos(self._display, bottom_y, left_offset) - self._writer.printstring(f"ver:{config.version}", False) + self._writer.printstring(f"вер:{config.version}", False) self._display.show() @@ -302,9 +299,7 @@ def draw_menu_items( string_line = self.trim_text_to_fit(menu_items[i], available_width) if is_highlighted: - self._display.fill_rect( - 0, y, self._screen_config.screen_width, line_height, 1 - ) + self._display.fill_rect(0, y, self._screen_config.screen_width, line_height, 1) self._writer.set_textpos(self._display, y, left_offset) self._writer.printstring(string_line, True) else: diff --git a/app/gui_management/gui_manager.py b/app/gui_management/gui_manager.py index e39dedd..a020d5a 100644 --- a/app/gui_management/gui_manager.py +++ b/app/gui_management/gui_manager.py @@ -1,21 +1,36 @@ +# isort: skip_file +import gc import sys import time import ujson as json + +from app.web_update import WebUpdateServer from app.config_management import ConfigManager from app.error_codes import ErrorCodes from app.routes_management import RoutesManager -from app.state_management import StateManager -from app.web_update import WebUpdateServer +from app.selection_management import SelectionManager from utils.error_handler import set_error_and_raise +from utils.gui_hooks import ( + register_error_hook, + register_initial_hook, + register_message_hook, +) from .gui_config import ( + RouteMenuData, + TripMenuData, +) +from .gui_drawer import GuiDrawer +from .states import ( + ErrorState, + InitialState, + MessageState, RouteMenuState, - ScreenConfig, - ScreenStates, + State, + StatusState, TripMenuState, ) -from .gui_drawer import GuiDrawer if sys.platform != "rp2": from lib.sh1106 import SH1106_I2C # for vs code @@ -23,9 +38,7 @@ class GuiManager: - def __init__( - self, display: SH1106_I2C, writer: Writer, screen_config: ScreenConfig - ): + def __init__(self, display: SH1106_I2C, writer: Writer): """ Initializes the GuiManager with the necessary configurations and display components. @@ -36,134 +49,108 @@ def __init__( """ self._routes_manager = RoutesManager() self._config_manager = ConfigManager() - self._route_menu_state = RouteMenuState() - self._trip_menu_state = TripMenuState() - self._screen_config = screen_config - self._state_manager = StateManager() self._web_update_server = WebUpdateServer( self._config_manager.config.ap_name, self._config_manager.config.ap_ip, self._config_manager.config.ap_password, ) - self._gui_drawer = GuiDrawer(display, writer, screen_config) + self._route_menu_data = RouteMenuData() + self._trip_menu_data = TripMenuData() + self._selection_manager = SelectionManager() + self._gui_drawer = GuiDrawer(display, writer) self._buttons_press_start_time = None self._buttons_press_active = False self._last_single_button_time = 0 self._single_button_cooldown = 150 - self._route_menu_state.load_from_saved_state() - self._trip_menu_state.load_from_saved_state() - self._routes_for_menu_display_list = [] # Cache for route display list - it optimizes performance + self._state = StatusState() + self.error_code = ErrorCodes.NONE + self._message_to_display = None + self._dirty = True + self.transition_to(self._state) + + register_error_hook(self._handle_error) + register_message_hook(self._handle_message) + register_initial_hook(self._handle_initial) + + def transition_to(self, state: State): + self._state = state + self._state.context = self + + def _handle_error(self, error_code: int, message: str | None): + self.error_code = error_code + self._message_to_display = message + self.transition_to(ErrorState()) + self.mark_dirty() + + def _handle_message(self, message: str, error_code: int | None): + self.error_code = error_code if error_code is not None else ErrorCodes.NONE + self._message_to_display = message + self.transition_to(MessageState()) + self.mark_dirty() + + def _handle_initial(self): + self.transition_to(InitialState()) + self.mark_dirty() + + def mark_dirty(self): + self._dirty = True + + def mark_clean(self): + self._dirty = False + + def is_dirty(self): + return self._dirty + def draw_current_screen(self): if not self.is_dirty(): return - current_screen = self._screen_config.current_screen - - if current_screen == ScreenStates.ROUTE_MENU: - if len(self._routes_for_menu_display_list) == 0: - self._routes_for_menu_display_list = self.get_route_list_to_display( - self._routes_manager._db_file_path - ) - highlighted_item_index = self._get_menu_state( - ScreenStates.ROUTE_MENU - )._highlighted_item_index - number_of_menu_items = self.get_number_of_menu_items() - - self._gui_drawer._draw_menu( - self._routes_for_menu_display_list, - "Маршрут:", - highlighted_item_index, - number_of_menu_items, - ) - elif current_screen == ScreenStates.TRIP_MENU: - route = self._routes_manager.get_route_by_index( - self._route_menu_state.highlighted_item_index - ) - menu_items = self.get_trip_list_to_display(route) - highlighted_item_index = self._get_menu_state( - ScreenStates.TRIP_MENU - )._highlighted_item_index - number_of_menu_items = self.get_number_of_menu_items() - - self._gui_drawer._draw_menu( - menu_items, - "Напрямок:", - highlighted_item_index, - number_of_menu_items, - f"M:{route['route_number']}", - ) - elif current_screen == ScreenStates.STATUS_SCREEN: - route = self._routes_manager.get_route_by_index( - self._route_menu_state.selected_item_index - ) - selected_trip_name_list = route["dirs"][ - self._trip_menu_state.selected_item_index - ]["full_name"] - if len(selected_trip_name_list) == 2: - selected_trip_name = selected_trip_name_list[1] - else: - selected_trip_name = selected_trip_name_list[0] - - self._gui_drawer.draw_status_screen( - selected_trip_name, - route["route_number"], - self._trip_menu_state.selected_item_index + 1, - int( - route["dirs"][self._trip_menu_state.selected_item_index]["point_id"] - ), - ) - elif current_screen == ScreenStates.ERROR_SCREEN: - self._gui_drawer.draw_error_screen( - str(self._screen_config.error_code), - self._screen_config.message_to_display, - ) - - elif current_screen == ScreenStates.INITIAL_SCREEN: - self._gui_drawer.draw_initial_screen() - elif current_screen == ScreenStates.SETTINGS_SCREEN: - self._gui_drawer.draw_active_settings_screen(self._config_manager.config) - elif current_screen == ScreenStates.UPDATE_SCREEN: - self._gui_drawer.draw_update_mode_screen( - self._config_manager.config.ap_ip, self._config_manager.config.ap_name - ) - elif current_screen == ScreenStates.MESSAGE_SCREEN: - self._gui_drawer.draw_message_screen(self._screen_config.message_to_display) - - self._screen_config.mark_clean() + self._state.draw_current_screen() + gc.collect() + self.mark_clean() - def navigate_up(self, menu_type: str) -> None: - menu_state = self._get_menu_state(menu_type) + def navigate_up(self, menu_type: RouteMenuState | TripMenuState) -> None: + menu_state = self._get_menu_data(menu_type) if menu_state.highlighted_item_index > 0: menu_state.highlighted_item_index -= 1 - def navigate_down(self, menu_type: str) -> None: - menu_state = self._get_menu_state(menu_type) + def navigate_down(self, menu_type: RouteMenuState | TripMenuState) -> None: + menu_state = self._get_menu_data(menu_type) get_number_of_menu_items = self.get_number_of_menu_items() if menu_state.highlighted_item_index < get_number_of_menu_items - 1: menu_state.highlighted_item_index += 1 - def _get_menu_state(self, menu_type: str) -> RouteMenuState | TripMenuState: - if menu_type == ScreenStates.ROUTE_MENU: - return self._route_menu_state - elif menu_type == ScreenStates.TRIP_MENU: - return self._trip_menu_state + def _get_menu_data(self, menu_type: RouteMenuState | TripMenuState) -> RouteMenuData | TripMenuData: + if isinstance(menu_type, RouteMenuState): + return self._route_menu_data + elif isinstance(menu_type, TripMenuState): + return self._trip_menu_data else: set_error_and_raise( ErrorCodes.UNKNOWN_MENU_TYPE, ValueError(f"Unknown menu type: {menu_type}"), - True, + show_message=True, ) - def _check_buttons_press_timer( + def get_number_of_menu_items(self) -> int: + if isinstance(self._state, RouteMenuState): + return self._routes_manager.get_length_of_routes() + elif isinstance(self._state, TripMenuState): + return self._routes_manager.get_length_of_trips(self._route_menu_data.highlighted_item_index) + else: + return 0 + + def _is_in_cooldown(self, current_time) -> bool: + return time.ticks_diff(current_time, self._last_single_button_time) < self._single_button_cooldown + + def _is_long_pressed( self, buttons_pressed: list[int], - current_screen: str, - target_screen: str, current_time, ) -> bool: """ @@ -187,183 +174,14 @@ def _check_buttons_press_timer( duration = time.ticks_diff(current_time, self._buttons_press_start_time) if duration >= 3000: - if self._screen_config.current_screen == current_screen: - self._screen_config.current_screen = target_screen - - self._buttons_press_active = False - self._buttons_press_start_time = None - return True - else: - self._buttons_press_active = False - self._buttons_press_start_time = None + self._buttons_press_active = False + self._buttons_press_start_time = None + return True return False - def handle_buttons( - self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int - ) -> None: - current_time = time.ticks_ms() - - if not btn_up and not btn_down: - if self._check_buttons_press_timer( - [btn_up, btn_down], - ScreenStates.STATUS_SCREEN, - ScreenStates.SETTINGS_SCREEN, - current_time, - ): - self.mark_dirty() - return - return - - if not btn_down and not btn_select: - if self._screen_config.current_screen == ScreenStates.SETTINGS_SCREEN: - if self._check_buttons_press_timer( - [btn_down, btn_select], - ScreenStates.SETTINGS_SCREEN, - ScreenStates.UPDATE_SCREEN, - current_time, - ): - self._web_update_server.ensure_started() - self.mark_dirty() - return - - elif self._screen_config.current_screen == ScreenStates.ERROR_SCREEN: - if self._check_buttons_press_timer( - [btn_down, btn_select], - ScreenStates.ERROR_SCREEN, - ScreenStates.UPDATE_SCREEN, - current_time, - ): - self._web_update_server.ensure_started() - self.mark_dirty() - return - - elif self._screen_config.current_screen == ScreenStates.INITIAL_SCREEN: - if self._check_buttons_press_timer( - [btn_down, btn_select], - ScreenStates.INITIAL_SCREEN, - ScreenStates.UPDATE_SCREEN, - current_time, - ): - self._web_update_server.ensure_started() - self.mark_dirty() - return - return - - if ( - time.ticks_diff(current_time, self._last_single_button_time) - < self._single_button_cooldown - ): - return - - if not btn_menu: - if self._screen_config.current_screen == ScreenStates.STATUS_SCREEN: - self._screen_config.current_screen = ScreenStates.ROUTE_MENU - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.ROUTE_MENU: - self._screen_config.current_screen = ScreenStates.STATUS_SCREEN - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.TRIP_MENU: - self._screen_config.current_screen = ScreenStates.ROUTE_MENU - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.SETTINGS_SCREEN: - self._screen_config.current_screen = ScreenStates.STATUS_SCREEN - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.UPDATE_SCREEN: - if self._screen_config.error_code: - if self._check_buttons_press_timer( - [btn_menu], - ScreenStates.UPDATE_SCREEN, - ScreenStates.ERROR_SCREEN, - current_time, - ): - self._web_update_server.stop() - self.mark_dirty() - return - elif self._screen_config.current_screen == ScreenStates.INITIAL_SCREEN: - if self._check_buttons_press_timer( - [btn_menu], - ScreenStates.UPDATE_SCREEN, - ScreenStates.INITIAL_SCREEN, - current_time, - ): - self._web_update_server.stop() - self.mark_dirty() - return - else: - if self._check_buttons_press_timer( - [btn_menu], - ScreenStates.UPDATE_SCREEN, - ScreenStates.STATUS_SCREEN, - current_time, - ): - self._web_update_server.stop() - self.mark_dirty() - return - return - self._last_single_button_time = current_time - - if not btn_up: - if self._screen_config.current_screen in ( - ScreenStates.ROUTE_MENU, - ScreenStates.TRIP_MENU, - ): - self.navigate_up(self._screen_config.current_screen) - self.mark_dirty() - if self._screen_config.current_screen == ScreenStates.STATUS_SCREEN: - self._screen_config.current_screen = ScreenStates.TRIP_MENU - self.mark_dirty() - self._last_single_button_time = current_time - - if not btn_down: - if self._screen_config.current_screen in ( - ScreenStates.ROUTE_MENU, - ScreenStates.TRIP_MENU, - ): - self.navigate_down(self._screen_config.current_screen) - self.mark_dirty() - self._last_single_button_time = current_time - - if not btn_select: - if self._screen_config.current_screen == ScreenStates.ROUTE_MENU: - self._screen_config.current_screen = ScreenStates.TRIP_MENU - self._trip_menu_state.highlighted_item_index = 0 - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.TRIP_MENU: - self._route_menu_state.selected_item_index = ( - self._route_menu_state.highlighted_item_index - ) - self._trip_menu_state.selected_item_index = ( - self._trip_menu_state.highlighted_item_index - ) - route = self._routes_manager.get_route_by_index( - self._route_menu_state.selected_item_index - ) - self._config_manager.update_current_selection( - route["route_number"], - route["dirs"][self._trip_menu_state.selected_item_index], - route.get("no_line_telegram", False), - ) - self._state_manager.save_state( - self._route_menu_state.highlighted_item_index, - self._trip_menu_state.highlighted_item_index, - ) - self._screen_config.current_screen = ScreenStates.STATUS_SCREEN - self.mark_dirty() - elif self._screen_config.current_screen == ScreenStates.MESSAGE_SCREEN: - self._screen_config.current_screen = ScreenStates.STATUS_SCREEN - self.mark_dirty() - - self._last_single_button_time = current_time - - self._buttons_press_active = False - self._buttons_press_start_time = None - - def mark_dirty(self): - self._screen_config.mark_dirty() - - def is_dirty(self) -> bool: - return self._screen_config.is_dirty() + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int) -> None: + self._state.handle_buttons(btn_menu, btn_up, btn_down, btn_select) def get_route_list_to_display(self, route_file_path) -> list[str]: routes = self._routes_manager._route_list @@ -371,7 +189,7 @@ def get_route_list_to_display(self, route_file_path) -> list[str]: labels = {} try: - with open(route_file_path, "r") as f: + with open(route_file_path) as f: for line in f: try: record = json.loads(line) @@ -415,13 +233,3 @@ def get_trip_list_to_display(self, route) -> list[str]: menu_items.append(f"{d.get('trip_id')} {name}") return menu_items - - def get_number_of_menu_items(self) -> int: - if self._screen_config.current_screen == ScreenStates.ROUTE_MENU: - return self._routes_manager.get_length_of_routes() - elif self._screen_config.current_screen == ScreenStates.TRIP_MENU: - return self._routes_manager.get_length_of_trips( - self._route_menu_state.highlighted_item_index - ) - else: - return 0 diff --git a/app/gui_management/states/__init__.py b/app/gui_management/states/__init__.py new file mode 100644 index 0000000..551b988 --- /dev/null +++ b/app/gui_management/states/__init__.py @@ -0,0 +1,21 @@ +from .error_state import ErrorState +from .initial_state import InitialState +from .message_state import MessageState +from .route_menu_state import RouteMenuState +from .settings_state import SettingsState +from .state import State +from .status_state import StatusState +from .trip_menu_state import TripMenuState +from .update_state import UpdateState + +__all__ = [ + "State", + "StatusState", + "RouteMenuState", + "TripMenuState", + "SettingsState", + "UpdateState", + "ErrorState", + "MessageState", + "InitialState", +] diff --git a/app/gui_management/states/error_state.py b/app/gui_management/states/error_state.py new file mode 100644 index 0000000..f65856c --- /dev/null +++ b/app/gui_management/states/error_state.py @@ -0,0 +1,29 @@ +import time + +from .state import State + + +class ErrorState(State): + def draw_current_screen(self): + ctx = self.context + ctx._gui_drawer.draw_error_screen( + str(ctx.error_code), + ctx._message_to_display, + ) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .update_state import UpdateState + + current_time = time.ticks_ms() + ctx = self.context + + if not btn_down and not btn_select: + if ctx._is_long_pressed( + [btn_down, btn_select], + current_time, + ): + ctx._web_update_server.ensure_started() + ctx.transition_to(UpdateState(ErrorState())) + ctx.mark_dirty() + return + return diff --git a/app/gui_management/states/initial_state.py b/app/gui_management/states/initial_state.py new file mode 100644 index 0000000..26c4fb5 --- /dev/null +++ b/app/gui_management/states/initial_state.py @@ -0,0 +1,26 @@ +import time + +from .state import State + + +class InitialState(State): + def draw_current_screen(self): + ctx = self.context + ctx._gui_drawer.draw_initial_screen() + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .update_state import UpdateState + + current_time = time.ticks_ms() + ctx = self.context + + if not btn_down and not btn_select: + if ctx._is_long_pressed( + [btn_down, btn_select], + current_time, + ): + ctx.transition_to(UpdateState(InitialState())) + ctx._web_update_server.ensure_started() + ctx.mark_dirty() + return + return diff --git a/app/gui_management/states/message_state.py b/app/gui_management/states/message_state.py new file mode 100644 index 0000000..e4e58ba --- /dev/null +++ b/app/gui_management/states/message_state.py @@ -0,0 +1,27 @@ +import time + +from app.error_codes import ErrorCodes + +from .state import State + + +class MessageState(State): + def draw_current_screen(self): + ctx = self.context + error_code = ctx.error_code if ctx.error_code != ErrorCodes.NONE else None + ctx._gui_drawer.draw_message_screen(ctx._message_to_display, error_code) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .status_state import StatusState + + current_time = time.ticks_ms() + ctx = self.context + + if ctx._is_in_cooldown(current_time): + return + + if not btn_select: + ctx.transition_to(StatusState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return diff --git a/app/gui_management/states/route_menu_state.py b/app/gui_management/states/route_menu_state.py new file mode 100644 index 0000000..157b445 --- /dev/null +++ b/app/gui_management/states/route_menu_state.py @@ -0,0 +1,54 @@ +import time + +from .state import State + + +class RouteMenuState(State): + def draw_current_screen(self): + ctx = self.context + if len(ctx._routes_for_menu_display_list) == 0: + ctx._routes_for_menu_display_list = ctx.get_route_list_to_display(ctx._routes_manager._db_file_path) + highlighted_item_index = ctx._get_menu_data(self).highlighted_item_index + number_of_menu_items = ctx.get_number_of_menu_items() + + ctx._gui_drawer._draw_menu( + ctx._routes_for_menu_display_list, + "Маршрут:", + highlighted_item_index, + number_of_menu_items, + ) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .status_state import StatusState + from .trip_menu_state import TripMenuState + + current_time = time.ticks_ms() + ctx = self.context + + if ctx._is_in_cooldown(current_time): + return + + if not btn_menu: + ctx.transition_to(StatusState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_up: + ctx.navigate_up(self) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_down: + ctx.navigate_down(self) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_select: + ctx.transition_to(TripMenuState()) + ctx._trip_menu_data.highlighted_item_index = 0 + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return diff --git a/app/gui_management/states/settings_state.py b/app/gui_management/states/settings_state.py new file mode 100644 index 0000000..b200ca9 --- /dev/null +++ b/app/gui_management/states/settings_state.py @@ -0,0 +1,36 @@ +import time + +from .state import State + + +class SettingsState(State): + def draw_current_screen(self): + ctx = self.context + ctx._gui_drawer.draw_active_settings_screen(ctx._config_manager.config) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .status_state import StatusState + from .update_state import UpdateState + + current_time = time.ticks_ms() + ctx = self.context + + if not btn_down and not btn_select: + if ctx._is_long_pressed( + [btn_down, btn_select], + current_time, + ): + ctx._web_update_server.ensure_started() + ctx.transition_to(UpdateState(StatusState())) + ctx.mark_dirty() + return + return + + if ctx._is_in_cooldown(current_time): + return + + if not btn_menu: + ctx.transition_to(StatusState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return diff --git a/app/gui_management/states/state.py b/app/gui_management/states/state.py new file mode 100644 index 0000000..e156499 --- /dev/null +++ b/app/gui_management/states/state.py @@ -0,0 +1,14 @@ +class State: + @property + def context(self): + return self._context + + @context.setter + def context(self, context): + self._context = context + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + raise NotImplementedError("handle_buttons method should be implemented in the subclass") + + def draw_current_screen(self): + raise NotImplementedError("draw_current_screen method should be implemented in the subclass") diff --git a/app/gui_management/states/status_state.py b/app/gui_management/states/status_state.py new file mode 100644 index 0000000..7c0b61b --- /dev/null +++ b/app/gui_management/states/status_state.py @@ -0,0 +1,53 @@ +import time + +from .state import State + + +class StatusState(State): + def draw_current_screen(self): + ctx = self.context + route = ctx._routes_manager.get_route_by_index(ctx._route_menu_data.selected_item_index) + selected_trip_name_list = route["dirs"][ctx._trip_menu_data.selected_item_index]["full_name"] + if len(selected_trip_name_list) == 2: + selected_trip_name = selected_trip_name_list[1] + else: + selected_trip_name = selected_trip_name_list[0] + ctx._gui_drawer.draw_status_screen( + selected_trip_name, + route["route_number"], + ctx._trip_menu_data.selected_item_index + 1, + int(route["dirs"][ctx._trip_menu_data.selected_item_index]["point_id"]), + ) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .route_menu_state import RouteMenuState + from .settings_state import SettingsState + from .trip_menu_state import TripMenuState + + current_time = time.ticks_ms() + ctx = self.context + + if not btn_up and not btn_down: + if ctx._is_long_pressed( + [btn_up, btn_down], + current_time, + ): + ctx.transition_to(SettingsState()) + ctx.mark_dirty() + return + return + + if ctx._is_in_cooldown(current_time): + return + + if not btn_menu: + ctx.transition_to(RouteMenuState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_up: + ctx.transition_to(TripMenuState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return diff --git a/app/gui_management/states/trip_menu_state.py b/app/gui_management/states/trip_menu_state.py new file mode 100644 index 0000000..daab0e0 --- /dev/null +++ b/app/gui_management/states/trip_menu_state.py @@ -0,0 +1,59 @@ +import time + +from .state import State + + +class TripMenuState(State): + def draw_current_screen(self): + ctx = self.context + route = ctx._routes_manager.get_route_by_index(ctx._route_menu_data.highlighted_item_index) + menu_items = ctx.get_trip_list_to_display(route) + highlighted_item_index = ctx._get_menu_data(self).highlighted_item_index + number_of_menu_items = ctx.get_number_of_menu_items() + ctx._gui_drawer._draw_menu( + menu_items, + "Напрямок:", + highlighted_item_index, + number_of_menu_items, + f"M:{route['route_number']}", + ) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .route_menu_state import RouteMenuState + from .status_state import StatusState + + current_time = time.ticks_ms() + ctx = self.context + + if ctx._is_in_cooldown(current_time): + return + + if not btn_menu: + ctx.transition_to(RouteMenuState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_up: + ctx.navigate_up(self) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_down: + ctx.navigate_down(self) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + if not btn_select: + ctx._route_menu_data.selected_item_index = ctx._route_menu_data.highlighted_item_index + ctx._trip_menu_data.selected_item_index = ctx._trip_menu_data.highlighted_item_index + ctx._selection_manager.update_selection( + ctx._route_menu_data.highlighted_item_index, + ctx._trip_menu_data.highlighted_item_index, + ) + ctx.transition_to(StatusState()) + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return diff --git a/app/gui_management/states/update_state.py b/app/gui_management/states/update_state.py new file mode 100644 index 0000000..5f3561b --- /dev/null +++ b/app/gui_management/states/update_state.py @@ -0,0 +1,36 @@ +import gc +import time + +from .state import State + + +class UpdateState(State): + def __init__(self, return_state=None): + self._return_state = return_state + + def draw_current_screen(self): + ctx = self.context + ctx._gui_drawer.draw_update_mode_screen(ctx._config_manager.config.ap_ip, ctx._config_manager.config.ap_name) + + def handle_buttons(self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int): + from .status_state import StatusState + + current_time = time.ticks_ms() + ctx = self.context + + if ctx._is_in_cooldown(current_time): + return + + if not btn_menu: + if ctx._is_long_pressed( + [btn_menu], + current_time, + ): + ctx._web_update_server.stop() + ctx.transition_to(self._return_state or StatusState()) + gc.collect() + ctx.mark_dirty() + ctx._last_single_button_time = current_time + return + + return diff --git a/app/ibis_management/ibis_manager.py b/app/ibis_management/ibis_manager.py index d342876..eea9faf 100644 --- a/app/ibis_management/ibis_manager.py +++ b/app/ibis_management/ibis_manager.py @@ -1,9 +1,12 @@ import uasyncio as asyncio import ujson as json -from app.config_management import ConfigManager, SystemConfig + +from app.config_management import SystemConfig from app.error_codes import ErrorCodes +from app.selection_management import SelectionManager +from utils.custom_error import CustomError from utils.error_handler import set_error_and_raise -from utils.message_handler import set_message +from utils.gui_hooks import trigger_message from utils.singleton_decorator import singleton try: @@ -33,9 +36,9 @@ class IBISManager: def __init__(self, uart, telegramTypes): self.uart = uart self._running = False - self._task = None + self.task = None self.telegramTypes = telegramTypes - self.config_manager = ConfigManager() + self.selection_manager = SelectionManager() self._system_config = SystemConfig() self._failed_telegrams = set() @@ -84,62 +87,57 @@ def sanitize_ibis_text(self, text): return sanitized def DS001(self): - value = self.config_manager.get_current_selection().route_number + value = self.selection_manager.get_active_selection().route_number format = TELEGRAM_FORMATS["DS001"] if value is None: - set_error_and_raise(ErrorCodes.ROUTE_NUMBER_IS_NONE) + raise CustomError(ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться") try: formatted = format.format(int(value)) - except Exception: - set_message("Номер маршруту не виводиться") - self._failed_telegrams.add("DS001") - formatted = format.format(0) + except Exception as err: + raise CustomError(ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) def DS001neu(self): - value = self.config_manager.get_current_selection().route_number + value = self.selection_manager.get_active_selection().route_number format = TELEGRAM_FORMATS["DS001neu"] if isinstance(value, str): value = self.sanitize_ibis_text(value) if value is None: - set_error_and_raise(ErrorCodes.ROUTE_NUMBER_IS_NONE) + raise CustomError(ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться") try: formatted = format.format(value) - except Exception: - set_message("Номер маршруту не виводиться") - self._failed_telegrams.add("DS001neu") - formatted = format.format("0000") + except Exception as err: + raise CustomError(ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) def DS003(self): - trip = self.config_manager.get_current_selection().trip + trip = self.selection_manager.get_active_selection().trip if trip is None: - set_error_and_raise(ErrorCodes.TRIP_INFO_IS_NONE) - return + raise CustomError(ErrorCodes.TRIP_INFO_IS_NONE, "Код напрямку не відправляється") value = trip.point_id format = TELEGRAM_FORMATS["DS003"] if value is None: - set_error_and_raise(ErrorCodes.POINT_ID_IS_NONE) + raise CustomError(ErrorCodes.POINT_ID_IS_NONE, "Код напрямку не відправляється") try: formatted = format.format(int(value)) - except Exception: - set_message("Код напрямку не відправляється") - self._failed_telegrams.add("DS003") - formatted = format.format(0) + except Exception as err: + raise CustomError(ErrorCodes.POINT_ID_VALUE_IS_WRONG, "Код напрямку не відправляється") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) def DS003a(self): - trip = self.config_manager.get_current_selection().trip + trip = self.selection_manager.get_active_selection().trip if trip is None: - set_error_and_raise(ErrorCodes.TRIP_INFO_IS_NONE) - return + raise CustomError( + ErrorCodes.TRIP_INFO_IS_NONE, + "Текст на зовнішньому табло не відображається", + ) value = trip.get_proper_trip_name() if len(value) == 2: if self._system_config.show_start_and_end_stops: @@ -155,27 +153,33 @@ def DS003a(self): format = TELEGRAM_FORMATS["DS003a"] try: formatted = format.format(value[:32]) - except Exception: - set_message("Текст на зовнішньому табло не відображається") - self._failed_telegrams.add("DS003a") - formatted = "zA2" + (" " * 32) + except Exception as err: + raise CustomError( + ErrorCodes.TRIP_NAME_IS_WRONG, + "Текст на зовнішньому табло не відображається", + ) from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) def DS003c(self): if self._system_config.show_info_on_stop_board: - route_number = self.config_manager.get_current_selection().route_number - trip = self.config_manager.get_current_selection().trip + route_number = self.selection_manager.get_active_selection().route_number + trip = self.selection_manager.get_active_selection().trip if trip is None: - set_error_and_raise(ErrorCodes.TRIP_INFO_IS_NONE) - return + raise CustomError( + ErrorCodes.TRIP_INFO_IS_NONE, + "Текст на внутрішньому табло не відображається", + ) format = TELEGRAM_FORMATS["DS003c"] + if route_number is None: + raise CustomError( + ErrorCodes.ROUTE_NUMBER_IS_NONE, + "Текст на внутрішньому табло не відображається", + ) if isinstance(route_number, str): route_number = self.sanitize_ibis_text(route_number) - if route_number is None: - set_error_and_raise(ErrorCodes.ROUTE_NUMBER_IS_NONE) trip_name = trip.get_proper_trip_name() @@ -184,14 +188,20 @@ def DS003c(self): else: trip_name = trip_name[0] + if trip_name is None: + raise CustomError( + ErrorCodes.TRIP_NAME_IS_NONE, + "Текст на внутрішньому табло не відображається", + ) if isinstance(trip_name, str): trip_name = self.sanitize_ibis_text(trip_name) try: formatted = format.format((route_number + " > " + trip_name)[:24]) - except Exception: - set_message("Текст на внутрішньому табло не відображається") - self._failed_telegrams.add("DS003c") - formatted = "zI6" + (" " * 24) + except Exception as err: + raise CustomError( + ErrorCodes.TRIP_NAME_OR_ROUTE_NUMBER_IS_WRONG, + "Текст на внутрішньому табло не відображається", + ) from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) @@ -201,43 +211,44 @@ def DS003c(self): async def send_ibis_telegrams(self): self._running = True while self._running: - current_selection = self.config_manager.get_current_selection() - if current_selection.is_updated: + active_selection = self.selection_manager.get_active_selection() + if active_selection.is_updated: self._failed_telegrams.clear() - current_selection.is_updated = False - if ( - current_selection.route_number is not None - and current_selection.trip is not None - ): + active_selection.is_updated = False + if active_selection.route_number is not None and active_selection.trip is not None: for code in self.telegramTypes: if code in self._failed_telegrams: continue - if ( - code in ("DS001", "DS001neu") - and current_selection.no_line_telegram - ): + if code in ("DS001", "DS001neu") and active_selection.no_line_telegram: continue handler = self.dispatch.get(code) if handler: - handler() + try: + handler() + except CustomError as err: + self._failed_telegrams.add(code) + trigger_message(err.detail, err.error_code) await asyncio.sleep_ms(5) else: - set_error_and_raise(ErrorCodes.UNKNOWN_TELEGRAM) + self._running = False + set_error_and_raise( + ErrorCodes.UNKNOWN_TELEGRAM, + RuntimeError(f"Невідомий тип телеграми: {code}"), + show_message=True, + raise_exception=False, + ) + break await asyncio.sleep(10) - @property - def task(self): - return self._task - def start(self): """Start async loop as a task""" - if not self._task: - self._task = asyncio.create_task(self.send_ibis_telegrams()) + if not self.task: + self.task = asyncio.create_task(self.send_ibis_telegrams()) def stop(self): """Stop async loop""" self._running = False - if self._task: - self._task.cancel() - self._task = None + if self.task: + self.task.cancel() + self.task = None diff --git a/app/routes_management/routes_manager.py b/app/routes_management/routes_manager.py index 4948342..0a9f4ee 100644 --- a/app/routes_management/routes_manager.py +++ b/app/routes_management/routes_manager.py @@ -1,7 +1,10 @@ +import gc import os import ujson as json + from app.error_codes import ErrorCodes +from utils.custom_error import CustomError from utils.error_handler import set_error_and_raise from utils.singleton_decorator import singleton @@ -19,33 +22,48 @@ def load_routes(self) -> None: try: os.stat(ROUTES_PATH) except OSError: - set_error_and_raise(ErrorCodes.ROUTES_FILE_NOT_FOUND) + set_error_and_raise(ErrorCodes.ROUTES_FILE_NOT_FOUND, raise_exception=False) + return try: self._route_list = self.build_route_list() print("Routes was loaded") + gc.collect() return - except Exception: + except (ValueError, RuntimeError): pass try: self.refresh_db(ROUTES_PATH) + except CustomError as err: + self.remove_db() + set_error_and_raise(err.error_code, err, show_message=True, raise_exception=False) + return + + try: self._route_list = self.build_route_list() print("Routes was loaded after refresh db") - except Exception as e: - set_error_and_raise(ErrorCodes.ROUTES_DB_OPEN_FAILED, e) + gc.collect() + except (ValueError, RuntimeError) as err: + set_error_and_raise(ErrorCodes.ROUTES_DB_OPEN_FAILED, err, raise_exception=False) def refresh_db(self, routes_path: str) -> None: """ Args: routes_path: The path to the routes.txt file. """ + self.remove_db() + self.import_routes_from_txt(routes_path) + + def remove_db(self) -> None: try: os.remove(DB_PATH) - except OSError: - pass - - self.import_routes_from_txt(routes_path) + self._route_list = [] + except OSError as err: + if err.args[0] == 2: + self._route_list = [] + else: + raise CustomError(ErrorCodes.ROUTES_DB_DELETE_FAILED, str(err)) from err def append_route( self, @@ -62,8 +80,8 @@ def append_route( rec["note"] = note with open(DB_PATH, "a") as f: f.write(json.dumps(rec) + "\n") - except OSError as e: - set_error_and_raise(ErrorCodes.ROUTES_DB_WRITE_FAILED, e) + except OSError as err: + raise CustomError(ErrorCodes.ROUTES_DB_WRITE_FAILED, str(err)) from err def append_direction( self, @@ -87,8 +105,8 @@ def append_direction( rec["s"] = short_name with open(DB_PATH, "a") as f: f.write(json.dumps(rec) + "\n") - except OSError as e: - set_error_and_raise(ErrorCodes.ROUTES_DB_WRITE_FAILED, e) + except OSError as err: + raise CustomError(ErrorCodes.ROUTES_DB_WRITE_FAILED, str(err)) from err def import_routes_from_txt(self, path_txt): next_route_id = 0 @@ -127,33 +145,45 @@ def import_routes_from_txt(self, path_txt): num_line = num_line[:-1].strip() if not num_line: - set_error_and_raise(ErrorCodes.ROUTES_EMPTY_ROUTE_NUMBER) + raise CustomError( + ErrorCodes.ROUTES_EMPTY_ROUTE_NUMBER, + ErrorCodes.get_message(ErrorCodes.ROUTES_EMPTY_ROUTE_NUMBER) + + "." + + f"Рядок:{line_number}", + ) current_route = num_line current_route_id = next_route_id - self.append_route( - current_route_id, current_route, no_line_telegram, note - ) + self.append_route(current_route_id, current_route, no_line_telegram, note) next_route_id += 1 has_routes = True expecting_route_after_separator = False continue if current_route is None or current_route_id is None: - set_error_and_raise(ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE) + raise CustomError( + ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE, + ErrorCodes.get_message(ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE) + + "." + + f"Рядок:{line_number}", + ) return parts = [p.strip() for p in line.split(",")] if len(parts) not in (3, 4): - set_error_and_raise( - ErrorCodes.ROUTES_DIRECTION_WRONG_PARTS_COUNT + raise CustomError( + ErrorCodes.ROUTES_DIRECTION_WRONG_PARTS_COUNT, + ErrorCodes.get_message(ErrorCodes.DS003_ERROR) + "." + f"Рядок:{line_number}", ) d_id, p_id, full_name_str = parts[0], parts[1], parts[2] if not d_id or not p_id: - set_error_and_raise(ErrorCodes.ROUTES_DIRECTION_EMPTY_ID) + raise CustomError( + ErrorCodes.ROUTES_DIRECTION_EMPTY_ID, + ErrorCodes.get_message(ErrorCodes.ROUTES_DIRECTION_EMPTY_ID) + "." + f"Рядок:{line_number}", + ) full_name = full_name_str.split("^") @@ -161,13 +191,19 @@ def import_routes_from_txt(self, path_txt): if len(parts) == 4: short_name_str = parts[3] if "^" not in short_name_str: - set_error_and_raise( - ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR + raise CustomError( + ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR, + ErrorCodes.get_message(ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR) + + "." + + f"Рядок:{line_number}", ) short_name = short_name_str.split("^") if len(short_name) < 2: - set_error_and_raise( - ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS + raise CustomError( + ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS, + ErrorCodes.get_message(ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS) + + "." + + f"Рядок:{line_number}", ) self.append_direction( @@ -180,35 +216,36 @@ def import_routes_from_txt(self, path_txt): ) if not has_routes: - set_error_and_raise(ErrorCodes.ROUTES_NO_ROUTES_FOUND) + raise CustomError( + ErrorCodes.ROUTES_NO_ROUTES_FOUND, + ErrorCodes.get_message(ErrorCodes.ROUTES_NO_ROUTES_FOUND) + "." + f"Рядок:{line_number}", + ) - except OSError as e: - if e.args[0] == 2: - set_error_and_raise(ErrorCodes.ROUTES_FILE_NOT_FOUND, e) + except OSError as err: + if err.args[0] == 2: + raise CustomError(ErrorCodes.ROUTES_FILE_NOT_FOUND, str(err)) from err else: - set_error_and_raise(ErrorCodes.ROUTES_FILE_OPEN_FAILED, e) + raise CustomError(ErrorCodes.ROUTES_FILE_OPEN_FAILED, str(err)) from err def build_route_list(self): routes_list = [] try: - with open(DB_PATH, "r") as f: + with open(DB_PATH) as f: for line in f: try: rec = json.loads(line) except Exception: continue if rec.get("t") == "route": - routes_list.append( - { - "id": rec.get("id"), - "n": rec.get("n"), - "nlt": rec.get("nlt", False), - "note": rec.get("note"), - } - ) - except OSError as e: - raise RuntimeError(f"Failed to open routes DB: {e}") + routes_list.append({ + "id": rec.get("id"), + "n": rec.get("n"), + "nlt": rec.get("nlt", False), + "note": rec.get("note"), + }) + except OSError as err: + raise RuntimeError(f"Failed to open routes DB: {err}") from err if not routes_list: raise ValueError("Routes DB is empty") @@ -232,21 +269,19 @@ def get_route_by_index(self, index: int): dirs = [] try: - with open(DB_PATH, "r") as f: + with open(DB_PATH) as f: for line in f: try: rec = json.loads(line) except Exception: continue if rec.get("t") == "dir" and rec.get("rid") == route_id: - dirs.append( - { - "trip_id": rec.get("d", ""), - "point_id": rec.get("p", ""), - "full_name": rec.get("f", ""), - "short_name": rec.get("s", None), - } - ) + dirs.append({ + "trip_id": rec.get("d", ""), + "point_id": rec.get("p", ""), + "full_name": rec.get("f", ""), + "short_name": rec.get("s", None), + }) except OSError: pass return { diff --git a/app/selection_management/__init__.py b/app/selection_management/__init__.py new file mode 100644 index 0000000..6fa3e5c --- /dev/null +++ b/app/selection_management/__init__.py @@ -0,0 +1,3 @@ +from .selection_manager import ActiveSelection, SelectionManager, TripInfo + +__all__ = ["SelectionManager", "TripInfo", "ActiveSelection"] diff --git a/app/selection_management/selection_manager.py b/app/selection_management/selection_manager.py new file mode 100644 index 0000000..662d9b7 --- /dev/null +++ b/app/selection_management/selection_manager.py @@ -0,0 +1,135 @@ +import os + +import ujson as json + +from app.config_management.config_info import SystemConfig +from app.error_codes import ErrorCodes +from app.routes_management.routes_manager import RoutesManager +from utils.error_handler import set_error_and_raise +from utils.singleton_decorator import singleton + +SELECTION_PATH = "app/selection_management/selection.json" +TEMP_SELECTION_PATH = "app/selection_management/selection.tmp" + + +class TripInfo: + def __init__( + self, + group_id: str, + point_id: str, + full_name: list[str], + short_name: list[str] | None, + ): + self.group_id = group_id + self.point_id = point_id + self.full_name = full_name + self.short_name = short_name + + @staticmethod + def trip_from_dict(d: dict | None): + if d is not None: + group_id = d.get("trip_id", "") + point_id = d.get("point_id", "") + full_name = d.get("full_name") or [] + short_name = d.get("short_name") or [] + return TripInfo(group_id, point_id, full_name, short_name) + else: + return None + + def get_proper_trip_name(self) -> list[str]: + if SystemConfig().force_short_names: + if self.short_name: + return self.short_name + else: + return self.full_name + else: + return self.full_name + + +class ActiveSelection: + def __init__(self): + self.route_number: str | None = None + self.trip: TripInfo | None = None + self.no_line_telegram: bool = False + self.is_updated: bool = False + + def apply(self, route_number: str | None, trip: dict | None, no_line_telegram: bool = False): + self.route_number = route_number + self.trip = TripInfo.trip_from_dict(trip) + self.no_line_telegram = no_line_telegram + self.is_updated = True + + def reset(self): + self.route_number = None + self.trip = None + self.no_line_telegram = False + self.is_updated = False + + +@singleton +class SelectionManager: + def __init__(self): + self._active_selection = ActiveSelection() + + def get_selection_ids(self): + selection_info = self._load_from_file() + if selection_info is None: + return {"route_id": 0, "trip_id": 0} + return selection_info + + def get_active_selection(self) -> ActiveSelection: + if self._active_selection.route_number is None: + ids = self.get_selection_ids() + self._enrich_from_routes(ids["route_id"], ids["trip_id"]) + return self._active_selection + + def update_selection(self, route_id: int, trip_id: int): + self._save_to_file(route_id, trip_id) + self._enrich_from_routes(route_id, trip_id) + + def reset_selection(self): + self._save_to_file(0, 0) + self._active_selection.reset() + + def _load_from_file(self) -> dict | None: + try: + with open(SELECTION_PATH) as file: + return json.loads(file.read()) + except OSError: + pass + + try: + with open(TEMP_SELECTION_PATH) as file: + return json.loads(file.read()) + except OSError: + return None + + def _save_to_file(self, selected_route_id, selected_trip_id): + try: + rec = { + "route_id": selected_route_id, + "trip_id": selected_trip_id, + } + with open(TEMP_SELECTION_PATH, "w") as file: + file.write(json.dumps(rec)) + file.flush() + + os.sync() + os.rename(TEMP_SELECTION_PATH, SELECTION_PATH) + os.sync() + + except OSError as err: + set_error_and_raise(ErrorCodes.TEMP_SELECTION_WRITE_ERROR, err, show_message=True) + + def _enrich_from_routes(self, route_id: int, trip_id: int) -> None: + route = RoutesManager().get_route_by_index(route_id) + if route and route.get("dirs"): + dirs = route["dirs"] + trip = dirs[trip_id] if trip_id < len(dirs) else None + self._active_selection.apply( + route_number=route["route_number"], + trip=trip, + no_line_telegram=route.get("no_line_telegram", False), + ) + else: + self._active_selection.reset() diff --git a/app/state_management/__init__.py b/app/state_management/__init__.py deleted file mode 100644 index dd564be..0000000 --- a/app/state_management/__init__.py +++ /dev/null @@ -1,3 +0,0 @@ -from .state_manager import StateManager - -__all__ = ["StateManager"] diff --git a/app/state_management/state_manager.py b/app/state_management/state_manager.py deleted file mode 100644 index 42aa921..0000000 --- a/app/state_management/state_manager.py +++ /dev/null @@ -1,54 +0,0 @@ -import os - -import ujson as json -from app.error_codes import ErrorCodes -from utils.error_handler import set_error_and_raise -from utils.singleton_decorator import singleton - -STATE_PATH = "app/state_management/state.json" -TEMP_STATE_PATH = "app/state_management/state.tmp" - - -@singleton -class StateManager: - def __init__(self): - self._state = {} - - def save_state(self, selected_route_id, selected_trip_id): - try: - rec = { - "route_id": selected_route_id, - "trip_id": selected_trip_id, - } - with open(TEMP_STATE_PATH, "w") as file: - file.write(json.dumps(rec)) - file.flush() - - os.sync() - os.rename(TEMP_STATE_PATH, STATE_PATH) - os.sync() - - except OSError as e: - set_error_and_raise(ErrorCodes.TEMP_STATE_WRITE_ERROR, e, True) - - def get_state(self): - state_info = self._load_state() - if state_info is None: - return {"route_id": 0, "trip_id": 0} - return state_info - - def reset_state(self): - self.save_state(0, 0) - - def _load_state(self) -> dict | None: - try: - with open(STATE_PATH, "r") as file: - return json.loads(file.read()) - except OSError: - pass - - try: - with open(TEMP_STATE_PATH, "r") as file: - return json.loads(file.read()) - except OSError: - return None diff --git a/app/web_update/safe_route_decorator.py b/app/web_update/safe_route_decorator.py index 2920d25..1ccbfbc 100644 --- a/app/web_update/safe_route_decorator.py +++ b/app/web_update/safe_route_decorator.py @@ -5,10 +5,8 @@ def outer(fn): async def inner(request, *args, **kwargs): try: return await fn(request, *args, **kwargs) - except Exception as e: - return server._error_response( - "Internal server error: {}".format(e), 500, "text/plain" - ) + except Exception as err: + return server._error_response(f"Internal server error: {err}", 500, "text/plain") return inner diff --git a/app/web_update/web_server.py b/app/web_update/web_server.py index 5e25cf4..4caeaa1 100644 --- a/app/web_update/web_server.py +++ b/app/web_update/web_server.py @@ -4,59 +4,193 @@ import machine import network import uasyncio as asyncio +from microdot import Microdot # type: ignore + from app.error_codes import ErrorCodes from app.routes_management import RoutesManager -from app.state_management import StateManager +from app.selection_management import SelectionManager from app.web_update.safe_route_decorator import safe_route -from microdot import Microdot # type: ignore from utils.error_handler import set_error_and_raise ALLOWED_CHARS = set( - " !\"'+,-./0123456789:<=>?ABCDEFGHIJKLMNOPQRSTUVWXYZ\\_abcdefghijklmnopqrstuvwxyz()ÓóĄąĆćĘꣳŚśŻżЄІЇАБВГДЕЖЗИЙКЛМНОПРСТУФХЦЧШЩЬЮЯабвгдежзийклмнопрстуфхцчшщьюяєії^#|\n\r,+" + " !\"'+,-./0123456789:<=>?ABCDEFGHIJKLMNOPQRSTUVWXYZ\\_abcdefghijklmnopqrstuvwxyz()" + "ÓóĄąĆćĘꣳŚśŻżЄІЇАБВГДЕЖЗИЙКЛМНОПРСТУФХЦЧШЩЬЮЯабвгдежзийклмнопрстуфхцчшщьюяєії^#|\n\r,+" ) -ALLOWED_CONFIG_CHARS = set( - " abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_=.-\n\r" -) +ALLOWED_CONFIG_CHARS = set(" abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789_=.-\n\r") VALID_CONFIG_KEYS = { + "line_telegram", + "destination_number_telegram", + "destination_telegram", "show_start_and_end_stops", "force_short_names", + "stop_board_telegram", "show_info_on_stop_board", + "ap_name", + "ap_password", + "ap_ip", "baudrate", "bits", "parity", "stop", - "line", - "destination_number", - "destination", - "stop_board_telegram", - "ap_name", - "ap_password", - "ap_ip", } -BASE_STYLE = """body{font-family:Arial;margin:0;padding:0;background:#f5f5f5;display:flex;align-items:center;justify-content:center;min-height:100vh}.c{background:#fff;padding:20px;border-radius:8px;box-shadow:0 2px 4px rgba(0,0,0,.1);max-width:400px;width:90%;text-align:center}h1{font-size:1.5em;margin-bottom:1em}""" - -UPLOAD_HTML = ( - """Завантажити

Режим оновлення

Завантажити config.txt:

Завантажити routes.txt:

""" -) - -SUCCESS_HTML = ( - """Успіх
✔

Успішно завантажено

Збережені файли: {files}

Пристрій перезавантажиться...

""" -) +def _get_base_style(): + return """body { + font-family: Arial; + margin: 0; + padding: 0; + background: #f5f5f5; + display: flex; + align-items: center; + justify-content: center; + min-height: 100vh; +} +.c { + background: #fff; + padding: 20px; + border-radius: 8px; + box-shadow: 0 2px 4px rgba(0, 0, 0, 0.1); + max-width: 400px; + width: 90%; + text-align: center; +} +h1 { + font-size: 1.5em; + margin-bottom: 1em; +} +""" + + +def _get_upload_html(): + return ( + """ + + + + + Завантажити + + + +
+

Режим оновлення

+
+

Завантажити config.txt:

+ +

Завантажити routes.txt:

+ + +
+
+ + +""" + ) + + +def _get_success_html(files: str): + return ( + """ + + + + + Успіх + + + +
+
✔
+

Успішно завантажено

+

Збережені файли: """ + + files + + """

+

Пристрій перезавантажиться...

+
+ + +""" + ) + + +def _get_error_html(message: str): + return ( + """ + + + + + Помилка + + + +
+
✘
+

Помилка

+

""" + + message + + """

+ Спробувати ще раз +
+ + +""" + ) -ERROR_HTML = ( - """Помилка
✘

Помилка

{message}

Спробувати ще раз
""" -) TMP_RAW = "/tmp_raw.bin" TMP_CONFIG = "/tmp_config.txt" @@ -77,7 +211,7 @@ def _check_invalid_chars_file(filepath: str, allowed_chars: set) -> list: line_num = 1 char_index = 0 try: - with open(filepath, "r") as f: + with open(filepath) as f: while True: chunk = f.read(128) if not chunk: @@ -93,8 +227,8 @@ def _check_invalid_chars_file(filepath: str, allowed_chars: set) -> list: errors.append("... (ще є помилки)") return errors char_index += 1 - except UnicodeDecodeError as e: - return [f"Невірне кодування файлу (позиція {e.start})"] + except UnicodeDecodeError as err: + return [f"Невірне кодування файлу (позиція {err.start})"] return errors @@ -105,7 +239,7 @@ def _file_is_empty(filepath: str) -> bool: return True except OSError: return True - with open(filepath, "r") as f: + with open(filepath) as f: while True: chunk = f.read(128) if not chunk: @@ -126,7 +260,7 @@ def _check_config_content_file(filepath: str) -> list: found_keys = set() line_num = 0 - with open(filepath, "r") as f: + with open(filepath) as f: while True: line = f.readline() if not line: @@ -167,9 +301,7 @@ def _check_config_content_file(filepath: str) -> list: missing_keys = VALID_CONFIG_KEYS - found_keys if missing_keys: - errors.append( - f"Відсутні обов'язкові параметри: {', '.join(sorted(missing_keys))}" - ) + errors.append(f"Відсутні обов'язкові параметри: {', '.join(sorted(missing_keys))}") return errors @@ -191,7 +323,7 @@ def _check_routes_content_file(filepath: str) -> list: line_num = 0 seen_p_ids = {} - with open(filepath, "r") as f: + with open(filepath) as f: while True: raw_line = f.readline() if not raw_line: @@ -221,9 +353,7 @@ def _check_routes_content_file(filepath: str) -> list: route_num_line = route_num_line[:-1].strip() if not route_num_line: - errors.append( - f"Рядок {line_num}: Порожній номер маршруту після роздільника" - ) + errors.append(f"Рядок {line_num}: Порожній номер маршруту після роздільника") else: current_route = route_num_line has_routes = True @@ -239,15 +369,13 @@ def _check_routes_content_file(filepath: str) -> list: parts = [p.strip() for p in line.split(",")] if len(parts) not in (3, 4): - errors.append( - f"Рядок {line_num}: Очікується 3 або 4 значення через кому, отримано {len(parts)}" - ) + errors.append(f"Рядок {line_num}: Очікується 3 або 4 значення через кому, отримано {len(parts)}") if len(errors) >= 10: errors.append("... (ще є помилки)") break continue - d_id, p_id, full_name_str = parts[0], parts[1], parts[2] + d_id, p_id, _ = parts[0], parts[1], parts[2] if not d_id or not p_id: errors.append(f"Рядок {line_num}: Порожній ID напрямку або точки") @@ -262,18 +390,14 @@ def _check_routes_content_file(filepath: str) -> list: if len(parts) == 4: short_name_str = parts[3] if "^" not in short_name_str: - errors.append( - f"Рядок {line_num}: Коротка назва має містити роздільник '^'" - ) + errors.append(f"Рядок {line_num}: Коротка назва має містити роздільник '^'") if len(errors) >= 10: errors.append("... (ще є помилки)") break if expecting_route: - errors.append( - f"Рядок {expecting_route_line}: Роздільник '|' без номера маршруту після нього" - ) + errors.append(f"Рядок {expecting_route_line}: Роздільник '|' без номера маршруту після нього") if not has_routes and not errors: errors.append("У файлі не знайдено жодного маршруту") @@ -433,6 +557,7 @@ def start(self): if self._running or self._task or self._start_guard: print("Already starting/running.") return + gc.collect() self._start_guard = True try: self._task = asyncio.create_task(self._start_servertask()) @@ -446,14 +571,12 @@ def stop(self): asyncio.create_task(self._stop_servertask()) def _upload_page(self): - return UPLOAD_HTML, 200, {"Content-Type": "text/html; charset=utf-8"} + return _get_upload_html(), 200, {"Content-Type": "text/html; charset=utf-8"} def _register_routes(self): @self._app.errorhandler(413) async def payload_too_large(request): - return self._error_response( - "Файл занадто великий. Максимум: 16KB", code=413 - ) + return self._error_response("Файл занадто великий. Максимум: 16KB", code=413) @safe_route(self) @self._app.route("/") @@ -474,17 +597,17 @@ async def upload(request): try: await _stream_body_to_file(request, TMP_RAW) - except Exception as e: + except Exception as err: _cleanup_tmp() - return self._error_response(f"Помилка при отриманні даних: {e}") + return self._error_response(f"Помилка при отриманні даних: {err}") gc.collect() try: parts = _extract_parts_from_file(TMP_RAW, boundary.encode()) - except Exception as e: + except Exception as err: _cleanup_tmp() - return self._error_response(f"Помилка при обробці файлів: {e}") + return self._error_response(f"Помилка при обробці файлів: {err}") try: os.remove(TMP_RAW) @@ -503,18 +626,14 @@ async def upload(request): if not filename.endswith(".txt"): _cleanup_tmp() - return self._error_response( - f"Дозволені лише .txt файли: '{filename}'" - ) + return self._error_response(f"Дозволені лише .txt файли: '{filename}'") if name == "config_file": if "config" in filename.lower(): errors = _check_config_content_file(tmp_path) if errors: _cleanup_tmp() - return self._error_response( - f"Помилка у config.txt: {'; '.join(errors)}" - ) + return self._error_response(f"Помилка у config.txt: {'; '.join(errors)}") files_to_save["config.txt"] = tmp_path else: _cleanup_tmp() @@ -527,9 +646,7 @@ async def upload(request): errors = _check_routes_content_file(tmp_path) if errors: _cleanup_tmp() - return self._error_response( - f"Помилка у routes.txt: {'; '.join(errors)}" - ) + return self._error_response(f"Помилка у routes.txt: {'; '.join(errors)}") files_to_save["routes.txt"] = tmp_path else: _cleanup_tmp() @@ -554,12 +671,12 @@ async def upload(request): if "routes.txt" in saved_files: try: routes_manager = RoutesManager() - state_manager = StateManager() + selection_manager = SelectionManager() routes_manager.refresh_db("/config/routes.txt") - state_manager.reset_state() + selection_manager.reset_selection() - except Exception as e: - set_error_and_raise(ErrorCodes.REFRESH_ROUTES_DB_ERROR, e, True) + except Exception as err: + set_error_and_raise(ErrorCodes.REFRESH_ROUTES_DB_ERROR, err, show_message=True) asyncio.create_task(self._delayed_reset()) return self._success_response(", ".join(saved_files)) @@ -585,12 +702,10 @@ async def _start_ap(self): print("IP address:", self._ap.ifconfig()[0]) def _success_response(self, files: str): - html = SUCCESS_HTML.replace("{files}", files) - return html, 200, {"Content-Type": "text/html; charset=utf-8"} + return _get_success_html(files), 200, {"Content-Type": "text/html; charset=utf-8"} def _error_response(self, message: str, code: int = 400): - html = ERROR_HTML.replace("{message}", message) - return html, code, {"Content-Type": "text/html; charset=utf-8"} + return _get_error_html(message), code, {"Content-Type": "text/html; charset=utf-8"} async def _start_servertask(self): try: @@ -598,8 +713,8 @@ async def _start_servertask(self): self._running = True print("Starting server...") await self._app.start_server(host=self.host, port=self.port) - except Exception as e: - set_error_and_raise(ErrorCodes.WEB_SERVER_ERROR, e, True) + except Exception as err: + set_error_and_raise(ErrorCodes.WEB_SERVER_ERROR, err, show_message=True) finally: self._running = False print("Server stopped") @@ -612,7 +727,7 @@ async def _stop_servertask(self): print("Access Point stopped.") try: self._app.shutdown() - except Exception as e: - set_error_and_raise(ErrorCodes.WEB_SERVER_SHUTDOWN_ERROR, e, True) + except Exception as err: + set_error_and_raise(ErrorCodes.WEB_SERVER_SHUTDOWN_ERROR, err, show_message=True) self._running = False diff --git a/main.py b/main.py index 05bbd53..f7b6cd7 100644 --- a/main.py +++ b/main.py @@ -1,21 +1,24 @@ +# isort: skip_file import os -from machine import Pin, I2C, UART -import uasyncio as asyncio +import gc import sh1106 # type: ignore +import uasyncio as asyncio import writer # type: ignore +from machine import I2C, UART, Pin from app.gui_management import ( + ErrorState, GuiManager, + InitialState, ScreenConfig, - ScreenStates, ) -from app.routes_management import RoutesManager from app.config_management import ConfigManager -from app.ibis_management import IBISManager from app.error_codes import ErrorCodes +from app.ibis_management import IBISManager +from app.routes_management import RoutesManager from utils.error_handler import set_error_and_raise - +from utils.gui_hooks import trigger_initial try: from config import lang # type: ignore @@ -41,7 +44,7 @@ def check_config_related_files(*paths): return if CONFIG_PATH in missing and ROUTES_PATH in missing: - screen_config.current_screen = ScreenStates.INITIAL_SCREEN + trigger_initial() if __name__ == "__main__": @@ -57,6 +60,9 @@ def check_config_related_files(*paths): writer = writer.Writer(display, lang) + gui_manager = GuiManager(display, writer) + gc.collect() + screen_config = ScreenConfig() config_manager = ConfigManager() @@ -64,25 +70,9 @@ def check_config_related_files(*paths): check_config_related_files(CONFIG_PATH, ROUTES_PATH, CONFIG_EXAMPLE_PATH) - if screen_config.current_screen not in ( - ScreenStates.INITIAL_SCREEN, - ScreenStates.ERROR_SCREEN, - ): - try: - config_manager.load_config(CONFIG_PATH) - except Exception: - set_error_and_raise( - ErrorCodes.CONFIG_FILE_LOAD_ERROR, raise_exception=False - ) - - try: - routes_manager.load_routes() - except Exception: - set_error_and_raise( - ErrorCodes.ROUTES_FILE_LOAD_ERROR, raise_exception=False - ) - - config_manager.get_current_selection().load_from_saved_state() + if not isinstance(gui_manager._state, InitialState) and not isinstance(gui_manager._state, ErrorState): + config_manager.load_config(CONFIG_PATH) + routes_manager.load_routes() btn_down = Pin(2, Pin.IN, Pin.PULL_UP) btn_select = Pin(3, Pin.IN, Pin.PULL_UP) @@ -98,7 +88,7 @@ def check_config_related_files(*paths): max_number_of_characters_in_line, ) - if screen_config.current_screen != ScreenStates.ERROR_SCREEN: + if not isinstance(gui_manager._state, ErrorState): uart = UART( 0, tx=Pin(0), @@ -111,8 +101,6 @@ def check_config_related_files(*paths): ibis_manager = IBISManager(uart, config_manager.get_telegram_types()) - gui_manager = GuiManager(display, writer, screen_config) - async def gui_loop(gui: GuiManager): try: while True: @@ -135,24 +123,34 @@ async def gui_loop(gui: GuiManager): gui.handle_buttons(m, u, d, s) gui.draw_current_screen() await asyncio.sleep_ms(30) - except Exception as e: - print(f"GUI loop error: {e}") + except Exception as err: + set_error_and_raise( + ErrorCodes.MAIN_LOOP_ERROR, + RuntimeError(f"GUI loop error: {err}"), + show_message=True, + raise_exception=False, + ) gui.draw_current_screen() raise async def main_loop(): gui_task = asyncio.create_task(gui_loop(gui_manager)) - if screen_config.current_screen != ScreenStates.ERROR_SCREEN: + if not isinstance(gui_manager._state, ErrorState): ibis_manager.start() try: if ibis_manager.task: - await asyncio.gather(ibis_manager.task, gui_task) + await asyncio.gather(gui_task, ibis_manager.task) else: await gui_task - except Exception as e: - print(f"Main loop error: {e}") + except Exception as err: + set_error_and_raise( + ErrorCodes.MAIN_LOOP_ERROR, + RuntimeError(f"Main loop error: {err}"), + show_message=True, + raise_exception=False, + ) if ibis_manager.task: ibis_manager.stop() await gui_task diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..4c0e21f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,18 @@ +[tool.ruff] +line-length = 120 +target-version = "py311" +exclude = [ + "lib/", + "config/", + "typings/", +] +preview = true + + +[tool.ruff.lint] +select = ["E", "F", "I", "B", "UP"] +ignore = [] + +[tool.ruff.format] +quote-style = "double" +indent-style = "space" diff --git a/requirements-dev.txt b/requirements-dev.txt new file mode 100644 index 0000000..c0a53d2 Binary files /dev/null and b/requirements-dev.txt differ diff --git a/utils/custom_error.py b/utils/custom_error.py new file mode 100644 index 0000000..949ed6c --- /dev/null +++ b/utils/custom_error.py @@ -0,0 +1,5 @@ +class CustomError(Exception): + def __init__(self, error_code: int, detail: str = ""): + self.error_code = error_code + self.detail = detail + super().__init__(detail) diff --git a/utils/error_handler.py b/utils/error_handler.py index 892b828..8401d9f 100644 --- a/utils/error_handler.py +++ b/utils/error_handler.py @@ -1,9 +1,8 @@ from app.error_codes import ErrorCodes +from utils.gui_hooks import trigger_error -def set_error_and_raise( - error_code: int, exception=None, show_message=False, raise_exception=True -): +def set_error_and_raise(error_code: int, exception=None, show_message=False, raise_exception=True): """ Sets the error screen with the code and raises an exception. @@ -11,14 +10,8 @@ def set_error_and_raise( error_code: code from ErrorCodes exception: Optional exception to re-raise """ - from app.gui_management import ScreenConfig, ScreenStates - - screen_config = ScreenConfig() - screen_config.current_screen = ScreenStates.ERROR_SCREEN - screen_config.error_code = error_code - if show_message: - screen_config.message_to_display = str(exception) if exception else None - screen_config.mark_dirty() + message = str(exception) if show_message and exception else None + trigger_error(error_code, message) if not raise_exception: return diff --git a/utils/gui_hooks.py b/utils/gui_hooks.py new file mode 100644 index 0000000..edbafc7 --- /dev/null +++ b/utils/gui_hooks.py @@ -0,0 +1,33 @@ +_on_error = None +_on_message = None +_on_initial = None + + +def register_error_hook(callback): + global _on_error + _on_error = callback + + +def register_message_hook(callback): + global _on_message + _on_message = callback + + +def register_initial_hook(callback): + global _on_initial + _on_initial = callback + + +def trigger_error(error_code, message=None): + if _on_error: + _on_error(error_code, message) + + +def trigger_message(message: str, error_code=None): + if _on_message: + _on_message(message, error_code) + + +def trigger_initial(): + if _on_initial: + _on_initial() diff --git a/utils/message_handler.py b/utils/message_handler.py deleted file mode 100644 index 8dbb7e6..0000000 --- a/utils/message_handler.py +++ /dev/null @@ -1,13 +0,0 @@ -def set_message(message: str): - """ - Sets the message screen with the info for user. User can switch to another state from this state. - - Args: - message: text to display on the message screen - """ - from app.gui_management import ScreenConfig, ScreenStates - - screen_config = ScreenConfig() - screen_config.current_screen = ScreenStates.MESSAGE_SCREEN - screen_config.mark_dirty() - screen_config.message_to_display = message