diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml new file mode 100644 index 0000000..be1c554 --- /dev/null +++ b/.github/workflows/ci.yml @@ -0,0 +1,8 @@ +name: CI +on: pull_request +jobs: + ruff: + runs-on: ubuntu-latest + steps: + - uses: actions/checkout@v4 + - uses: astral-sh/ruff-action@v3 \ No newline at end of file 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/config_manager.py b/app/config_management/config_manager.py index abdec5d..42449f6 100644 --- a/app/config_management/config_manager.py +++ b/app/config_management/config_manager.py @@ -28,40 +28,32 @@ def _convert_value(self, key: str, value: str): if key in {"baudrate", "bits", "parity", "stop"}: try: return int(value) - except ValueError: + except ValueError as err: raise CustomError( ErrorCodes.CONFIG_INVALID_VALUE, 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(): - raise CustomError( - ErrorCodes.CONFIG_FILE_EMPTY, "Config file is empty" - ) + raise CustomError(ErrorCodes.CONFIG_FILE_EMPTY, "Config file is empty") lines = content.splitlines() self._parse_config(lines) print("Config was loaded.") - except CustomError as e: - set_error_and_raise( - e.error_code, e, show_message=True, raise_exception=False - ) - except OSError as e: + 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, raise_exception=False - ) + 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, raise_exception=False - ) + set_error_and_raise(ErrorCodes.CONFIG_IO_ERROR, err, raise_exception=False) def _parse_config(self, lines: list[str]) -> None: for line in lines: @@ -70,9 +62,7 @@ def _parse_config(self, lines: list[str]) -> None: continue if "=" not in line: - raise CustomError( - ErrorCodes.CONFIG_NO_EQUALS_SIGN, f"Missing '=' in line: {line}" - ) + 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): diff --git a/app/gui_management/gui_drawer.py b/app/gui_management/gui_drawer.py index d2e00e4..3c8b476 100644 --- a/app/gui_management/gui_drawer.py +++ b/app/gui_management/gui_drawer.py @@ -66,13 +66,9 @@ def _draw_menu( self._writer.printstring(header_suffix, False) 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), @@ -99,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( @@ -130,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}", @@ -225,9 +217,7 @@ def draw_message_screen(self, message: str, error_code: int | 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: @@ -258,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() @@ -311,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 14b48af..c26f8b4 100644 --- a/app/gui_management/gui_manager.py +++ b/app/gui_management/gui_manager.py @@ -135,9 +135,7 @@ def navigate_down(self, menu_type: RouteMenuState | TripMenuState) -> None: if menu_state.highlighted_item_index < get_number_of_menu_items - 1: menu_state.highlighted_item_index += 1 - def _get_menu_data( - self, menu_type: RouteMenuState | TripMenuState - ) -> RouteMenuData | TripMenuData: + 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): @@ -153,17 +151,12 @@ 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 - ) + 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 - ) + return time.ticks_diff(current_time, self._last_single_button_time) < self._single_button_cooldown def _is_long_pressed( self, @@ -197,9 +190,7 @@ def _is_long_pressed( return False - def handle_buttons( - self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int - ) -> None: + 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]: @@ -208,7 +199,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) diff --git a/app/gui_management/states/__init__.py b/app/gui_management/states/__init__.py index 0ca021d..551b988 100644 --- a/app/gui_management/states/__init__.py +++ b/app/gui_management/states/__init__.py @@ -17,5 +17,5 @@ "UpdateState", "ErrorState", "MessageState", - "InitialState" + "InitialState", ] diff --git a/app/gui_management/states/error_state.py b/app/gui_management/states/error_state.py index f7f6801..f65856c 100644 --- a/app/gui_management/states/error_state.py +++ b/app/gui_management/states/error_state.py @@ -11,10 +11,9 @@ def draw_current_screen(self): ctx._message_to_display, ) - def handle_buttons( - self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int - ): + 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 @@ -27,4 +26,4 @@ def handle_buttons( ctx.transition_to(UpdateState(ErrorState())) ctx.mark_dirty() return - return \ No newline at end of file + return diff --git a/app/gui_management/states/initial_state.py b/app/gui_management/states/initial_state.py index 6fa1658..26c4fb5 100644 --- a/app/gui_management/states/initial_state.py +++ b/app/gui_management/states/initial_state.py @@ -8,10 +8,9 @@ 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 - ): + 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 @@ -24,4 +23,4 @@ def handle_buttons( ctx._web_update_server.ensure_started() ctx.mark_dirty() return - return \ No newline at end of file + return diff --git a/app/gui_management/states/message_state.py b/app/gui_management/states/message_state.py index 9c82563..e4e58ba 100644 --- a/app/gui_management/states/message_state.py +++ b/app/gui_management/states/message_state.py @@ -11,10 +11,9 @@ def draw_current_screen(self): 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 - ): + 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 @@ -25,4 +24,4 @@ def handle_buttons( ctx.transition_to(StatusState()) ctx.mark_dirty() ctx._last_single_button_time = current_time - return \ No newline at end of file + return diff --git a/app/gui_management/states/route_menu_state.py b/app/gui_management/states/route_menu_state.py index a067e36..157b445 100644 --- a/app/gui_management/states/route_menu_state.py +++ b/app/gui_management/states/route_menu_state.py @@ -7,9 +7,7 @@ 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 - ) + 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() @@ -20,11 +18,10 @@ def draw_current_screen(self): number_of_menu_items, ) - def handle_buttons( - self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int - ): + 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 diff --git a/app/gui_management/states/settings_state.py b/app/gui_management/states/settings_state.py index c2c5800..b200ca9 100644 --- a/app/gui_management/states/settings_state.py +++ b/app/gui_management/states/settings_state.py @@ -8,11 +8,10 @@ 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 - ): + 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 diff --git a/app/gui_management/states/state.py b/app/gui_management/states/state.py index b4af809..e156499 100644 --- a/app/gui_management/states/state.py +++ b/app/gui_management/states/state.py @@ -7,14 +7,8 @@ def context(self): 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 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" - ) + 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 index bcdfa58..7c0b61b 100644 --- a/app/gui_management/states/status_state.py +++ b/app/gui_management/states/status_state.py @@ -6,12 +6,8 @@ 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"] + 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: @@ -23,12 +19,11 @@ def draw_current_screen(self): 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 - ): + 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 diff --git a/app/gui_management/states/trip_menu_state.py b/app/gui_management/states/trip_menu_state.py index e126e4e..a4eb9b4 100644 --- a/app/gui_management/states/trip_menu_state.py +++ b/app/gui_management/states/trip_menu_state.py @@ -6,9 +6,7 @@ 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 - ) + 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() @@ -20,11 +18,10 @@ def draw_current_screen(self): f"M:{route['route_number']}", ) - def handle_buttons( - self, btn_menu: int, btn_up: int, btn_down: int, btn_select: int - ): + 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 @@ -50,15 +47,9 @@ def handle_buttons( 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 - ) - route = ctx._routes_manager.get_route_by_index( - ctx._route_menu_data.selected_item_index - ) + 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 + route = ctx._routes_manager.get_route_by_index(ctx._route_menu_data.selected_item_index) ctx._config_manager.update_current_selection( route["route_number"], route["dirs"][ctx._trip_menu_data.selected_item_index], diff --git a/app/gui_management/states/update_state.py b/app/gui_management/states/update_state.py index 63e5fc0..4e39d5e 100644 --- a/app/gui_management/states/update_state.py +++ b/app/gui_management/states/update_state.py @@ -9,14 +9,11 @@ def __init__(self, return_state=None): 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 - ) + 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 - ): + 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 diff --git a/app/ibis_management/ibis_manager.py b/app/ibis_management/ibis_manager.py index 740cd79..f7e89b8 100644 --- a/app/ibis_management/ibis_manager.py +++ b/app/ibis_management/ibis_manager.py @@ -89,15 +89,11 @@ def DS001(self): value = self.config_manager.get_current_selection().route_number format = TELEGRAM_FORMATS["DS001"] if value is None: - raise CustomError( - ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться" - ) + raise CustomError(ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться") try: formatted = format.format(int(value)) - except Exception: - raise CustomError( - ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться" - ) + except Exception as err: + raise CustomError(ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) @@ -108,15 +104,11 @@ def DS001neu(self): if isinstance(value, str): value = self.sanitize_ibis_text(value) if value is None: - raise CustomError( - ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться" - ) + raise CustomError(ErrorCodes.ROUTE_NUMBER_IS_NONE, "Номер маршруту не виводиться") try: formatted = format.format(value) - except Exception: - raise CustomError( - ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться" - ) + except Exception as err: + raise CustomError(ErrorCodes.ROUTE_VALUE_IS_WRONG, "Номер маршруту не виводиться") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) @@ -124,22 +116,16 @@ def DS001neu(self): def DS003(self): trip = self.config_manager.get_current_selection().trip if trip is None: - raise CustomError( - ErrorCodes.TRIP_INFO_IS_NONE, "Код напрямку не відправляється" - ) + raise CustomError(ErrorCodes.TRIP_INFO_IS_NONE, "Код напрямку не відправляється") value = trip.point_id format = TELEGRAM_FORMATS["DS003"] if value is None: - raise CustomError( - ErrorCodes.POINT_ID_IS_NONE, "Код напрямку не відправляється" - ) + raise CustomError(ErrorCodes.POINT_ID_IS_NONE, "Код напрямку не відправляється") try: formatted = format.format(int(value)) - except Exception: - raise CustomError( - ErrorCodes.POINT_ID_VALUE_IS_WRONG, "Код напрямку не відправляється" - ) + except Exception as err: + raise CustomError(ErrorCodes.POINT_ID_VALUE_IS_WRONG, "Код напрямку не відправляється") from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) @@ -166,11 +152,11 @@ def DS003a(self): format = TELEGRAM_FORMATS["DS003a"] try: formatted = format.format(value[:32]) - except Exception: + except Exception as err: raise CustomError( ErrorCodes.TRIP_NAME_IS_WRONG, "Текст на зовнішньому табло не відображається", - ) + ) from err packet = self.create_ibis_packet(formatted) self.uart.write(packet) @@ -210,11 +196,11 @@ def DS003c(self): trip_name = self.sanitize_ibis_text(trip_name) try: formatted = format.format((route_number + " > " + trip_name)[:24]) - except Exception: + 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) @@ -228,26 +214,20 @@ async def send_ibis_telegrams(self): if current_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 - ): + if current_selection.route_number is not None and current_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 current_selection.no_line_telegram: continue handler = self.dispatch.get(code) if handler: try: handler() - except CustomError as e: + except CustomError as err: self._failed_telegrams.add(code) - trigger_message(e.detail, e.error_code) + trigger_message(err.detail, err.error_code) await asyncio.sleep_ms(5) else: self._running = False diff --git a/app/routes_management/routes_manager.py b/app/routes_management/routes_manager.py index d9d729b..7a11c42 100644 --- a/app/routes_management/routes_manager.py +++ b/app/routes_management/routes_manager.py @@ -33,20 +33,16 @@ def load_routes(self) -> None: try: self.refresh_db(ROUTES_PATH) - except CustomError as e: + except CustomError as err: self.remove_db() - set_error_and_raise( - e.error_code, e, show_message=True, raise_exception=False - ) + 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 (ValueError, RuntimeError) as e: - set_error_and_raise( - ErrorCodes.ROUTES_DB_OPEN_FAILED, e, raise_exception=False - ) + 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: """ @@ -60,11 +56,11 @@ def remove_db(self) -> None: try: os.remove(DB_PATH) self._route_list = [] - except OSError as e: - if e.args[0] == 2: + except OSError as err: + if err.args[0] == 2: self._route_list = [] else: - raise CustomError(ErrorCodes.ROUTES_DB_DELETE_FAILED, str(e)) + raise CustomError(ErrorCodes.ROUTES_DB_DELETE_FAILED, str(err)) from err def append_route( self, @@ -81,8 +77,8 @@ def append_route( rec["note"] = note with open(DB_PATH, "a") as f: f.write(json.dumps(rec) + "\n") - except OSError as e: - raise CustomError(ErrorCodes.ROUTES_DB_WRITE_FAILED, str(e)) + except OSError as err: + raise CustomError(ErrorCodes.ROUTES_DB_WRITE_FAILED, str(err)) from err def append_direction( self, @@ -106,8 +102,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: - raise CustomError(ErrorCodes.ROUTES_DB_WRITE_FAILED, str(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 @@ -148,18 +144,14 @@ def import_routes_from_txt(self, path_txt): if not num_line: raise CustomError( ErrorCodes.ROUTES_EMPTY_ROUTE_NUMBER, - ErrorCodes.get_message( - 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 @@ -168,9 +160,7 @@ def import_routes_from_txt(self, path_txt): if current_route is None or current_route_id is None: raise CustomError( ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE, - ErrorCodes.get_message( - ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE - ) + ErrorCodes.get_message(ErrorCodes.ROUTES_DIRECTION_WITHOUT_ROUTE) + "." + f"Рядок:{line_number}", ) @@ -181,9 +171,7 @@ def import_routes_from_txt(self, path_txt): if len(parts) not in (3, 4): raise CustomError( ErrorCodes.ROUTES_DIRECTION_WRONG_PARTS_COUNT, - ErrorCodes.get_message(ErrorCodes.DS003_ERROR) - + "." - + f"Рядок:{line_number}", + ErrorCodes.get_message(ErrorCodes.DS003_ERROR) + "." + f"Рядок:{line_number}", ) d_id, p_id, full_name_str = parts[0], parts[1], parts[2] @@ -191,9 +179,7 @@ def import_routes_from_txt(self, path_txt): if not d_id or not p_id: raise CustomError( ErrorCodes.ROUTES_DIRECTION_EMPTY_ID, - ErrorCodes.get_message(ErrorCodes.ROUTES_DIRECTION_EMPTY_ID) - + "." - + f"Рядок:{line_number}", + ErrorCodes.get_message(ErrorCodes.ROUTES_DIRECTION_EMPTY_ID) + "." + f"Рядок:{line_number}", ) full_name = full_name_str.split("^") @@ -204,9 +190,7 @@ def import_routes_from_txt(self, path_txt): if "^" not in short_name_str: raise CustomError( ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR, - ErrorCodes.get_message( - ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR - ) + ErrorCodes.get_message(ErrorCodes.ROUTES_SHORT_NAME_NO_SEPARATOR) + "." + f"Рядок:{line_number}", ) @@ -214,9 +198,7 @@ def import_routes_from_txt(self, path_txt): if len(short_name) < 2: raise CustomError( ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS, - ErrorCodes.get_message( - ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS - ) + ErrorCodes.get_message(ErrorCodes.ROUTES_SHORT_NAME_TOO_FEW_PARTS) + "." + f"Рядок:{line_number}", ) @@ -233,22 +215,20 @@ def import_routes_from_txt(self, path_txt): if not has_routes: raise CustomError( ErrorCodes.ROUTES_NO_ROUTES_FOUND, - ErrorCodes.get_message(ErrorCodes.ROUTES_NO_ROUTES_FOUND) - + "." - + f"Рядок:{line_number}", + ErrorCodes.get_message(ErrorCodes.ROUTES_NO_ROUTES_FOUND) + "." + f"Рядок:{line_number}", ) - except OSError as e: - if e.args[0] == 2: - raise CustomError(ErrorCodes.ROUTES_FILE_NOT_FOUND, str(e)) + except OSError as err: + if err.args[0] == 2: + raise CustomError(ErrorCodes.ROUTES_FILE_NOT_FOUND, str(err)) from err else: - raise CustomError(ErrorCodes.ROUTES_FILE_OPEN_FAILED, str(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) @@ -263,8 +243,8 @@ def build_route_list(self): "note": rec.get("note"), } ) - except OSError as e: - raise RuntimeError(f"Failed to open routes DB: {e}") + 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") @@ -288,7 +268,7 @@ 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) diff --git a/app/selection_management/selection_manager.py b/app/selection_management/selection_manager.py index ef186e6..208f27e 100644 --- a/app/selection_management/selection_manager.py +++ b/app/selection_management/selection_manager.py @@ -29,10 +29,8 @@ def save_selection(self, selected_route_id, selected_trip_id): os.rename(TEMP_SELECTION_PATH, SELECTION_PATH) os.sync() - except OSError as e: - set_error_and_raise( - ErrorCodes.TEMP_SELECTION_WRITE_ERROR, e, show_message=True - ) + except OSError as err: + set_error_and_raise(ErrorCodes.TEMP_SELECTION_WRITE_ERROR, err, show_message=True) def get_selection(self): selection_info = self._load_selection() @@ -45,13 +43,13 @@ def reset_selection(self): def _load_selection(self) -> dict | None: try: - with open(SELECTION_PATH, "r") as file: + with open(SELECTION_PATH) as file: return json.loads(file.read()) except OSError: pass try: - with open(TEMP_SELECTION_PATH, "r") as file: + with open(TEMP_SELECTION_PATH) 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 41a4266..58f9861 100644 --- a/app/web_update/web_server.py +++ b/app/web_update/web_server.py @@ -13,12 +13,11 @@ 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 = { @@ -39,26 +38,150 @@ } -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}""" +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, 0.1); + max-width: 400px; + width: 90%; + text-align: center; +} +h1 { + font-size: 1.5em; + margin-bottom: 1em; +} +""" UPLOAD_HTML = ( - """Завантажити

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

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

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

