From 2b04a2fca821d09becfde7e984954e2ae3db232c Mon Sep 17 00:00:00 2001 From: stu2not Date: Mon, 15 Jun 2026 08:52:30 +0800 Subject: [PATCH] =?UTF-8?q?=E6=94=AF=E6=8C=81=E8=AE=BE=E5=A4=87=E5=AE=9A?= =?UTF-8?q?=E6=97=B6=E5=AE=9A=E4=BD=8D=E4=B8=8A=E6=8A=A5=E4=B8=8EOTA?= =?UTF-8?q?=E9=80=9A=E9=81=93=E9=80=89=E6=8B=A9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- database/talkingq_shared_schema.sql | 47 ++++ talkingq-url/banban/dao/device.py | 3 + talkingq-url/banban/dao/device_setting.py | 21 +- talkingq-url/banban/routers/devices.py | 9 + talkingq-url/banban/service/device.py | 20 +- talkingq-url/banban/service/device_setting.py | 66 +++++- talkingq-url/banban/service/ota_firmware.py | 113 ++++++++++ talkingq-url/database/init_db.py | 86 ++++++++ talkingq-url/database/models.py | 46 ++++ talkingq-url/handlers/mqtt_handler.py | 46 +++- talkingq-url/mysql/init/01-init.sql | 44 ++++ talkingq-url/mysql/init/02-init.sql | 47 ++++ .../services/device_update_manager.py | 57 ++++- .../tests/test_mqtt_device_info_report.py | 111 ++++++++++ .../tests/test_ota_channel_selection.py | 201 ++++++++++++++++++ 15 files changed, 893 insertions(+), 24 deletions(-) create mode 100644 talkingq-url/banban/service/ota_firmware.py create mode 100644 talkingq-url/tests/test_mqtt_device_info_report.py create mode 100644 talkingq-url/tests/test_ota_channel_selection.py diff --git a/database/talkingq_shared_schema.sql b/database/talkingq_shared_schema.sql index 17e72c4..4b65b40 100644 --- a/database/talkingq_shared_schema.sql +++ b/database/talkingq_shared_schema.sql @@ -151,6 +151,11 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( `serial_number` VARCHAR(64) NOT NULL, `mac_address` VARCHAR(512) NULL, `firmware_version` VARCHAR(64) NOT NULL, + `ota_channel` VARCHAR(32) NULL, + `target_version` VARCHAR(64) NULL, + `firmware_url` TEXT NULL, + `firmware_id` INT NULL, + `source` VARCHAR(32) NULL, `update_status` VARCHAR(32) NOT NULL DEFAULT 'success', `progress` FLOAT NULL DEFAULT 0.0, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -159,6 +164,38 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS `device_ota_channels` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `name` VARCHAR(64) NOT NULL, + `device_type` TINYINT NULL, + `enabled` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_channels_key` (`channel_key`), + KEY `idx_device_ota_channels_device_type` (`device_type`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_ota_firmwares` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `version` VARCHAR(64) NOT NULL, + `firmware_url` TEXT NOT NULL, + `checksum` VARCHAR(128) NULL, + `file_size` BIGINT NULL, + `release_status` VARCHAR(32) NOT NULL DEFAULT 'draft', + `release_channel` VARCHAR(32) NOT NULL DEFAULT 'stable', + `is_latest` TINYINT(1) NOT NULL DEFAULT 0, + `notes` TEXT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `published_at` DATETIME NULL, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_firmware_channel_version` (`channel_key`, `version`), + KEY `idx_device_ota_firmware_latest` (`channel_key`, `release_status`, `is_latest`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS `system_config` ( `id` INT NOT NULL AUTO_INCREMENT, `config_key` VARCHAR(128) NOT NULL, @@ -180,6 +217,13 @@ VALUES ('update_firmware_url', 'https://example.com/firmware/latest.bin', '固件更新URL') ON DUPLICATE KEY UPDATE `config_value` = `config_value`; +INSERT INTO `device_ota_channels` (`channel_key`, `name`, `device_type`, `enabled`) +VALUES + ('legacy', 'Legacy devices', NULL, 1), + ('no_gps', 'NFC without GPS', 1, 1), + ('gps', 'NFC with GPS', 2, 1) +ON DUPLICATE KEY UPDATE `channel_key` = `channel_key`; + INSERT INTO `roles` (`role_key`, `name`, `description`, `content`, `default_language`, `enabled`) VALUES ( 'assistant', @@ -435,6 +479,9 @@ CREATE TABLE IF NOT EXISTS `device_settings` ( `power` TINYINT UNSIGNED NULL, `signal` TINYINT UNSIGNED NULL, `version` VARCHAR(64) NULL, + `imei` VARCHAR(64) NULL, + `device_type` TINYINT NULL, + `ota_channel` VARCHAR(32) NULL, `disable_weekdays` VARCHAR(32) NULL, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, diff --git a/talkingq-url/banban/dao/device.py b/talkingq-url/banban/dao/device.py index 5951e28..1e34653 100644 --- a/talkingq-url/banban/dao/device.py +++ b/talkingq-url/banban/dao/device.py @@ -144,6 +144,9 @@ class DeviceDAO(BaseDAO): ds.volume, ds.`signal` AS signal_strength, ds.`version` AS version, + ds.imei, + ds.device_type, + ds.ota_channel, ds.updated_at AS settings_updated_at, cl.coord_type, cl.lat, diff --git a/talkingq-url/banban/dao/device_setting.py b/talkingq-url/banban/dao/device_setting.py index 327cf12..0beafef 100644 --- a/talkingq-url/banban/dao/device_setting.py +++ b/talkingq-url/banban/dao/device_setting.py @@ -23,16 +23,19 @@ class DeviceSettingDAO(BaseDAO): signal: Optional[int] = None, version: Optional[str] = None, power: Optional[int] = None, + imei: Optional[str] = None, + device_type: Optional[int] = None, + ota_channel: Optional[str] = None, ) -> int: result = await self.execute( """ INSERT INTO device_settings ( device_id, sleep_mode, manual_sleep_mode, schedule_suppressed_until, disable_time_start, disable_time_end, - timezone, volume, brightness, disable_weekdays, `signal`, `version`, power + timezone, volume, brightness, disable_weekdays, `signal`, `version`, power, imei, device_type, ota_channel ) VALUES ( :device_id, :sleep_mode, :manual_sleep_mode, :schedule_suppressed_until, :disable_time_start, :disable_time_end, - :timezone, :volume, :brightness, :disable_weekdays, :signal, :version, :power + :timezone, :volume, :brightness, :disable_weekdays, :signal, :version, :power, :imei, :device_type, :ota_channel ) """, { @@ -49,6 +52,9 @@ class DeviceSettingDAO(BaseDAO): "signal": signal, "version": version, "power": power, + "imei": imei, + "device_type": device_type, + "ota_channel": ota_channel, }, ) await self.commit() @@ -86,6 +92,9 @@ class DeviceSettingDAO(BaseDAO): signal: Optional[int] = None, version: Optional[str] = None, power: Optional[int] = None, + imei: Optional[str] = None, + device_type: Optional[int] = None, + ota_channel: Optional[str] = None, ) -> None: await self.execute( """ @@ -105,7 +114,10 @@ class DeviceSettingDAO(BaseDAO): disable_weekdays = COALESCE(:disable_weekdays, disable_weekdays), `signal` = COALESCE(:signal, `signal`), `version` = COALESCE(:version, `version`), - power = COALESCE(:power, power) + power = COALESCE(:power, power), + imei = COALESCE(:imei, imei), + device_type = COALESCE(:device_type, device_type), + ota_channel = COALESCE(:ota_channel, ota_channel) WHERE device_id = :device_id """, { @@ -123,6 +135,9 @@ class DeviceSettingDAO(BaseDAO): "signal": signal, "version": version, "power": power, + "imei": imei, + "device_type": device_type, + "ota_channel": ota_channel, }, ) await self.commit() diff --git a/talkingq-url/banban/routers/devices.py b/talkingq-url/banban/routers/devices.py index a732909..f36bfd3 100644 --- a/talkingq-url/banban/routers/devices.py +++ b/talkingq-url/banban/routers/devices.py @@ -84,6 +84,9 @@ class DeviceStatusResponse(BaseModel): volume: int | None = None signal: int | None = None version: str | None = None + imei: str | None = None + device_type: int | None = None + ota_channel: str | None = None settings_updated_at: datetime | None = None coord_type: str | None = None lat: float | None = None @@ -182,6 +185,9 @@ class DeviceFirmwareStatusResponse(BaseModel): device_id: str current_version: str | None = None latest_version: str | None = None + ota_channel: str | None = None + firmware_id: int | None = None + firmware_url: str | None = None update_available: bool can_update: bool update_status: str @@ -306,6 +312,9 @@ def _row_to_device_status_response(row: Mapping) -> DeviceStatusResponse: volume=row["volume"], signal=row["signal_strength"], version=row["version"], + imei=row.get("imei"), + device_type=row.get("device_type"), + ota_channel=row.get("ota_channel"), settings_updated_at=row["settings_updated_at"], coord_type=row["coord_type"], lat=float(row["lat"]) if row["lat"] is not None else None, diff --git a/talkingq-url/banban/service/device.py b/talkingq-url/banban/service/device.py index 43b00b2..fabefed 100644 --- a/talkingq-url/banban/service/device.py +++ b/talkingq-url/banban/service/device.py @@ -7,8 +7,8 @@ from fastapi import HTTPException from banban.dao.device import DeviceDAO from banban.service.device_setting import device_setting_service +from banban.service.ota_firmware import ota_firmware_service from services.device_update_manager import device_firmware_update_manager -from services.system_config_manager import system_config_manager DEVICE_COMMAND_SLEEP_BLOCKED_DETAIL = "设备不在线或在休眠中" @@ -310,10 +310,9 @@ class DeviceService(DatabaseServiceBase): status_row = await self.get_device_status(device_id=device_id, user_id=user_id) current_version = status_row.get("version") - latest_version_config = await system_config_manager.get_config("latest_firmware_version") - firmware_url_config = await system_config_manager.get_config("update_firmware_url") - latest_version = latest_version_config.config_value if latest_version_config else None - firmware_url = firmware_url_config.config_value if firmware_url_config else None + firmware_selection = await ota_firmware_service.select_latest_for_device(status_row) + latest_version = firmware_selection.version if firmware_selection else None + firmware_url = firmware_selection.firmware_url if firmware_selection else None update_available = self._compare_versions(current_version, latest_version) update_record = await device_firmware_update_manager.get_firmware_update_dict(device_id) @@ -332,6 +331,9 @@ class DeviceService(DatabaseServiceBase): "target_version": target_version, "updated_at": updated_at, "can_update": bool(update_available and latest_version and firmware_url), + "ota_channel": firmware_selection.channel_key if firmware_selection else None, + "firmware_id": firmware_selection.firmware_id if firmware_selection else None, + "firmware_url": firmware_url, } async def start_firmware_update( @@ -348,8 +350,7 @@ class DeviceService(DatabaseServiceBase): if not latest_version: raise HTTPException(status_code=404, detail="latest firmware version not configured") - firmware_url_config = await system_config_manager.get_config("update_firmware_url") - firmware_url = firmware_url_config.config_value if firmware_url_config else None + firmware_url = firmware_status.get("firmware_url") if not firmware_url: raise HTTPException(status_code=404, detail="firmware url not configured") @@ -364,6 +365,11 @@ class DeviceService(DatabaseServiceBase): firmware_version=latest_version, update_status="sent", progress=0.0, + ota_channel=firmware_status.get("ota_channel"), + target_version=latest_version, + firmware_url=firmware_url, + firmware_id=firmware_status.get("firmware_id"), + source="manual", ) msg_id = await service.send_ota_command(device_id, firmware_url, latest_version) diff --git a/talkingq-url/banban/service/device_setting.py b/talkingq-url/banban/service/device_setting.py index f6d1391..06c7228 100644 --- a/talkingq-url/banban/service/device_setting.py +++ b/talkingq-url/banban/service/device_setting.py @@ -7,6 +7,28 @@ from services.database_service_base import DatabaseServiceBase from banban.dao.device_setting import DeviceSettingDAO +def infer_ota_channel_from_device_type(device_type: int | None) -> str | None: + if device_type == 2: + return "gps" + if device_type == 1: + return "no_gps" + if device_type == 0: + return "legacy" + return None + + +def normalize_device_type(device_type: object) -> int | None: + if device_type is None or device_type == "": + return None + try: + value = int(device_type) + except (TypeError, ValueError): + return None + if value in (0, 1, 2): + return value + return None + + class DeviceSettingService(DatabaseServiceBase): def __init__(self): super().__init__(service_name="device_setting_service") @@ -26,6 +48,9 @@ class DeviceSettingService(DatabaseServiceBase): disable_weekdays: Optional[str] = None, signal_strength: Optional[int] = None, version_str: Optional[str] = None, + imei: Optional[str] = None, + device_type: Optional[int] = None, + ota_channel: Optional[str] = None, ) -> int: db_session = await self.get_session() try: @@ -44,6 +69,9 @@ class DeviceSettingService(DatabaseServiceBase): disable_weekdays=disable_weekdays, signal=signal_strength, version=version_str, + imei=imei, + device_type=device_type, + ota_channel=ota_channel, ) finally: await db_session.close() @@ -72,6 +100,9 @@ class DeviceSettingService(DatabaseServiceBase): disable_weekdays: Optional[str] = None, signal_strength: Optional[int] = None, version_str: Optional[str] = None, + imei: Optional[str] = None, + device_type: Optional[int] = None, + ota_channel: Optional[str] = None, ) -> None: db_session = await self.get_session() try: @@ -91,6 +122,9 @@ class DeviceSettingService(DatabaseServiceBase): power=power, signal=signal_strength, version=version_str, + imei=imei, + device_type=device_type, + ota_channel=ota_channel, ) finally: await db_session.close() @@ -110,13 +144,39 @@ class DeviceSettingService(DatabaseServiceBase): volume: Optional[int], signal_strength: Optional[int], version_str: Optional[str], - ) -> None: + imei: Optional[str] = None, + device_type: Optional[int] = None, + ota_channel: Optional[str] = None, + ) -> None: try: + normalized_device_type = normalize_device_type(device_type) current_row = await self.get_setting_by_device_id(device_id=device_id) if current_row: - await self.update_setting(device_id=device_id, power=power, volume=volume, signal_strength=signal_strength, version_str=version_str) + resolved_ota_channel = ota_channel + if not resolved_ota_channel and not current_row.get("ota_channel"): + resolved_ota_channel = infer_ota_channel_from_device_type(normalized_device_type) + await self.update_setting( + device_id=device_id, + power=power, + volume=volume, + signal_strength=signal_strength, + version_str=version_str, + imei=imei, + device_type=normalized_device_type, + ota_channel=resolved_ota_channel, + ) else: - await self.create_setting(device_id=device_id, power=power, volume=volume, signal_strength=signal_strength, version_str=version_str) + resolved_ota_channel = ota_channel or infer_ota_channel_from_device_type(normalized_device_type) + await self.create_setting( + device_id=device_id, + power=power, + volume=volume, + signal_strength=signal_strength, + version_str=version_str, + imei=imei, + device_type=normalized_device_type, + ota_channel=resolved_ota_channel, + ) finally: pass diff --git a/talkingq-url/banban/service/ota_firmware.py b/talkingq-url/banban/service/ota_firmware.py new file mode 100644 index 0000000..e6210ee --- /dev/null +++ b/talkingq-url/banban/service/ota_firmware.py @@ -0,0 +1,113 @@ +from collections.abc import Mapping +from dataclasses import dataclass +from typing import Any + +from sqlalchemy import text + +from services.database_service_base import DatabaseServiceBase +from services.system_config_manager import system_config_manager + + +LEGACY_OTA_CHANNEL = "legacy" + + +@dataclass(frozen=True) +class OtaFirmwareSelection: + channel_key: str + version: str + firmware_url: str + firmware_id: int | None = None + source: str = "channel" + + +class OtaFirmwareService(DatabaseServiceBase): + def __init__(self): + super().__init__(service_name="ota_firmware_service") + + def infer_channel_from_device_type(self, device_type: Any) -> str | None: + if device_type is None or device_type == "": + return None + try: + value = int(device_type) + except (TypeError, ValueError): + return None + if value == 2: + return "gps" + if value == 1: + return "no_gps" + if value == 0: + return LEGACY_OTA_CHANNEL + return None + + async def resolve_channel_for_status(self, status_row: Mapping[str, Any]) -> str | None: + configured_channel = str(status_row.get("ota_channel") or "").strip() + if configured_channel: + return configured_channel + return self.infer_channel_from_device_type(status_row.get("device_type")) + + async def get_latest_firmware_for_channel(self, channel_key: str) -> OtaFirmwareSelection | None: + db_session = await self.get_session() + try: + result = await db_session.execute( + text( + """ + SELECT + f.id, + f.channel_key, + f.version, + f.firmware_url + FROM device_ota_firmwares AS f + JOIN device_ota_channels AS c + ON c.channel_key = f.channel_key + WHERE f.channel_key = :channel_key + AND f.release_status = 'published' + AND f.is_latest = 1 + AND c.enabled = 1 + ORDER BY f.published_at DESC, f.updated_at DESC, f.id DESC + LIMIT 1 + """ + ), + {"channel_key": channel_key}, + ) + row = result.mappings().first() + if not row: + return None + return OtaFirmwareSelection( + firmware_id=int(row["id"]), + channel_key=str(row["channel_key"]), + version=str(row["version"]), + firmware_url=str(row["firmware_url"]), + source="channel", + ) + finally: + await db_session.close() + + async def get_legacy_firmware(self) -> OtaFirmwareSelection | None: + latest_version_config = await system_config_manager.get_config("latest_firmware_version") + firmware_url_config = await system_config_manager.get_config("update_firmware_url") + latest_version = latest_version_config.config_value if latest_version_config else None + firmware_url = firmware_url_config.config_value if firmware_url_config else None + if not latest_version or not firmware_url: + return None + return OtaFirmwareSelection( + channel_key=LEGACY_OTA_CHANNEL, + version=str(latest_version), + firmware_url=str(firmware_url), + source="legacy_config", + ) + + async def select_latest_for_device(self, status_row: Mapping[str, Any]) -> OtaFirmwareSelection | None: + channel_key = await self.resolve_channel_for_status(status_row) + if not channel_key: + return None + + selection = await self.get_latest_firmware_for_channel(channel_key) + if selection: + return selection + + if channel_key == LEGACY_OTA_CHANNEL: + return await self.get_legacy_firmware() + return None + + +ota_firmware_service = OtaFirmwareService() diff --git a/talkingq-url/database/init_db.py b/talkingq-url/database/init_db.py index bbfa32b..b44bcbd 100644 --- a/talkingq-url/database/init_db.py +++ b/talkingq-url/database/init_db.py @@ -94,6 +94,89 @@ async def _ensure_device_alarm_location_columns(conn) -> None: ) +async def _ensure_device_settings_metadata_columns(conn) -> None: + await _ensure_columns( + conn, + table_name="device_settings", + columns=[ + ("imei", "imei VARCHAR(64) NULL AFTER `version`"), + ("device_type", "device_type TINYINT NULL AFTER imei"), + ("ota_channel", "ota_channel VARCHAR(32) NULL AFTER device_type"), + ], + ) + + +async def _ensure_device_firmware_update_ota_columns(conn) -> None: + await _ensure_columns( + conn, + table_name="device_firmware_update", + columns=[ + ("ota_channel", "ota_channel VARCHAR(32) NULL AFTER firmware_version"), + ("target_version", "target_version VARCHAR(64) NULL AFTER ota_channel"), + ("firmware_url", "firmware_url TEXT NULL AFTER target_version"), + ("firmware_id", "firmware_id INT NULL AFTER firmware_url"), + ("source", "source VARCHAR(32) NULL AFTER firmware_id"), + ], + ) + + +async def _ensure_device_ota_tables(conn) -> None: + await conn.execute( + text( + """ + CREATE TABLE IF NOT EXISTS device_ota_channels ( + id INT NOT NULL AUTO_INCREMENT, + channel_key VARCHAR(32) NOT NULL, + name VARCHAR(64) NOT NULL, + device_type TINYINT NULL, + enabled TINYINT(1) NOT NULL DEFAULT 1, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uq_device_ota_channels_key (channel_key), + KEY idx_device_ota_channels_device_type (device_type) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + """ + ) + ) + await conn.execute( + text( + """ + CREATE TABLE IF NOT EXISTS device_ota_firmwares ( + id INT NOT NULL AUTO_INCREMENT, + channel_key VARCHAR(32) NOT NULL, + version VARCHAR(64) NOT NULL, + firmware_url TEXT NOT NULL, + checksum VARCHAR(128) NULL, + file_size BIGINT NULL, + release_status VARCHAR(32) NOT NULL DEFAULT 'draft', + release_channel VARCHAR(32) NOT NULL DEFAULT 'stable', + is_latest TINYINT(1) NOT NULL DEFAULT 0, + notes TEXT NULL, + created_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + published_at DATETIME NULL, + updated_at DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (id), + UNIQUE KEY uq_device_ota_firmware_channel_version (channel_key, version), + KEY idx_device_ota_firmware_latest (channel_key, release_status, is_latest) + ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci + """ + ) + ) + await conn.execute( + text( + """ + INSERT INTO device_ota_channels (channel_key, name, device_type, enabled) + VALUES + ('legacy', 'Legacy devices', NULL, 1), + ('no_gps', 'NFC without GPS', 1, 1), + ('gps', 'NFC with GPS', 2, 1) + ON DUPLICATE KEY UPDATE channel_key = channel_key + """ + ) + ) + + async def _ensure_child_location_address_columns(conn) -> None: await _ensure_columns( conn, @@ -283,6 +366,9 @@ async def init_db(): await conn.run_sync(Base.metadata.create_all) await _ensure_manual_sleep_mode_column(conn) await _ensure_schedule_suppressed_until_column(conn) + await _ensure_device_settings_metadata_columns(conn) + await _ensure_device_firmware_update_ota_columns(conn) + await _ensure_device_ota_tables(conn) await _ensure_device_alarm_location_columns(conn) await _ensure_child_location_address_columns(conn) await _ensure_bind_session_card_columns(conn) diff --git a/talkingq-url/database/models.py b/talkingq-url/database/models.py index 219d7b1..59fe8ea 100644 --- a/talkingq-url/database/models.py +++ b/talkingq-url/database/models.py @@ -128,6 +128,11 @@ class DeviceFirmwareUpdate(Base): serial_number = Column(String(64), nullable=False, default="") mac_address = Column(String(512), nullable=True) firmware_version = Column(String(64), nullable=False) + ota_channel = Column(String(32), nullable=True) + target_version = Column(String(64), nullable=True) + firmware_url = Column(Text, nullable=True) + firmware_id = Column(Integer, nullable=True) + source = Column(String(32), nullable=True) update_status = Column(String(32), nullable=False, default="success") # 更新状态，如 updating/success/failed created_at = Column(DateTime, nullable=False, server_default=func.now()) updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) @@ -137,6 +142,44 @@ class DeviceFirmwareUpdate(Base): {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} ) + +class DeviceOtaChannel(Base): + __tablename__ = "device_ota_channels" + id = Column(Integer, primary_key=True, autoincrement=True) + channel_key = Column(String(32), unique=True, nullable=False, index=True) + name = Column(String(64), nullable=False) + device_type = Column(Integer, nullable=True, index=True) + enabled = Column(Boolean, nullable=False, default=True) + created_at = Column(DateTime, nullable=False, server_default=func.now()) + updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) + + __table_args__ = ( + {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} + ) + + +class DeviceOtaFirmware(Base): + __tablename__ = "device_ota_firmwares" + id = Column(Integer, primary_key=True, autoincrement=True) + channel_key = Column(String(32), nullable=False, index=True) + version = Column(String(64), nullable=False) + firmware_url = Column(Text, nullable=False) + checksum = Column(String(128), nullable=True) + file_size = Column(BigInteger, nullable=True) + release_status = Column(String(32), nullable=False, default="draft", index=True) + release_channel = Column(String(32), nullable=False, default="stable") + is_latest = Column(Boolean, nullable=False, default=False, index=True) + notes = Column(Text, nullable=True) + created_at = Column(DateTime, nullable=False, server_default=func.now()) + published_at = Column(DateTime, nullable=True) + updated_at = Column(DateTime, nullable=False, server_default=func.now(), onupdate=func.now()) + + __table_args__ = ( + UniqueConstraint("channel_key", "version", name="uq_device_ota_firmware_channel_version"), + Index("idx_device_ota_firmware_latest", "channel_key", "release_status", "is_latest"), + {'mysql_charset': 'utf8mb4', 'mysql_collate': 'utf8mb4_unicode_ci'} + ) + class SystemConfig(Base): __tablename__ = "system_config" id = Column(Integer, primary_key=True, autoincrement=True) @@ -325,6 +368,9 @@ class DeviceSetting(Base): power: Mapped[Optional[int]] = mapped_column(Integer) signal: Mapped[Optional[int]] = mapped_column("signal", Integer) version: Mapped[Optional[str]] = mapped_column("version", String(64)) + imei: Mapped[Optional[str]] = mapped_column(String(64)) + device_type: Mapped[Optional[int]] = mapped_column(Integer) + ota_channel: Mapped[Optional[str]] = mapped_column(String(32)) disable_weekdays: Mapped[Optional[str]] = mapped_column(String(32)) created_at: Mapped[Optional[datetime]] = mapped_column(DateTime, server_default=text("CURRENT_TIMESTAMP")) updated_at: Mapped[Optional[datetime]] = mapped_column( diff --git a/talkingq-url/handlers/mqtt_handler.py b/talkingq-url/handlers/mqtt_handler.py index deabbc3..b0aa940 100644 --- a/talkingq-url/handlers/mqtt_handler.py +++ b/talkingq-url/handlers/mqtt_handler.py @@ -158,7 +158,18 @@ class TalkingQMQTTService: signal = data.get("signal") version = data.get("version") voice = data.get("voice") - logger.info(device_id, "", f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, 版本号={version}, 音量={voice}") + imei = data.get("imei") + device_type = data.get("device_type") + latitude = data.get("latitude") + longitude = data.get("longitude") + logger.info( + device_id, + "", + ( + f"[设备信息] 设备 {d_id} 信息: 电量={power}, 信号强度={signal}, " + f"版本号={version}, 音量={voice}, IMEI={imei}, 设备类型={device_type}" + ), + ) await self._schedule_persistence( device_id, @@ -169,8 +180,27 @@ class TalkingQMQTTService: signal_strength=data.get("signal"), version_str=data.get("version"), volume=data.get("voice"), + imei=str(imei).strip() if imei is not None and str(imei).strip() else None, + device_type=device_type, ), ) + if latitude is not None and longitude is not None: + await self._schedule_persistence( + device_id, + "device_info_location", + location_service.report_mqtt_device_location( + device_id=device_id, + latitude=latitude, + longitude=longitude, + coord_type=data.get("coord_type"), + accuracy_m=data.get("accuracy_m"), + altitude_m=data.get("altitude_m"), + speed_mps=data.get("speed_mps"), + heading_deg=data.get("heading_deg"), + source=data.get("source", 0), + battery_pct=data.get("battery_pct") or data.get("power"), + ), + ) # await self._sync_desired_volume(device_id, current_volume) await self._publish(f"device/{device_id}/event_resp", {"msg_id": "000", "status": "success"}) @@ -272,13 +302,13 @@ class TalkingQMQTTService: await self._schedule_persistence( device_id, - "ota", - device_firmware_update_manager.update_firmware_update( - device_id=device_id, - firmware_version=data.get("version") or data.get("new_version") or "unknown", - update_status="accepted", - progress=progress_value if progress_value is not None else 0.0, - ), + "ota", + device_firmware_update_manager.update_firmware_update( + device_id=device_id, + firmware_version=data.get("target_version") or data.get("version") or data.get("new_version") or "unknown", + update_status="accepted", + progress=progress_value if progress_value is not None else 0.0, + ), ) elif status == "updating": await self._schedule_persistence( diff --git a/talkingq-url/mysql/init/01-init.sql b/talkingq-url/mysql/init/01-init.sql index f6bfa9c..ab364f1 100644 --- a/talkingq-url/mysql/init/01-init.sql +++ b/talkingq-url/mysql/init/01-init.sql @@ -116,6 +116,11 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( `serial_number` VARCHAR(64) NOT NULL, `mac_address` VARCHAR(512) NULL, `firmware_version` VARCHAR(64) NOT NULL, + `ota_channel` VARCHAR(32) NULL, + `target_version` VARCHAR(64) NULL, + `firmware_url` TEXT NULL, + `firmware_id` INT NULL, + `source` VARCHAR(32) NULL, `update_status` VARCHAR(32) NOT NULL DEFAULT 'success', `progress` FLOAT NULL DEFAULT 0.0, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -124,6 +129,38 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS `device_ota_channels` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `name` VARCHAR(64) NOT NULL, + `device_type` TINYINT NULL, + `enabled` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_channels_key` (`channel_key`), + KEY `idx_device_ota_channels_device_type` (`device_type`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_ota_firmwares` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `version` VARCHAR(64) NOT NULL, + `firmware_url` TEXT NOT NULL, + `checksum` VARCHAR(128) NULL, + `file_size` BIGINT NULL, + `release_status` VARCHAR(32) NOT NULL DEFAULT 'draft', + `release_channel` VARCHAR(32) NOT NULL DEFAULT 'stable', + `is_latest` TINYINT(1) NOT NULL DEFAULT 0, + `notes` TEXT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `published_at` DATETIME NULL, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_firmware_channel_version` (`channel_key`, `version`), + KEY `idx_device_ota_firmware_latest` (`channel_key`, `release_status`, `is_latest`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + -- 创建system_config表 CREATE TABLE IF NOT EXISTS `system_config` ( `id` INT NOT NULL AUTO_INCREMENT, @@ -148,6 +185,13 @@ VALUES ('update_firmware_url', 'https://example.com/firmware/latest.bin', '固件更新URL') ON DUPLICATE KEY UPDATE `config_value`=`config_value`; +INSERT INTO `device_ota_channels` (`channel_key`, `name`, `device_type`, `enabled`) +VALUES +('legacy', 'Legacy devices', NULL, 1), +('no_gps', 'NFC without GPS', 1, 1), +('gps', 'NFC with GPS', 2, 1) +ON DUPLICATE KEY UPDATE `channel_key`=`channel_key`; + -- 添加默认角色 INSERT INTO `roles` (`role_key`, `name`, `description`, `content`, `default_language`, `enabled`) VALUES ( diff --git a/talkingq-url/mysql/init/02-init.sql b/talkingq-url/mysql/init/02-init.sql index 17e72c4..4b65b40 100644 --- a/talkingq-url/mysql/init/02-init.sql +++ b/talkingq-url/mysql/init/02-init.sql @@ -151,6 +151,11 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( `serial_number` VARCHAR(64) NOT NULL, `mac_address` VARCHAR(512) NULL, `firmware_version` VARCHAR(64) NOT NULL, + `ota_channel` VARCHAR(32) NULL, + `target_version` VARCHAR(64) NULL, + `firmware_url` TEXT NULL, + `firmware_id` INT NULL, + `source` VARCHAR(32) NULL, `update_status` VARCHAR(32) NOT NULL DEFAULT 'success', `progress` FLOAT NULL DEFAULT 0.0, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, @@ -159,6 +164,38 @@ CREATE TABLE IF NOT EXISTS `device_firmware_update` ( UNIQUE INDEX `device_id_UNIQUE` (`device_id` ASC) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; +CREATE TABLE IF NOT EXISTS `device_ota_channels` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `name` VARCHAR(64) NOT NULL, + `device_type` TINYINT NULL, + `enabled` TINYINT(1) NOT NULL DEFAULT 1, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_channels_key` (`channel_key`), + KEY `idx_device_ota_channels_device_type` (`device_type`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + +CREATE TABLE IF NOT EXISTS `device_ota_firmwares` ( + `id` INT NOT NULL AUTO_INCREMENT, + `channel_key` VARCHAR(32) NOT NULL, + `version` VARCHAR(64) NOT NULL, + `firmware_url` TEXT NOT NULL, + `checksum` VARCHAR(128) NULL, + `file_size` BIGINT NULL, + `release_status` VARCHAR(32) NOT NULL DEFAULT 'draft', + `release_channel` VARCHAR(32) NOT NULL DEFAULT 'stable', + `is_latest` TINYINT(1) NOT NULL DEFAULT 0, + `notes` TEXT NULL, + `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + `published_at` DATETIME NULL, + `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (`id`), + UNIQUE KEY `uq_device_ota_firmware_channel_version` (`channel_key`, `version`), + KEY `idx_device_ota_firmware_latest` (`channel_key`, `release_status`, `is_latest`) +) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_unicode_ci; + CREATE TABLE IF NOT EXISTS `system_config` ( `id` INT NOT NULL AUTO_INCREMENT, `config_key` VARCHAR(128) NOT NULL, @@ -180,6 +217,13 @@ VALUES ('update_firmware_url', 'https://example.com/firmware/latest.bin', '固件更新URL') ON DUPLICATE KEY UPDATE `config_value` = `config_value`; +INSERT INTO `device_ota_channels` (`channel_key`, `name`, `device_type`, `enabled`) +VALUES + ('legacy', 'Legacy devices', NULL, 1), + ('no_gps', 'NFC without GPS', 1, 1), + ('gps', 'NFC with GPS', 2, 1) +ON DUPLICATE KEY UPDATE `channel_key` = `channel_key`; + INSERT INTO `roles` (`role_key`, `name`, `description`, `content`, `default_language`, `enabled`) VALUES ( 'assistant', @@ -435,6 +479,9 @@ CREATE TABLE IF NOT EXISTS `device_settings` ( `power` TINYINT UNSIGNED NULL, `signal` TINYINT UNSIGNED NULL, `version` VARCHAR(64) NULL, + `imei` VARCHAR(64) NULL, + `device_type` TINYINT NULL, + `ota_channel` VARCHAR(32) NULL, `disable_weekdays` VARCHAR(32) NULL, `created_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, `updated_at` DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, diff --git a/talkingq-url/services/device_update_manager.py b/talkingq-url/services/device_update_manager.py index d88b511..9c8856a 100644 --- a/talkingq-url/services/device_update_manager.py +++ b/talkingq-url/services/device_update_manager.py @@ -33,13 +33,30 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): return { "device_id": row.device_id, "firmware_version": row.firmware_version, + "ota_channel": getattr(row, "ota_channel", None), + "target_version": getattr(row, "target_version", None), + "firmware_url": getattr(row, "firmware_url", None), + "firmware_id": getattr(row, "firmware_id", None), + "source": getattr(row, "source", None), "update_status": row.update_status, "progress": row.progress, "created_at": row.created_at, "updated_at": row.updated_at, } - async def create_firmware_update(self, device_id: str, firmware_version: str, update_status: str = "success", progress: float = 0.0): + async def create_firmware_update( + self, + device_id: str, + firmware_version: str, + update_status: str = "success", + progress: float = 0.0, + *, + ota_channel: str | None = None, + target_version: str | None = None, + firmware_url: str | None = None, + firmware_id: int | None = None, + source: str | None = None, + ): """新增固件更新记录""" await self._init_database() db_session = await self.get_session() @@ -55,7 +72,12 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): .values( firmware_version=firmware_version, update_status=update_status, - progress=progress + progress=progress, + ota_channel=ota_channel, + target_version=target_version, + firmware_url=firmware_url, + firmware_id=firmware_id, + source=source, ) ) await db_session.execute(update_stmt) @@ -66,7 +88,12 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): serial_number=serial_number, firmware_version=firmware_version, update_status=update_status, - progress=progress + progress=progress, + ota_channel=ota_channel, + target_version=target_version, + firmware_url=firmware_url, + firmware_id=firmware_id, + source=source, ) await db_session.execute(insert_stmt) @@ -75,6 +102,11 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): self._update_cache(device_id, { "device_id": device_id, "firmware_version": firmware_version, + "ota_channel": ota_channel, + "target_version": target_version, + "firmware_url": firmware_url, + "firmware_id": firmware_id, + "source": source, "update_status": update_status, "progress": progress }) @@ -113,6 +145,12 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): firmware_version: str, update_status: str, progress: Optional[float] = None, + *, + ota_channel: str | None = None, + target_version: str | None = None, + firmware_url: str | None = None, + firmware_id: int | None = None, + source: str | None = None, ) -> bool: """更新设备固件信息（支持部分字段更新）""" await self._init_database() @@ -124,6 +162,14 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): } if progress is not None: update_values["progress"] = progress + optional_values = { + "ota_channel": ota_channel, + "target_version": target_version, + "firmware_url": firmware_url, + "firmware_id": firmware_id, + "source": source, + } + update_values.update({key: value for key, value in optional_values.items() if value is not None}) query = select(DeviceFirmwareUpdate).where(DeviceFirmwareUpdate.device_id == device_id) result = await db_session.execute(query) @@ -144,6 +190,11 @@ class DeviceFirmwareUpdateManager(DatabaseServiceBase): firmware_version=firmware_version, update_status=update_status, progress=progress if progress is not None else 0.0, + ota_channel=ota_channel, + target_version=target_version, + firmware_url=firmware_url, + firmware_id=firmware_id, + source=source, ) await db_session.execute(insert_stmt) diff --git a/talkingq-url/tests/test_mqtt_device_info_report.py b/talkingq-url/tests/test_mqtt_device_info_report.py new file mode 100644 index 0000000..cd8c244 --- /dev/null +++ b/talkingq-url/tests/test_mqtt_device_info_report.py @@ -0,0 +1,111 @@ +import pytest + +from handlers.mqtt_handler import TalkingQMQTTService + + +@pytest.mark.asyncio +async def test_device_info_report_persists_metadata_and_periodic_location(monkeypatch): + service = TalkingQMQTTService({"broker": "127.0.0.1"}) + calls = [] + published = [] + + async def fake_publish(topic, payload): + published.append((topic, payload)) + return True + + async def fake_schedule_persistence(device_id, label, coro): + calls.append(("schedule", device_id, label)) + await coro + + async def fake_insert_or_update(**kwargs): + calls.append(("settings", kwargs)) + + async def fake_report_mqtt_device_location(**kwargs): + calls.append(("location", kwargs)) + return {"device_id": kwargs["device_id"], "lat": kwargs["latitude"], "lng": kwargs["longitude"]} + + monkeypatch.setattr(service, "_publish", fake_publish) + monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence) + monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.insert_or_update", fake_insert_or_update) + monkeypatch.setattr("handlers.mqtt_handler.location_service.report_mqtt_device_location", fake_report_mqtt_device_location) + + await service._handle_device_info( + "TalkingQ_XQSN00001005", + { + "msg_id": "000", + "data": { + "id": "TalkingQ_XQSN00001005", + "power": 80, + "signal": 4, + "voice": 60, + "imei": "861921071409242", + "version": "1.0.1", + "latitude": 23.1173393, + "longitude": 113.2855975, + "device_type": "2", + }, + }, + ) + + assert published == [ + ("device/TalkingQ_XQSN00001005/event_resp", {"msg_id": "000", "status": "success"}), + ] + assert calls == [ + ("schedule", "TalkingQ_XQSN00001005", "device_info"), + ( + "settings", + { + "device_id": "TalkingQ_XQSN00001005", + "power": 80, + "signal_strength": 4, + "version_str": "1.0.1", + "volume": 60, + "imei": "861921071409242", + "device_type": "2", + }, + ), + ("schedule", "TalkingQ_XQSN00001005", "device_info_location"), + ( + "location", + { + "device_id": "TalkingQ_XQSN00001005", + "latitude": 23.1173393, + "longitude": 113.2855975, + "coord_type": None, + "accuracy_m": None, + "altitude_m": None, + "speed_mps": None, + "heading_deg": None, + "source": 0, + "battery_pct": 80, + }, + ), + ] + + +@pytest.mark.asyncio +async def test_device_info_report_without_location_does_not_write_location(monkeypatch): + service = TalkingQMQTTService({"broker": "127.0.0.1"}) + calls = [] + + async def fake_publish(topic, payload): + del topic, payload + return True + + async def fake_schedule_persistence(device_id, label, coro): + calls.append(label) + await coro + + async def fake_insert_or_update(**kwargs): + calls.append(("settings", kwargs["device_type"])) + + monkeypatch.setattr(service, "_publish", fake_publish) + monkeypatch.setattr(service, "_schedule_persistence", fake_schedule_persistence) + monkeypatch.setattr("handlers.mqtt_handler.device_setting_service.insert_or_update", fake_insert_or_update) + + await service._handle_device_info( + "TalkingQ_XQSN00001005", + {"msg_id": "000", "data": {"power": 20, "signal": 2, "voice": 40, "version": "1.0.0", "device_type": 1}}, + ) + + assert calls == ["device_info", ("settings", 1)] diff --git a/talkingq-url/tests/test_ota_channel_selection.py b/talkingq-url/tests/test_ota_channel_selection.py new file mode 100644 index 0000000..10be94f --- /dev/null +++ b/talkingq-url/tests/test_ota_channel_selection.py @@ -0,0 +1,201 @@ +import pytest +from fastapi import HTTPException + +from banban.service.device import DeviceService +from banban.service.ota_firmware import OtaFirmwareSelection +from banban.service.device_setting import DeviceSettingService + + +@pytest.mark.asyncio +async def test_firmware_status_selects_gps_channel(monkeypatch): + service = DeviceService() + + async def fake_get_device_status(**kwargs): + del kwargs + return { + "device_id": "TalkingQ_XQSN00001005", + "version": "1.0.0", + "device_type": 2, + "ota_channel": None, + } + + async def fake_select_latest_for_device(status_row): + assert status_row["device_type"] == 2 + return OtaFirmwareSelection( + channel_key="gps", + version="1.1.0", + firmware_url="https://example.com/gps-1.1.0.bin", + firmware_id=10, + ) + + async def fake_get_firmware_update_dict(device_id): + del device_id + return None + + monkeypatch.setattr(service, "get_device_status", fake_get_device_status) + monkeypatch.setattr("banban.service.device.ota_firmware_service.select_latest_for_device", fake_select_latest_for_device) + monkeypatch.setattr("banban.service.device.device_firmware_update_manager.get_firmware_update_dict", fake_get_firmware_update_dict) + + result = await service.get_firmware_status(device_id="TalkingQ_XQSN00001005", user_id=1) + + assert result["ota_channel"] == "gps" + assert result["firmware_id"] == 10 + assert result["latest_version"] == "1.1.0" + assert result["firmware_url"] == "https://example.com/gps-1.1.0.bin" + assert result["update_available"] is True + assert result["can_update"] is True + + +@pytest.mark.asyncio +async def test_firmware_status_unknown_channel_cannot_update(monkeypatch): + service = DeviceService() + + async def fake_get_device_status(**kwargs): + del kwargs + return { + "device_id": "TalkingQ_XQSN00001005", + "version": "1.0.0", + "device_type": None, + "ota_channel": None, + } + + async def fake_select_latest_for_device(status_row): + assert status_row["device_type"] is None + return None + + async def fake_get_firmware_update_dict(device_id): + del device_id + return None + + monkeypatch.setattr(service, "get_device_status", fake_get_device_status) + monkeypatch.setattr("banban.service.device.ota_firmware_service.select_latest_for_device", fake_select_latest_for_device) + monkeypatch.setattr("banban.service.device.device_firmware_update_manager.get_firmware_update_dict", fake_get_firmware_update_dict) + + result = await service.get_firmware_status(device_id="TalkingQ_XQSN00001005", user_id=1) + + assert result["ota_channel"] is None + assert result["latest_version"] is None + assert result["can_update"] is False + + +@pytest.mark.asyncio +async def test_start_firmware_update_records_channel_and_keeps_mqtt_payload(monkeypatch): + service = DeviceService() + calls = [] + + async def fake_get_firmware_status(**kwargs): + del kwargs + return { + "device_id": "TalkingQ_XQSN00001005", + "current_version": "1.0.0", + "latest_version": "1.1.0", + "ota_channel": "no_gps", + "firmware_id": 22, + "firmware_url": "https://example.com/no-gps-1.1.0.bin", + "update_available": True, + "can_update": True, + "update_status": "idle", + "progress": 0.0, + "target_version": None, + "updated_at": None, + } + + async def fake_update_firmware_update(**kwargs): + calls.append(("record", kwargs)) + return True + + class FakeMqttService: + async def send_ota_command(self, device_id, url, version): + calls.append(("mqtt", {"device_id": device_id, "url": url, "version": version})) + return "003" + + async def fake_get_instance(): + return FakeMqttService() + + monkeypatch.setattr(service, "get_firmware_status", fake_get_firmware_status) + monkeypatch.setattr("banban.service.device.device_firmware_update_manager.update_firmware_update", fake_update_firmware_update) + monkeypatch.setattr("handlers.mqtt_handler.TalkingQMQTTService.get_instance", fake_get_instance) + + result = await service.start_firmware_update(device_id="TalkingQ_XQSN00001005", user_id=1) + + assert calls == [ + ( + "record", + { + "device_id": "TalkingQ_XQSN00001005", + "firmware_version": "1.1.0", + "update_status": "sent", + "progress": 0.0, + "ota_channel": "no_gps", + "target_version": "1.1.0", + "firmware_url": "https://example.com/no-gps-1.1.0.bin", + "firmware_id": 22, + "source": "manual", + }, + ), + ( + "mqtt", + { + "device_id": "TalkingQ_XQSN00001005", + "url": "https://example.com/no-gps-1.1.0.bin", + "version": "1.1.0", + }, + ), + ] + assert result["msg_id"] == "003" + assert result["ota_channel"] == "no_gps" + assert result["target_version"] == "1.1.0" + + +@pytest.mark.asyncio +async def test_start_firmware_update_without_known_channel_is_blocked(monkeypatch): + service = DeviceService() + + async def fake_get_firmware_status(**kwargs): + del kwargs + return { + "device_id": "TalkingQ_XQSN00001005", + "current_version": "1.0.0", + "latest_version": None, + "update_available": False, + "can_update": False, + "update_status": "idle", + "progress": 0.0, + "target_version": None, + "updated_at": None, + } + + monkeypatch.setattr(service, "get_firmware_status", fake_get_firmware_status) + + with pytest.raises(HTTPException) as exc_info: + await service.start_firmware_update(device_id="TalkingQ_XQSN00001005", user_id=1) + + assert exc_info.value.status_code == 409 + + +@pytest.mark.asyncio +async def test_device_type_update_preserves_existing_manual_ota_channel(monkeypatch): + service = DeviceSettingService() + calls = [] + + async def fake_get_setting_by_device_id(device_id): + assert device_id == "TalkingQ_XQSN00001005" + return {"device_id": device_id, "ota_channel": "factory_gps"} + + async def fake_update_setting(**kwargs): + calls.append(kwargs) + + monkeypatch.setattr(service, "get_setting_by_device_id", fake_get_setting_by_device_id) + monkeypatch.setattr(service, "update_setting", fake_update_setting) + + await service.insert_or_update( + device_id="TalkingQ_XQSN00001005", + power=80, + volume=60, + signal_strength=4, + version_str="1.0.0", + device_type=1, + ) + + assert calls[0]["device_type"] == 1 + assert calls[0]["ota_channel"] is None