Merge branch 'release/0.1.0'

.travis.yml

@@ -0,0 +1,15 @@
+language: python
+
+python:
+  - "2.7"
+
+env:
+    - PIP_USE_MIRRORS=true
+
+script: "python setup.py test"
+
+notifications:
+  slack:
+    rooms:
+      secure: YLa2vYsbfHUpLkBxqYrD1W4WiP9hvFKvte9imGgSbC1aZsjT//poSHRX505taxsyb1JbXGpQ+cwE4S7Mw7DfGRWhqOBuSZds+U109H5fHUN0ea3rkDBrENTRQNoVUbjps9lMS3qVadx5UzBIDavKb4QfYwRO7VfLQwxTVXU6dvyJSyesdsmf6KI/JJmS0b3a0fLqZcPzR/yGv9OmOntPJN/ZPnGmGc227oxFBFy1k5qGKxXDbf8F2wH+ZaK9mRir1n1QU5QGDJtDI9J7WXsepBnkAQa9UG4FeZ/mGgMvCM3cPDftKs67xo9Q3ohiSqZbYDqFy5FIcMOld//JtAn3UlPjOiAiuA37BZ1Cjpl+ahTyEMQseMDtN3G2Gzq4TfXlaEu5NkKU306juCMAJRokihYroUvaSFfiQRWkEz6wPc7dVOdoWpEw4ZlBhFPEUk2rtlJnNlgpFK7QVB9R5fFnpWDpDtgf97v6/i6KdcSCs68NqOPX7+CD+NjPcrbNFDZ2pzMdwjIqp6w/GXZ22Nco8n9FzyGvEnVGzY1AgaMaNvgaiVZxMuMw7MUuDs2bi9ce3tUfmx/TdHDaKc/0RMoPsy1h3V94YIIeQ+Iz5vp4IolBlxY2GIzumCgsJCl9a+V9QUflkSByNL568dsEkfrvb0KPhJpHq4JEtc/2aLhnyoM=
+    on_success: always

README.md

