diff --git a/backend/api/v1/routers/modem.py b/backend/api/v1/routers/modem.py index 6e9d94b..7555e8d 100644 --- a/backend/api/v1/routers/modem.py +++ b/backend/api/v1/routers/modem.py @@ -13,6 +13,7 @@ ModemDeviceDetails, ModemPosition, ModemSignalQuality, + ModemFunctionality, ModemSIMStatus, OperatorInfo, PDPContext, @@ -100,6 +101,17 @@ async def fetch_serving_cell_info_by_id(modem_id: str) -> ModemCellInfo: return await modem.get_cell_info() +@modem_router_v1.get("/{modem_id}/functionality", status_code=status.HTTP_200_OK) +@modem_to_http_exception +async def fetch_functionality_by_id(modem_id: str) -> ModemFunctionality: + """ + Get functionality of a modem by modem id. + """ + modem = Modem.get_device(modem_id) + + return await modem.get_functionality() + + @modem_router_v1.post("/{modem_id}/commander", status_code=status.HTTP_200_OK) @modem_to_http_exception async def command_by_id( @@ -133,6 +145,17 @@ async def reboot_by_id(modem_id: str) -> None: return await modem.reboot() +@modem_router_v1.post("/{modem_id}/disable", status_code=status.HTTP_204_NO_CONTENT) +@modem_to_http_exception +async def disable_by_id(modem_id: str) -> None: + """ + Disable a modem by modem id. + """ + modem = Modem.get_device(modem_id) + + return await modem.disable() + + @modem_router_v1.post("/{modem_id}/reset", status_code=status.HTTP_204_NO_CONTENT) @modem_to_http_exception async def reset_by_id(modem_id: str) -> None: diff --git a/backend/modem/adapters/quectel/base.py b/backend/modem/adapters/quectel/base.py index dd5ff85..a9041de 100644 --- a/backend/modem/adapters/quectel/base.py +++ b/backend/modem/adapters/quectel/base.py @@ -28,21 +28,23 @@ def _detected(self) -> bool: # As base it should never be detected as a modem return False - async def at_commander(self, timeout: int = 10) -> ATCommander: + async def at_commander(self, timeout: int = 20) -> ATCommander: # Usually the third port is the AT port in Quectel modems, so try it first ports = [self.ports[2]] + self.ports[:2] + self.ports[3:] if len(self.ports) > 3 else self.ports end_time = time.monotonic() + timeout - for port in ports: - while time.monotonic() < end_time: + while time.monotonic() < end_time: + for port in ports: if not ATCommander.is_locked(port.device): + commander = None try: commander = ATCommander(port.device) await commander.setup() return commander except Exception: - break - await asyncio.sleep(0.1) + if commander is not None: + commander._close() + await asyncio.sleep(0.1) if time.monotonic() < end_time: raise ATConnectionError(f"Unable to detect any AT port for device {self.device}") diff --git a/backend/modem/at.py b/backend/modem/at.py index 5e90141..14575b5 100644 --- a/backend/modem/at.py +++ b/backend/modem/at.py @@ -78,12 +78,12 @@ async def setup(self) -> None: if not await self.check_ok() and not await self._configure_terminators() and not await self.check_ok(): raise ATConnectionError(f"Failed to connect to {self.port}") # Terminators we can get to work even when not perfect, but echo mode should be disabled - await self.command(ATCommand.SET_ECHO_MODE, ATDivider.UNDEFINED, "0", delay=0.5) + await self.command(ATCommand.SET_ECHO_MODE, ATDivider.UNDEFINED, "0", delay=0.1) async def _configure_terminators(self) -> None: # Set terminators - await self.command(ATCommand.SET_CMD_LINE_TERM, ATDivider.EQ, "13", delay=0.5) - await self.command(ATCommand.SET_RESP_FORMAT_CHAR, ATDivider.EQ, "10", delay=0.5) + await self.command(ATCommand.SET_CMD_LINE_TERM, ATDivider.EQ, "13", delay=0.1) + await self.command(ATCommand.SET_RESP_FORMAT_CHAR, ATDivider.EQ, "10", delay=0.1) def _close(self) -> None: if self.ser and self.ser.is_open: @@ -156,7 +156,7 @@ def __del__(self): async def raw_command( self, command: str, - delay: Optional[int] = 0.5, + delay: Optional[int] = 0.3, cmd_id_response: Optional[str] = None, raw_response: bool = False ) -> ATResponse: @@ -174,7 +174,7 @@ async def command( divider: ATDivider = ATDivider.UNDEFINED, data: str = "", cmd_id_response: bool = True, - delay: float = 0.5 + delay: float = 0.3 ) -> ATResponse: # If commands have AT+ it should include in response it, for async commands like AT+QPING # that will return OK as soon as hit, but after some time return the result as +QPING: ...... @@ -187,7 +187,7 @@ async def command( ) async def check_ok(self) -> bool: - response = await self.command(ATCommand.AT, delay=0.3) + response = await self.command(ATCommand.AT, delay=0.1) return response.status == ATResultCode.OK async def get_mt_info(self) -> ATResponse: @@ -230,7 +230,10 @@ async def reboot_modem(self) -> ATResponse: return await self.command(ATCommand.CONFIGURE_FUNCTIONALITY, ATDivider.EQ, '1,1', cmd_id_response=False) async def disable_modem(self) -> ATResponse: - return await self.command(ATCommand.CONFIGURE_FUNCTIONALITY, ATDivider.EQ, '0,1', cmd_id_response=False) + return await self.command(ATCommand.CONFIGURE_FUNCTIONALITY, ATDivider.EQ, '4,0', cmd_id_response=False) + + async def get_modem_functionality(self) -> ATResponse: + return await self.command(ATCommand.CONFIGURE_FUNCTIONALITY, ATDivider.QUESTION) async def reset_to_factory(self) -> ATResponse: return await self.command(ATCommand.RESET_TO_FACTORY, ATDivider.UNDEFINED, '0', cmd_id_response=False) diff --git a/backend/modem/models.py b/backend/modem/models.py index 5eea3b6..bc3d4f9 100644 --- a/backend/modem/models.py +++ b/backend/modem/models.py @@ -43,6 +43,12 @@ class ModemSIMStatus(Enum): CONNECTED = "1" UNKNOWN = "2" + +class ModemFunctionality(Enum): + MINIMAL = "0" + FULL = "1" + BLOCKED = "4" + # Configurations related class USBNetMode(Enum): diff --git a/backend/modem/modem.py b/backend/modem/modem.py index e384485..2abc155 100644 --- a/backend/modem/modem.py +++ b/backend/modem/modem.py @@ -18,6 +18,7 @@ ModemPosition, ModemSignalQuality, ModemSIMStatus, + ModemFunctionality, OperatorInfo, PDPContext, USBNetMode, @@ -128,6 +129,15 @@ async def get_data_usage_details(self) -> DataUsageSettings: async def reboot(self, cmd: ATCommander) -> None: await cmd.reboot_modem() + @with_at_commander + async def disable(self, cmd: ATCommander) -> None: + await cmd.disable_modem() + + @with_at_commander + async def get_functionality(self, cmd: ATCommander) -> ModemFunctionality: + response = await cmd.command(ATCommand.CONFIGURE_FUNCTIONALITY, ATDivider.QUESTION) + return ModemFunctionality(response.data[0][0]) + @with_at_commander async def factory_reset(self, cmd: ATCommander) -> None: await cmd.reset_to_factory()