complete the check sim function
This commit is contained in:
+46
-30
@@ -2,6 +2,7 @@ import re
|
||||
import time
|
||||
|
||||
import serial
|
||||
from gsmmodem import GsmModem
|
||||
from serial import Serial
|
||||
|
||||
from definitions import BAUDRATE
|
||||
@@ -9,7 +10,9 @@ from error.SIMError import SIMError
|
||||
from logs.LogSender import LOG_APPOINTMENT_SUCCESS, SUBJECT_SIM_INFO
|
||||
from params import firebase_store_manager, oracle_log_sender
|
||||
from pojo.SimInfoPojo import SimInfoPojo
|
||||
from pojo.serial_modem import SerialModem
|
||||
from utils.excel_reader import ExcelHelper
|
||||
from utils.operator import check_operator, Operator
|
||||
|
||||
|
||||
class ModemPool:
|
||||
@@ -43,42 +46,55 @@ class ModemPool:
|
||||
return msg
|
||||
|
||||
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))
|
||||
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(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)
|
||||
print("phone is " + phone_number)
|
||||
sim_position = index + 1
|
||||
position = (slot_position - 1) * len(self._port_list) + sim_position
|
||||
# unlock sim
|
||||
unlock_cmd = 'AT+CPIN="{0}\r"'.format("0000")
|
||||
self._send_command(unlock_cmd, ser, 10)
|
||||
cmd = "AT+CCID\r"
|
||||
response = str(self._send_command(cmd, ser))
|
||||
ccid_group = re.search("[0-9F]+", response)
|
||||
ccid = ccid_group.group(0)
|
||||
sim_position = index + 1
|
||||
position = (slot_position - 1) * len(self._port_list) + sim_position
|
||||
if phone_number:
|
||||
self._db_manager.save_sim_info(SimInfoPojo(phone=phone_number, ccid=ccid, position=position))
|
||||
self._log_sender.send_log(phone_number, source=self.TAG, subject=SUBJECT_SIM_INFO,
|
||||
type=LOG_APPOINTMENT_SUCCESS)
|
||||
# write the number to sim card's phonebook
|
||||
cmd = f'AT+CPBW={self.phone_number_position},\"{phone_number}\"\r'
|
||||
self._send_command(cmd, ser, wait_time_in_s=2)
|
||||
self.get_own_number(ser)
|
||||
operator = check_operator(ccid)
|
||||
if operator == Operator.SFR or operator == Operator.CHINA_TELECOM:
|
||||
contacts = self._excel_helper.read_contacts()
|
||||
contact = [contact for contact in contacts if
|
||||
contact.ccid.replace("F", "") == ccid.replace("F", "")]
|
||||
if len(contact) > 0:
|
||||
phone_number = contact[0].phone
|
||||
self._db_manager.save_sim_info(SimInfoPojo(phone=str(phone_number), ccid=ccid, position=position))
|
||||
else:
|
||||
error_msg = "slot({}),sim({})".format(slot_position, sim_position)
|
||||
oracle_log_sender.send_contact_not_found(error_msg)
|
||||
else:
|
||||
print("will get phone number for slot({}) SIM({}), port:{}".format(slot_position, index + 1, ser.port))
|
||||
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(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)
|
||||
print("phone is " + phone_number)
|
||||
|
||||
def get_own_number(self, ser: Serial):
|
||||
print("saved phone number: " + str(self._send_command(f'AT+CPBR={self.phone_number_position}\r', ser)))
|
||||
if phone_number:
|
||||
self._db_manager.save_sim_info(SimInfoPojo(phone=phone_number, ccid=ccid, position=position))
|
||||
self._log_sender.send_log(phone_number, source=self.TAG, subject=SUBJECT_SIM_INFO,
|
||||
type=LOG_APPOINTMENT_SUCCESS)
|
||||
# write the number to sim card's phonebook
|
||||
cmd = f'AT+CPBW={self.phone_number_position},\"{phone_number}\"\r'
|
||||
self._send_command(cmd, ser, wait_time_in_s=2)
|
||||
|
||||
def _select_sim_storage(self, ser) -> bool:
|
||||
# use SIM Card storage
|
||||
|
||||
Reference in New Issue
Block a user