@@ -30,6 +30,11 @@ $ cat files.txt | xargs rm -rf
 ```
 
 # Revision history
+* 0.1.0
+   - Remove modem_auto_connect()
+   - Add a new function to enable ACM, modem_enable_acm()
+   - Add a new function delete APN, apn_del()
+   - Add a new static function to resolve modem serial port
 
  * 0.0.1
     - Initial beta release

lib/candy_board_amt/__init__.py

@@ -11,6 +11,7 @@
 import termios
 import threading
 import time
+import glob
 
 # SerialPort class was imported from John Wiseman's https://github.com/wiseman/arduino-serial/blob/master/arduinoserial.py
 
@@ -111,6 +112,24 @@ def write(self, str):
     def write_byte(self, byte):
         os.write(self.fd, chr(byte))
 
+    def close(self):
+        try:
+            os.close(self.fd)
+        except OSError:
+            pass
+
+    @staticmethod
+    def resolve_modem_port(self):
+        for p in glob.glob("/dev/tty*"):
+            port = candy_board_amt.SerialPort(p, 115200)
+            port.write("AT\r")
+            time.sleep(0.1)
+            ret = port.read_line()
+            port.close()
+            if ret == "AT":
+                return p
+        return None
+
 class SockServer(threading.Thread):
     def __init__(self, version, apn, sock_path="/var/run/candy-board-service.sock", serial=None):
         super(SockServer, self).__init__()
@@ -120,7 +139,12 @@ def __init__(self, version, apn, sock_path="/var/run/candy-board-service.sock",
         self.serial = serial
         self.debug = False
         if apn:
-            self.apn_set(apn['apn'], apn['user'], apn['password'])
+            cmd = {
+                'name': apn['apn'],
+                'user_id': apn['user'],
+                'password': apn['password']
+            }
+            self.apn_set(cmd)
 
     def recv(self, connection, size):
         ready, _, _ = select.select([connection], [], [], 5)
@@ -166,25 +190,16 @@ def run(self):
                     connection.close()
 
     def perform(self, cmd):
-        if cmd['category'] == "apn":
-            if cmd['action'] == "ls":
-                return self.apn_ls()
-            elif cmd['action'] == "set":
-                return self.apn_set(cmd['name'], cmd['user_id'], cmd['password'])
-        elif cmd['category'] == "network":
-            if cmd['action'] == "show":
-                return self.network_show()
-        elif cmd['category'] == "sim":
-            if cmd['action'] == "show":
-                return self.sim_show()
-        elif cmd['category'] == "modem":
-            if cmd['action'] == "show":
-                return self.modem_show()
-        elif cmd['category'] == "service":
-            if cmd['action'] == "version":
-                return self.service_version()
-
-        return "Unknown Command"
+        try:
+            m = getattr(self.__class__, "%s_%s" % (cmd['category'], cmd['action']))
+            return m(self, cmd)
+        except AttributeError:
+            return self.error_message("Unknown Command")
+        except KeyError:
+            return self.error_message("Invalid Args")
+
+    def error_message(self, msg):
+        return json.dumps({"status":"ERROR","result":msg})
 
     def read_line(self):
         line = self.serial.read_line()
@@ -215,20 +230,22 @@ def send_at(self, cmd):
             print("cmd:[%s] => status:[%s], result:[%s]" % (cmd, status, result))
         return (status, result.strip())
 
-    def apn_ls(self):
+    def apn_ls(self, cmd):
         status, result = self.send_at("AT+CGDCONT?")
         apn_list = []
         if status == "OK":
-            name_list = map(lambda e: e[10:].split(",")[2].translate(None, '"'), result.split("\n"))
+            id_name_list = map(lambda e: e[10:].split(",")[0] + "," + e[10:].split(",")[2].translate(None, '"'), result.split("\n"))
             status, result = self.send_at("AT$QCPDPP?")
             creds_list = []
             if status == "OK":
                 creds_list = map(lambda e: e[2].translate(None, '"'),
                     filter(lambda e: len(e) > 2,
                         map(lambda e: e[9:].split(","), result.split("\n"))))
-            for i in range(len(name_list)):
+            for i in range(len(id_name_list)):
+                id_name = id_name_list[i].split(",")
                 apn = {
-                    'apn': name_list[i]
+                    'apn_id': id_name[0],
+                    'apn': id_name[1]
                 }
                 if i < len(creds_list):
                     apn['user'] = creds_list[i]
@@ -241,17 +258,32 @@ def apn_ls(self):
         }
         return json.dumps(message)
 
-    def apn_set(self, name, user_id, password):
-        status, result = self.send_at("AT+CGDCONT=1,\"IPV4V6\",\"%s\",\"0.0.0.0\",0,0" % name)
+    def apn_set(self, cmd):
+        (name, user_id, password) = (cmd['name'], cmd['user_id'], cmd['password'])
+        apn_id = "1"
+        if 'apn_id' in cmd:
+            apn_id = cmd['apn_id']
+        status, result = self.send_at("AT+CGDCONT=%s,\"IPV4V6\",\"%s\",\"0.0.0.0\",0,0" % (apn_id, name))
         if status == "OK":
-            status, result = self.send_at("AT$QCPDPP=1,3,\"%s\",\"%s\"" % (password, user_id))
+            status, result = self.send_at("AT$QCPDPP=%s,3,\"%s\",\"%s\"" % (apn_id, password, user_id))
+        message = {
+            'status': status,
+            'result': result
+        }
+        return json.dumps(message)
+
+    def apn_del(self, cmd):
+        apn_id = "1"
+        if 'apn_id' in cmd:
+            apn_id = cmd['apn_id']
+        status, result = self.send_at("AT+CGDCONT=%s" % apn_id) # removes QCPDPP as well
         message = {
             'status': status,
             'result': result
         }
         return json.dumps(message)
 
-    def network_show(self):
+    def network_show(self, cmd):
         status, result = self.send_at("AT+CSQ")
         rssi = ""
         network = "UNKNOWN"
@@ -287,7 +319,7 @@ def network_show(self):
         }
         return json.dumps(message)
 
-    def sim_show(self):
+    def sim_show(self, cmd):
         state = "SIM_STATE_ABSENT"
         msisdn = ""
         imsi = ""
@@ -307,7 +339,7 @@ def sim_show(self):
         }
         return json.dumps(message)
 
-    def modem_show(self):
+    def modem_show(self, cmd):
         status, result = self.send_at("ATI")
         man = "UNKNOWN"
         mod = "UNKNOWN"
@@ -330,7 +362,23 @@ def modem_show(self):
         }
         return json.dumps(message)
 
-    def service_version(self):
+    def modem_enable_ecm(self, cmd):
+        status, result = self.send_at("AT@USBCHG=ECM") # modem will reboot, @AUTOCONN=1
+        message = {
+            'status': status,
+            'result': result
+        }
+        return json.dumps(message)
+
+    def modem_enable_acm(self, cmd):
+        status, result = self.send_at("AT@USBCHG=ACM") # modem will reboot, @AUTOCONN=0
+        message = {
+            'status': status,
+            'result': result
+        }
+        return json.dumps(message)
+
+    def service_version(self, cmd):
         message = {
             'status': 'OK',
             'result': {

pytest.ini

@@ -0,0 +1,2 @@
+[pytest]
+addopts = --ignore=setup.py --maxfail=1 --capture=sys --cov-report term-missing --cov candy_board_amt

setup.py

@@ -3,8 +3,20 @@
 import os
 import sys
 from setuptools import setup, find_packages
+from setuptools.command.test import test as TestCommand
 
-version = "0.0.1"
+version = "0.1.0"
+
+class PyTest(TestCommand):
+    def finalize_options(self):
+        TestCommand.finalize_options(self)
+        self.test_args = []
+        self.test_suite = True
+    def run_tests(self):
+        #import here, cause outside the eggs aren't loaded
+        import pytest
+        errno = pytest.main(self.test_args)
+        sys.exit(errno)
 
 if sys.argv[-1] == 'publish':
     os.system('python setup.py sdist upload')
@@ -37,4 +49,8 @@
     keywords=(
         'CANDY EGG', 'CANDY LINE'
     ),
+    tests_require=['pytest-cov>=2.2.0',
+                   'pytest>=2.8.2',
+                   'terminaltables>=1.2.1'],
+    cmdclass = {'test': PyTest}
 )