Add support for talking direct to cameras, and setting micro LED status

tests/test_camera.py

@@ -0,0 +1,69 @@
+try:
+    import httplib
+except ImportError:
+    from http import client as httplib
+
+import unittest
+
+import mock
+
+from uvcclient import camera
+
+
+class TestCamera(unittest.TestCase):
+    def test_set_led_on(self):
+        c = camera.UVCCameraClient('foo', 'ubnt', 'ubnt')
+        with mock.patch.object(c, '_cfgwrite') as mock_write:
+            c.set_led(False)
+            mock_write.assert_called_once_with('led.front.status', 0)
+
+    def test_set_led_off(self):
+        c = camera.UVCCameraClient('foo', 'ubnt', 'ubnt')
+        with mock.patch.object(c, '_cfgwrite') as mock_write:
+            c.set_led(True)
+            mock_write.assert_called_once_with('led.front.status', 1)
+
+    def test_cfgwrite(self):
+        c = camera.UVCCameraClient('foo', 'ubnt', 'ubnt')
+        c._cookie = 'foo-cookie'
+        with mock.patch.object(httplib, 'HTTPConnection') as mock_h:
+            conn = mock_h.return_value
+            conn.getresponse.return_value.status = 200
+            self.assertTrue(c._cfgwrite('foo', 'bar'))
+            headers = {'Cookie': 'foo-cookie'}
+            conn.request.assert_called_once_with('GET',
+                                                 '/cfgwrite.cgi?foo=bar',
+                                                 headers=headers)
+
+    @mock.patch.object(httplib, 'HTTPConnection')
+    def test_login(self, mock_h):
+        c = camera.UVCCameraClient('foo', 'ubnt', 'ubnt')
+        first = mock.MagicMock()
+        second = mock.MagicMock()
+        counter = [0]
+
+        def fake_conn(*a, **k):
+            if counter[0] == 0:
+                counter[0] += 1
+                return first
+            elif counter[0] == 1:
+                counter[0] += 1
+                return second
+
+        mock_h.side_effect = fake_conn
+        cookie = 'thecookie AIROS_SESSIONID=foo; bar'
+        first.getresponse.return_value.getheaders.return_value = [
+            ('set-cookie', cookie)]
+        second.getresponse.return_value.status = 200
+        c.login()
+        first.request.assert_called_once_with('GET', '/')
+        self.assertEqual('POST',
+                         second.request.call_args_list[0][0][0])
+        self.assertEqual('/login.cgi',
+                         second.request.call_args_list[0][0][1])
+        formdata = 'AIROS_SESSIONID=foo&password=ubnt&username=ubnt'
+        self.assertEqual(
+            sorted(formdata.split('&')),
+            sorted(second.request.call_args_list[0][0][2].split('&')))
+        self.assertEqual(cookie,
+                         second.request.call_args_list[0][0][3]['Cookie'])

uvcclient/camera.py

@@ -0,0 +1,76 @@
+#   Copyright 2015 Dan Smith ([email protected])
+#
+#   This program is free software: you can redistribute it and/or modify
+#   it under the terms of the GNU General Public License as published by
+#   the Free Software Foundation, either version 3 of the License, or
+#   (at your option) any later version.
+#
+#   This program is distributed in the hope that it will be useful,
+#   but WITHOUT ANY WARRANTY; without even the implied warranty of
+#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+#   GNU General Public License for more details.
+#
+#   You should have received a copy of the GNU General Public License
+#   along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import logging
+
+# Python3 compatibility
+try:
+    import httplib
+except ImportError:
+    from http import client as httplib
+try:
+    import urlparse
+    import urllib
+except ImportError:
+    import urllib.parse as urlparse
+
+
+class UVCCameraClient(object):
+    def __init__(self, host, username, password, port=80):
+        self._host = host
+        self._port = port
+        self._username = username
+        self._password = password
+        self._cookie = None
+        self._log = logging.getLogger('UVCCamera(%s)' % self._host)
+
+    def login(self):
+        conn = httplib.HTTPConnection(self._host, self._port)
+        conn.request('GET', '/')
+        resp = conn.getresponse()
+        headers = dict(resp.getheaders())
+        self._cookie = headers['set-cookie']
+        session = self._cookie.split('=')[1].split(';')[0]
+
+        try:
+            urlencode = urllib.urlencode
+        except NameError:
+            urlencode = urlparse.urlencode
+
+        data = urlencode({'username': self._username,
+                          'password': self._password,
+                          'AIROS_SESSIONID': session})
+        headers = {"Content-type": "application/x-www-form-urlencoded",
+                   "Accept": "*",
+                   'Cookie': self._cookie}
+        conn = httplib.HTTPConnection(self._host, self._port)
+        req = conn.request('POST', '/login.cgi', data, headers)
+        resp = conn.getresponse()
+        if resp.status != 200:
+            raise Exception('Failed to login: %s' % resp.reason)
+
+    def _cfgwrite(self, setting, value):
+        conn = httplib.HTTPConnection(self._host, self._port)
+        headers = {'Cookie': self._cookie}
+        conn.request('GET', '/cfgwrite.cgi?%s=%s' % (setting, value),
+                     headers=headers)
+        resp = conn.getresponse()
+        self._log.debug('Setting %s=%s: %s %s' % (setting, value,
+                                                  resp.status,
+                                                  resp.reason))
+        return resp.status == 200
+
+    def set_led(self, enabled):
+        return self._cfgwrite('led.front.status', int(enabled))

uvcclient/main.py

@@ -18,6 +18,16 @@
 import optparse
 
 from uvcclient import nvr
+from uvcclient import camera
+
+
+def do_led(camera_info, enabled):
+    cam_client = camera.UVCCameraClient(camera_info['host'],
+                                        camera_info['username'],
+                                        'ubnt')  # FIXME
+    cam_client.login()
+    print enabled
+    cam_client.set_led(enabled)
 
 
 def main():
@@ -46,6 +56,8 @@ def main():
                       default=None,
                       help=('Set picture settings with a string like that '
                             'returned from --get-picture-settings'))
+    parser.add_option('--set-led', default=None, metavar='ENABLED',
+                      help='Enable/Disable front LED (on,off)')
     opts, args = parser.parse_args()
 
     if not all([host, port, apikey]):
@@ -116,3 +128,12 @@ def main():
             if type(result[k])(settings[k]) != result[k]:
                 print('Rejected: %s' % k)
         return 0
+    elif opts.set_led is not None:
+        camera = client.get_camera(opts.uuid)
+        if not camera:
+            print 'No such camera'
+            return 1
+        if 'Micro' not in camera['model']:
+            print 'Only micro cameras support LED status'
+            return 2
+        do_led(camera, opts.set_led.lower() == 'on')