/
http.py
105 lines (89 loc) · 3.42 KB
/
http.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
# -*- coding: utf-8 -*-
#
# 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, version 2 of the License.
#
# 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.
#
# vim:sw=4:ts=4:et
import re
import requests
from requests.adapters import HTTPAdapter
from amcrest.audio import Audio
from amcrest.event import Event
from amcrest.log import Log
from amcrest.motion_detection import MotionDetection
from amcrest.nas import Nas
from amcrest.network import Network
from amcrest.ptz import Ptz
from amcrest.record import Record
from amcrest.snapshot import Snapshot
from amcrest.special import Special
from amcrest.storage import Storage
from amcrest.system import System
from amcrest.user_management import UserManagement
from amcrest.utils import Utils
from amcrest.video import Video
from amcrest.config import TIMEOUT_HTTP_PROTOCOL, MAX_RETRY_HTTP_CONNECTION
class Http(System, Network, MotionDetection, Snapshot,
UserManagement, Event, Audio, Record, Video,
Log, Ptz, Special, Storage, Utils, Nas):
def __init__(self, host, port, user,
password, verbose=True, protocol='http',
retries_connection=None, timeout_protocol=None):
self._host = self.__clean_url(host)
self._port = port
self._user = user
self._password = password
self._verbose = verbose
self._protocol = protocol
self._token = requests.auth.HTTPBasicAuth(self._user, self._password)
self._base_url = self.__base_url()
if timeout_protocol is None:
self._timeout_protocol = TIMEOUT_HTTP_PROTOCOL
else:
self._timeout_protocol = timeout_protocol
if retries_connection is None:
self._retries_conn = MAX_RETRY_HTTP_CONNECTION
else:
self._retries_conn = retries_connection
# Base methods
def __clean_url(self, url):
host = re.sub(r'^http[s]?://', '', url, flags=re.IGNORECASE)
host = re.sub(r'/$', '', host)
return host
def __base_url(self, param=""):
return '%s://%s:%s/cgi-bin/%s' % (self._protocol, self._host,
str(self._port), param)
def get_base_url(self):
return self._base_url
def command(self, cmd, retries=None, timeout_cmd=None):
"""
Args:
cmd - command to execute via http
timeout_cmd - timeout, default 3sec
retries - maximum number of retries each connection should attempt
"""
if timeout_cmd is not None:
self._timeout_protocol = timeout_cmd
if retries is not None:
self._retries_conn = retries
session = requests.Session()
session.mount('http://', HTTPAdapter(max_retries=self._retries_conn))
session.mount('https://', HTTPAdapter(max_retries=self._retries_conn))
url = self.__base_url(cmd)
try:
resp = session.get(
url,
auth=self._token,
stream=True,
timeout=self._timeout_protocol,
)
resp.raise_for_status()
except:
raise
return resp