can detected disabled sim card
This commit is contained in:
+25
-8
@@ -4,6 +4,7 @@ import time
|
||||
import serial
|
||||
from serial import Serial
|
||||
|
||||
from SIMError import SIMError
|
||||
from utils.excel_reader import ExcelHelper
|
||||
|
||||
|
||||
@@ -27,14 +28,26 @@ class ModemPool:
|
||||
# wait for 20 second, so that the modem can init all the sims
|
||||
time.sleep(20)
|
||||
|
||||
def _generate_error_msg(self, slot_position, index, error: SIMError):
|
||||
return "error for slot({}) SIM({}), error:{}".format(slot_position, index + 1,
|
||||
error.value)
|
||||
|
||||
def get_raw_phone_number(self, slot_position):
|
||||
for index, ser in enumerate(self._serial_list):
|
||||
print("will get phone number for slot({}) SIM({}), port:{}".format(slot_position, index + 1, ser.port))
|
||||
self._select_sim_storage(ser)
|
||||
if not self._select_sim_storage(ser):
|
||||
print(self._generate_error_msg(slot_position, index, SIMError.STORAGE_ERROR))
|
||||
continue
|
||||
msg = self._execut_USSD_cmd("AT+CUSD=1, *132#\r", ser)
|
||||
if "Unfortunately" in str(msg):
|
||||
print("error for for slot({}) SIM({}), port:{}".format(slot_position, index + 1, ser.port))
|
||||
return
|
||||
print(self._generate_error_msg(slot_position, index, SIMError.SIM_DISABLED))
|
||||
continue
|
||||
elif "CME ERROR" in str(msg):
|
||||
print(self._generate_error_msg(slot_position, index, SIMError.CME_ERROR))
|
||||
continue
|
||||
elif len(msg) == 0:
|
||||
print(self._generate_error_msg(slot_position, index, SIMError.TIMEOUT))
|
||||
continue
|
||||
# find phone number
|
||||
match = re.search(r'33\d{9}', str(msg))
|
||||
phone_number = match.group(0)
|
||||
@@ -49,13 +62,17 @@ class ModemPool:
|
||||
def get_own_number(self, ser: Serial):
|
||||
print("saved phone number: " + str(self._send_command(f'AT+CPBR={self.phone_number_position}\r', ser)))
|
||||
|
||||
def _select_sim_storage(self, ser):
|
||||
def _select_sim_storage(self, ser) -> bool:
|
||||
# use SIM Card storage
|
||||
cmd_sm = "AT+CPBS=\"SM\"\r"
|
||||
self._send_command(cmd_sm, ser)
|
||||
result = self._send_command(cmd_sm, ser)
|
||||
if "ERROR" in str(result):
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
def _send_command(self, cmd: str, ser, wait_time_in_s: int = 0) -> bytes:
|
||||
print("send command {}".format(cmd))
|
||||
# print("send command {}".format(cmd))
|
||||
ser.write(cmd.encode())
|
||||
msg = ser.read(100)
|
||||
count = 0
|
||||
@@ -67,6 +84,6 @@ class ModemPool:
|
||||
print(msg)
|
||||
return msg
|
||||
|
||||
def _execut_USSD_cmd(self, cmd, ser) -> str:
|
||||
def _execut_USSD_cmd(self, cmd, ser) -> bytes:
|
||||
# the timeout for ussd command can be 120 s in mac
|
||||
return str(self._send_command(cmd, ser, 120))
|
||||
return self._send_command(cmd, ser, 120)
|
||||
|
||||
@@ -3,3 +3,6 @@ from enum import Enum
|
||||
|
||||
class SIMError(Enum):
|
||||
SIM_DISABLED = "SIM_DISABLED"
|
||||
CME_ERROR = "CME_ERROR"
|
||||
STORAGE_ERROR = "STORAGE_ERROR"
|
||||
TIMEOUT = "TIMEOUT"
|
||||
|
||||
+1
-1
@@ -31,7 +31,7 @@ class CardPool:
|
||||
if slot_number < 10:
|
||||
self._send_command("AT+SWIT00-000{}\r".format(slot_number))
|
||||
else:
|
||||
self._send_command("AT+SWIT00-0{}\r".format(slot_number))
|
||||
self._send_command("AT+SWIT00-00{}\r".format(slot_number))
|
||||
|
||||
# not work for the pool
|
||||
def find_current_slot(self):
|
||||
|
||||
Binary file not shown.
@@ -164,9 +164,6 @@ def on_message_received(ch, method, properties, body):
|
||||
is_finished = True
|
||||
|
||||
|
||||
# save the result to db
|
||||
|
||||
|
||||
def start_listen():
|
||||
logger.info("start to listen to message queue")
|
||||
receiver = MessageReceiver()
|
||||
@@ -177,14 +174,15 @@ def read_all_the_phone_number():
|
||||
card_pool = CardPool("/dev/tty.usbmodem11301")
|
||||
slot_number = 1
|
||||
slot_sum = 32
|
||||
card_pool.reset()
|
||||
card_pool.switch_to_slot(15)
|
||||
|
||||
for i in range(2, slot_sum + 1):
|
||||
# i = 2
|
||||
card_pool.switch_to_slot(i)
|
||||
modem_pool = ModemPool(get_devices_ports())
|
||||
modem_pool.reset_all_modems()
|
||||
modem_pool.get_raw_phone_number(i)
|
||||
# for i in range(10, slot_sum + 1):
|
||||
# card_pool.reset()
|
||||
# print("will switch to " + str(i))
|
||||
# card_pool.switch_to_slot(i)
|
||||
# modem_pool = ModemPool(get_devices_ports())
|
||||
# modem_pool.reset_all_modems()
|
||||
# modem_pool.get_raw_phone_number(i)
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
|
||||
Reference in New Issue
Block a user