Merge 1f8f717 into 6744cee

pygsmmodule/at_protocol_support/at_protocol_support.py

@@ -0,0 +1,19 @@
+from pygsmmodule.uart_support.base_uart import BaseUart
+
+
+class AtProtocolSupport(BaseUart):
+    def __init__(self, serial):
+        super().__init__(serial)
+
+    def initialize_port(self):
+        commands_to_execute = [
+            "ATE0"
+        ]
+        for command in commands_to_execute:
+            res = self.send_simple_request(command, remove_result=False)
+
+            res = res.split("\n")
+            if "OK" not in res:
+                return False
+
+        return True

pygsmmodule/supplementary/get_imei.py

@@ -1,13 +1,8 @@
-from pygsmmodule.uart_support.base_uart import BaseUart
-
-
 class GetImei(object):
     def __init__(self):
         pass
 
     @staticmethod
-    def get_imei(serial, timeout=1000):
-        communicator = BaseUart(serial)
-
-        res = communicator.send_simple_request("AT+GSN", timeout)
+    def exec_command(at_executor, timeout=1000):
+        res = at_executor.send_simple_request("AT+GSN", timeout)
         return str(res).strip() if res else res

pygsmmodule/supplementary/ussd_executor.py

@@ -0,0 +1,41 @@
+class UssdExecutor(object):
+    def __init__(self):
+        pass
+
+    @staticmethod
+    def exec_command(at_executor, command, timeout=1000, read_timeout=20000):
+        cmd = "AT+CUSD=1,\"{0}\",15".format(command)
+
+        res = at_executor.send_simple_request(cmd, timeout, remove_result=False)
+        if not res:
+            return None
+
+        res = str(res).strip()
+
+        if res != "OK":
+            return None
+
+        ntr = at_executor.read_pattern_terminated_string(read_timeout)
+
+        ntr = str(ntr).strip()
+
+        ntr = UssdExecutor._remove_prefix(ntr, "+CUSD:")
+
+        return UssdExecutor._remove_quotes(ntr)
+
+    @staticmethod
+    def _remove_prefix(value, prefix):
+        if not str(value).startswith(prefix):
+            return value
+
+        return value[len(prefix):]
+
+    @staticmethod
+    def _remove_quotes(value):
+        lindex = str(value).index("\"")
+        rindex = str(value).rindex("\"")
+
+        if (lindex == rindex) or (-1 in [lindex, rindex]):
+            return value
+
+        return value[lindex + 1: rindex]

pygsmmodule/uart_support/base_uart.py

@@ -32,7 +32,7 @@ def writeln(self, data, timeout=1000):
 
         return self._send_raw_bytes(data, timeout)
 
-    def send_simple_request(self, command, timeout=1000, possible_results = None):
+    def send_simple_request(self, command, timeout=1000, possible_results=None, remove_result=True):
         if type(command) == str:
             command = str(command).encode("ascii")
 
@@ -43,9 +43,39 @@ def send_simple_request(self, command, timeout=1000, possible_results = None):
         spent_time = self.writeln(command, timeout)
         timeout -= spent_time
 
-        return self._read_simple_result(timeout, possible_results)
+        return self._read_simple_result(timeout, possible_results, remove_result)
 
-    def _read_simple_result(self, timeout, possible_results):
+    def read_zero_terminated_string(self, read_timeout=20000, codepage="ascii"):
+        return self.read_pattern_terminated_string(read_timeout, codepage, bytearray([0x00]))
+
+    def read_pattern_terminated_string(self, read_timeout=20000, codepage="ascii", pattern=None):
+        tm_begin = time.time()
+
+        if pattern is None:
+            pattern = bytearray([0x00, 0xff, 0x0d, 0x0a])
+
+        buffer = bytearray()
+        while True:
+            if SimSupportFunctions.time_delta(tm_begin) >= read_timeout:
+                raise PyGsmModuleTimeoutError
+
+            while True:
+                b = self._serial.read(100)
+                if (b is None) or (len(b) == 0):
+                    break
+
+                idx = b.find(pattern)
+                if idx == -1:
+                    buffer += b
+                else:
+                    buffer += b[:idx]
+                    return buffer.decode(codepage)
+
+            time.sleep(0.05)
+
+        return None
+
+    def _read_simple_result(self, timeout, possible_results, remove_result=True):
         tm_begin = time.time()
 
         read_bytes_qty = 0
@@ -78,6 +108,9 @@ def _read_simple_result(self, timeout, possible_results):
             last_string = SimSupportFunctions.get_last_non_empty_string(strings[:])
 
             if last_string in possible_results:
-                return SimSupportFunctions.convert_to_string(SimSupportFunctions.remove_end_result(strings))
+                if remove_result:
+                    return SimSupportFunctions.convert_to_string(SimSupportFunctions.remove_end_result(strings))
+                else:
+                    return SimSupportFunctions.convert_to_string(strings)
 
         return None

script/exec_ussd.py

@@ -0,0 +1,33 @@
+from pygsmmodule.at_protocol_support.at_protocol_support import AtProtocolSupport
+from pygsmmodule.supplementary.ussd_executor import UssdExecutor
+
+from .shared import init_serial, load_dev_config
+
+
+def main(**kwargs):
+    settings = load_dev_config()
+    serial = init_serial(settings.uart.name, settings.uart.speed)
+    at_executor = AtProtocolSupport(serial)
+
+    print("initializing port...")
+    ok = at_executor.initialize_port()
+
+    if ok:
+        print("initialization [  OK  ]")
+    else:
+        print("initialization [FAILED]")
+        return 1
+
+    ussd_command = "*111#"
+    result = UssdExecutor.exec_command(at_executor, ussd_command)
+
+    result = str(result).strip()
+    print("executing USSD command '{}' result:\n---\n{}\n---".format(ussd_command, result))
+
+    serial.close()
+    return 0
+
+
+if __name__ == "__main__":
+    res = main()
+    exit(res)

script/get_imei.py

@@ -1,3 +1,4 @@
+from pygsmmodule.at_protocol_support.at_protocol_support import AtProtocolSupport
 from pygsmmodule.supplementary.get_imei import GetImei
 
 from .shared import init_serial, load_dev_config
@@ -6,8 +7,10 @@
 def main(**kwargs):
     settings = load_dev_config()
     serial = init_serial(settings.uart.name, settings.uart.speed)
+    at_executor = AtProtocolSupport(serial)
+
+    imei = GetImei.exec_command(at_executor)
 
-    imei = GetImei.get_imei(serial)
     print("imei = '{}'".format(imei))
 
     serial.close()