#!/usr/bin/env python3 # -*- coding: utf-8 -*- from typing import Any, Dict, Optional, List from dataclasses import dataclass, field from enum import Enum, auto import logging from datetime import datetime from .events import event_bus, Event, EventTypes class ConnectionState(Enum): """Состояния подключения.""" DISCONNECTED = auto() CONNECTING = auto() CONNECTED = auto() ERROR = auto() class TransferState(Enum): """Состояния передачи данных.""" IDLE = auto() PREPARING = auto() TRANSFERRING = auto() COMPLETED = auto() ERROR = auto() @dataclass class DeviceInfo: """Информация об устройстве.""" version: Optional[str] = None model: Optional[str] = None hostname: Optional[str] = None interfaces: List[Dict[str, Any]] = field(default_factory=list) last_update: Optional[datetime] = None @dataclass class ConfigInfo: """Информация о файле конфигурации.""" name: str path: str size: int modified: datetime created: datetime is_watched: bool = False @dataclass class ApplicationState: """Состояние приложения.""" connection_state: ConnectionState = ConnectionState.DISCONNECTED transfer_state: TransferState = TransferState.IDLE device_info: DeviceInfo = field(default_factory=DeviceInfo) configs: Dict[str, ConfigInfo] = field(default_factory=dict) current_config: Optional[str] = None transfer_progress: float = 0.0 status_message: str = "" error_message: Optional[str] = None is_tftp_server_running: bool = False custom_data: Dict[str, Any] = field(default_factory=dict) class StateManager: """Менеджер состояния приложения.""" def __init__(self): self._state = ApplicationState() self._logger = logging.getLogger(__name__) self._setup_event_handlers() def _setup_event_handlers(self) -> None: """Настройка обработчиков событий.""" event_bus.subscribe(EventTypes.CONNECTION_ESTABLISHED, self._handle_connection_established) event_bus.subscribe(EventTypes.CONNECTION_LOST, self._handle_connection_lost) event_bus.subscribe(EventTypes.CONNECTION_ERROR, self._handle_connection_error) event_bus.subscribe(EventTypes.TRANSFER_STARTED, self._handle_transfer_started) event_bus.subscribe(EventTypes.TRANSFER_PROGRESS, self._handle_transfer_progress) event_bus.subscribe(EventTypes.TRANSFER_COMPLETED, self._handle_transfer_completed) event_bus.subscribe(EventTypes.TRANSFER_ERROR, self._handle_transfer_error) event_bus.subscribe(EventTypes.TFTP_SERVER_STARTED, self._handle_tftp_server_started) event_bus.subscribe(EventTypes.TFTP_SERVER_STOPPED, self._handle_tftp_server_stopped) # Подписка на события конфигурации event_bus.subscribe(EventTypes.CONFIG_CREATED, self._handle_config_created) event_bus.subscribe(EventTypes.CONFIG_MODIFIED, self._handle_config_modified) event_bus.subscribe(EventTypes.CONFIG_DELETED, self._handle_config_deleted) def _handle_connection_established(self, event: Event) -> None: self._state.connection_state = ConnectionState.CONNECTED self._state.error_message = None self._update_status("Подключение установлено") def _handle_connection_lost(self, event: Event) -> None: self._state.connection_state = ConnectionState.DISCONNECTED self._update_status("Подключение потеряно") def _handle_connection_error(self, event: Event) -> None: self._state.connection_state = ConnectionState.ERROR self._state.error_message = str(event.data) self._update_status(f"Ошибка подключения: {event.data}") def _handle_transfer_started(self, event: Event) -> None: self._state.transfer_state = TransferState.TRANSFERRING self._state.transfer_progress = 0.0 self._update_status("Передача данных начата") def _handle_transfer_progress(self, event: Event) -> None: self._state.transfer_progress = float(event.data) self._update_status(f"Прогресс передачи: {self._state.transfer_progress:.1f}%") def _handle_transfer_completed(self, event: Event) -> None: self._state.transfer_state = TransferState.COMPLETED self._state.transfer_progress = 100.0 self._update_status("Передача данных завершена") def _handle_transfer_error(self, event: Event) -> None: self._state.transfer_state = TransferState.ERROR self._state.error_message = str(event.data) self._update_status(f"Ошибка передачи: {event.data}") def _handle_tftp_server_started(self, event: Event) -> None: self._state.is_tftp_server_running = True self._update_status("TFTP сервер запущен") def _handle_tftp_server_stopped(self, event: Event) -> None: self._state.is_tftp_server_running = False self._update_status("TFTP сервер остановлен") def _handle_config_created(self, event: Event) -> None: """ Обработка создания конфигурации. Args: event: Событие создания конфигурации """ try: self.update_config_info(event.data) self._update_status(f"Конфигурация создана: {event.data['name']}") except Exception as e: self._logger.error(f"Ошибка обработки создания конфигурации: {e}") def _handle_config_modified(self, event: Event) -> None: """ Обработка изменения конфигурации. Args: event: Событие изменения конфигурации """ try: self.update_config_info(event.data) self._update_status(f"Конфигурация изменена: {event.data['name']}") except Exception as e: self._logger.error(f"Ошибка обработки изменения конфигурации: {e}") def _handle_config_deleted(self, event: Event) -> None: """ Обработка удаления конфигурации. Args: event: Событие удаления конфигурации """ try: config_name = event.data['name'] self.remove_config_info(config_name) self._update_status(f"Конфигурация удалена: {config_name}") except Exception as e: self._logger.error(f"Ошибка обработки удаления конфигурации: {e}") def _update_status(self, message: str) -> None: """Обновление статусного сообщения.""" self._state.status_message = message event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, message)) def update_device_info(self, info: Dict[str, Any]) -> None: """ Обновление информации об устройстве. Args: info: Словарь с информацией об устройстве """ try: self._state.device_info = DeviceInfo( version=info.get("version"), model=info.get("model"), hostname=info.get("hostname"), interfaces=info.get("interfaces", []), last_update=datetime.now() ) # Обновляем состояние подключения self._state.connection_state = ConnectionState.CONNECTED # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": f"Подключено к {self._state.device_info.hostname}" })) self._logger.info("Информация об устройстве обновлена") except Exception as e: self._logger.error(f"Ошибка обновления информации об устройстве: {e}") self._state.connection_state = ConnectionState.ERROR self._state.error_message = str(e) def clear_device_info(self) -> None: """Очистка информации об устройстве.""" self._state.device_info = DeviceInfo() self._state.connection_state = ConnectionState.DISCONNECTED self._state.error_message = None # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": "Отключено от устройства" })) self._logger.info("Информация об устройстве очищена") def update_config_info(self, info: Dict[str, Any]) -> None: """ Обновление информации о файле конфигурации. Args: info: Словарь с информацией о файле """ try: config = ConfigInfo( name=info["name"], path=info["path"], size=info["size"], modified=info["modified"], created=info["created"], is_watched=info.get("is_watched", False) ) self._state.configs[config.name] = config # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": f"Конфигурация обновлена: {config.name}" })) self._logger.debug(f"Обновлена информация о конфигурации: {config.name}") except Exception as e: self._logger.error(f"Ошибка обновления информации о конфигурации: {e}") def remove_config_info(self, config_name: str) -> None: """ Удаление информации о файле конфигурации. Args: config_name: Имя файла конфигурации """ if config_name in self._state.configs: self._state.configs.pop(config_name) # Если удаляется текущая конфигурация, очищаем её if self._state.current_config == config_name: self._state.current_config = None # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": f"Конфигурация удалена: {config_name}" })) self._logger.debug(f"Удалена информация о конфигурации: {config_name}") def set_current_config(self, config_name: Optional[str]) -> None: """ Установка текущей конфигурации. Args: config_name: Имя файла конфигурации """ if config_name is None or config_name in self._state.configs: self._state.current_config = config_name if config_name: self._logger.debug(f"Установлена текущая конфигурация: {config_name}") else: self._logger.debug("Текущая конфигурация очищена") def update_transfer_state(self, state: TransferState, progress: float = 0.0, error: Optional[str] = None) -> None: """ Обновление состояния передачи. Args: state: Новое состояние progress: Прогресс передачи error: Сообщение об ошибке """ self._state.transfer_state = state self._state.transfer_progress = progress self._state.error_message = error # Формируем сообщение о состоянии if state == TransferState.IDLE: message = "Готов к передаче" elif state == TransferState.PREPARING: message = "Подготовка к передаче..." elif state == TransferState.TRANSFERRING: message = f"Передача данных: {progress:.1f}%" elif state == TransferState.COMPLETED: message = "Передача завершена" elif state == TransferState.ERROR: message = f"Ошибка передачи: {error}" else: message = "Неизвестное состояние" # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": message })) self._logger.debug(f"Состояние передачи обновлено: {state.name}") def set_tftp_server_state(self, is_running: bool) -> None: """ Установка состояния TFTP сервера. Args: is_running: Флаг работы сервера """ self._state.is_tftp_server_running = is_running # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": "TFTP сервер запущен" if is_running else "TFTP сервер остановлен" })) self._logger.debug(f"Состояние TFTP сервера: {'запущен' if is_running else 'остановлен'}") def set_status_message(self, message: str) -> None: """ Установка сообщения о состоянии. Args: message: Сообщение """ self._state.status_message = message # Уведомляем об изменении состояния event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": message })) def set_error_message(self, error: Optional[str]) -> None: """ Установка сообщения об ошибке. Args: error: Сообщение об ошибке """ self._state.error_message = error if error: # Уведомляем об ошибке event_bus.publish(Event(EventTypes.UI_STATUS_CHANGED, { "message": f"Ошибка: {error}" })) def get_device_info(self) -> DeviceInfo: """ Получение информации об устройстве. Returns: DeviceInfo: Информация об устройстве """ return self._state.device_info def get_config_info(self, config_name: str) -> Optional[ConfigInfo]: """ Получение информации о конфигурации. Args: config_name: Имя файла конфигурации Returns: Optional[ConfigInfo]: Информация о конфигурации или None """ return self._state.configs.get(config_name) def get_configs(self) -> Dict[str, ConfigInfo]: """ Получение списка всех конфигураций. Returns: Dict[str, ConfigInfo]: Словарь с информацией о конфигурациях """ return self._state.configs.copy() def get_current_config(self) -> Optional[str]: """ Получение текущей конфигурации. Returns: Optional[str]: Имя текущей конфигурации или None """ return self._state.current_config @property def state(self) -> ApplicationState: """Получение текущего состояния.""" return self._state def set_custom_data(self, key: str, value: Any) -> None: """Установка пользовательских данных.""" self._state.custom_data[key] = value def get_custom_data(self, key: str, default: Any = None) -> Any: """Получение пользовательских данных.""" return self._state.custom_data.get(key, default) # Создаем глобальный экземпляр менеджера состояния state_manager = StateManager()