""" + + """p { + margin: 0.5em 0 0.2em; + text-align: left; + } + input[type="file"] { + width: 100%; + } + input[type="submit"] { + margin-top: 1em; + width: 100%; + padding: 0.7em; + font-size: 1em; + background: #4caf50; + color: #fff; + border: none; + border-radius: 4px; + cursor: pointer; + } + + + +
+

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

+
+

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

+ +

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

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

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

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

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

""" + + """.ok { + color: #4caf50; + font-size: 3em; + } + p { + margin: 1em 0; + } + + + +
+
✔
+

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

+

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

+

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

+
+ + +""" ) + ERROR_HTML = ( - """Помилка
✘

Помилка

{message}

Спробувати ще раз
""" + + """.err { + color: #f44336; + font-size: 3em; + } + p { + margin: 1em 0; + } + a { + display: inline-block; + margin-top: 1em; + padding: 0.5em 1em; + background: #4caf50; + color: #fff; + text-decoration: none; + border-radius: 4px; + } + + + +
+
✘
+

Помилка

+

{message}

+ Спробувати ще раз +
+ + +""" ) + TMP_RAW = "/tmp_raw.bin" TMP_CONFIG = "/tmp_config.txt" TMP_ROUTES = "/tmp_routes.txt" @@ -78,7 +201,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: @@ -94,8 +217,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 @@ -106,7 +229,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: @@ -127,7 +250,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: @@ -168,9 +291,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 @@ -192,7 +313,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: @@ -222,9 +343,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 @@ -240,15 +359,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 напрямку або точки") @@ -263,18 +380,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("У файлі не знайдено жодного маршруту") @@ -452,9 +565,7 @@ def _upload_page(self): 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("/") @@ -475,17 +586,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) @@ -504,18 +615,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() @@ -528,9 +635,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() @@ -559,10 +664,8 @@ async def upload(request): routes_manager.refresh_db("/config/routes.txt") selection_manager.reset_selection() - except Exception as e: - set_error_and_raise( - ErrorCodes.REFRESH_ROUTES_DB_ERROR, e, show_message=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)) @@ -601,8 +704,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, show_message=True) + except Exception as err: + set_error_and_raise(ErrorCodes.WEB_SERVER_ERROR, err, show_message=True) finally: self._running = False print("Server stopped") @@ -615,9 +718,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, show_message=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 e6c6885..7168a3f 100644 --- a/main.py +++ b/main.py @@ -67,9 +67,7 @@ def check_config_related_files(*paths): check_config_related_files(CONFIG_PATH, ROUTES_PATH, CONFIG_EXAMPLE_PATH) - if not isinstance(gui_manager._state, InitialState) and not isinstance( - gui_manager._state, ErrorState - ): + if not isinstance(gui_manager._state, InitialState) and not isinstance(gui_manager._state, ErrorState): config_manager.load_config(CONFIG_PATH) routes_manager.load_routes() @@ -124,8 +122,8 @@ 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: + print(f"GUI loop error: {err}") gui.draw_current_screen() raise @@ -140,8 +138,8 @@ async def main_loop(): 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: + print(f"Main loop error: {err}") if ibis_manager.task: ibis_manager.stop() await gui_task diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9a59818 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,17 @@ +[tool.ruff] +line-length = 120 +target-version = "py311" +exclude = [ + "lib/", + "config/", + "typings/", +] + + +[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/error_handler.py b/utils/error_handler.py index d5399a5..8401d9f 100644 --- a/utils/error_handler.py +++ b/utils/error_handler.py @@ -2,9 +2,7 @@ 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.