mirror of
https://github.com/openmv/openmv.git
synced 2025-09-26 23:09:13 +08:00
202 lines
5.7 KiB
Python
202 lines
5.7 KiB
Python
# The MIT License (MIT)
|
|
#
|
|
# Copyright (c) 2013, 2014 micropython-lib contributors
|
|
#
|
|
# Permission is hereby granted, free of charge, to any person obtaining a copy
|
|
# of this software and associated documentation files (the "Software"), to deal
|
|
# in the Software without restriction, including without limitation the rights
|
|
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
|
|
# copies of the Software, and to permit persons to whom the Software is
|
|
# furnished to do so, subject to the following conditions:
|
|
#
|
|
# The above copyright notice and this permission notice shall be included in
|
|
# all copies or substantial portions of the Software.
|
|
#
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
|
|
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
|
|
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
|
|
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
|
|
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
|
|
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
|
|
# THE SOFTWARE.
|
|
|
|
# Source: improved version of micropython-lib's requests.
|
|
# Some useful links for future updates:
|
|
# https://www.w3.org/TR/html401/interact/forms.html#h-17.13.4
|
|
# https://docs.python-requests.org/en/master/
|
|
|
|
import socket
|
|
import binascii
|
|
|
|
|
|
class Response:
|
|
def __init__(self, code, reason, headers=None, content=None):
|
|
self.encoding = "utf-8"
|
|
self._content = content
|
|
self._headers = headers
|
|
self.reason = reason
|
|
self.status_code = code
|
|
|
|
@property
|
|
def headers(self):
|
|
return str(self._headers, self.encoding)
|
|
|
|
@property
|
|
def content(self):
|
|
return str(self._content, self.encoding)
|
|
|
|
def json(self):
|
|
import json
|
|
|
|
return json.loads(self._content)
|
|
|
|
|
|
def readline(s):
|
|
l = bytearray()
|
|
while True:
|
|
try:
|
|
l += s.read(1)
|
|
if l[-1] == b"\n":
|
|
break
|
|
except:
|
|
break
|
|
return l
|
|
|
|
|
|
def socket_readall(s):
|
|
buf = b""
|
|
while True:
|
|
recv = b""
|
|
try:
|
|
recv = s.recv(1)
|
|
except:
|
|
pass
|
|
if len(recv) == 0:
|
|
break
|
|
buf += recv
|
|
return buf
|
|
|
|
|
|
def request(method, url, data=None, json=None, files=None, headers={}, auth=None, stream=None):
|
|
try:
|
|
proto, dummy, host, path = url.split("/", 3)
|
|
except ValueError:
|
|
proto, dummy, host = url.split("/", 2)
|
|
path = ""
|
|
if proto == "http:":
|
|
port = 80
|
|
elif proto == "https:":
|
|
import ssl
|
|
|
|
port = 443
|
|
else:
|
|
raise ValueError("Unsupported protocol: " + proto)
|
|
|
|
if ":" in host:
|
|
host, port = host.split(":", 1)
|
|
port = int(port)
|
|
|
|
if auth:
|
|
headers["Authorization"] = b"Basic %s" % (
|
|
binascii.b2a_base64("%s:%s" % (auth[0], auth[1]))[0:-1]
|
|
)
|
|
|
|
resp_code = 0
|
|
resp_reason = None
|
|
resp_headers = []
|
|
|
|
ai = socket.getaddrinfo(host, port)[0]
|
|
s = socket.socket(ai[0], ai[1], ai[2])
|
|
try:
|
|
s.connect(ai[-1])
|
|
s.settimeout(5.0)
|
|
if proto == "https:":
|
|
s = ssl.wrap_socket(s, server_hostname=host)
|
|
|
|
s.write(b"%s /%s HTTP/1.0\r\n" % (method, path))
|
|
|
|
if "Host" not in headers:
|
|
s.write(b"Host: %s\r\n" % host)
|
|
|
|
# Iterate over keys to avoid tuple alloc
|
|
for k in headers:
|
|
s.write(k)
|
|
s.write(b": ")
|
|
s.write(headers[k])
|
|
s.write(b"\r\n")
|
|
|
|
if json is not None:
|
|
import json
|
|
|
|
data = json.dumps(json)
|
|
s.write(b"Content-Type: application/json\r\n")
|
|
|
|
if files is not None:
|
|
data = bytearray()
|
|
boundary = b"37a4bcce91521f74142f1868e328a6b9"
|
|
s.write(b"Content-Type: multipart/form-data; boundary=%s\r\n" % (boundary))
|
|
for name, fileobj in files.items():
|
|
data += b"--%s\r\n" % (boundary)
|
|
data += b'Content-Disposition: form-data; name="%s"; filename="%s"\r\n\r\n' % (
|
|
name,
|
|
fileobj[0],
|
|
)
|
|
data += fileobj[1].read()
|
|
data += b"\r\n"
|
|
data += b"\r\n--%s--\r\n" % (boundary)
|
|
|
|
if data:
|
|
s.write(b"Content-Length: %d\r\n\r\n" % len(data))
|
|
s.write(data)
|
|
else:
|
|
s.write(b"\r\n")
|
|
|
|
response = socket_readall(s).split(b"\r\n")
|
|
while response:
|
|
l = response.pop(0).strip()
|
|
if not l or l == b"\r\n":
|
|
break
|
|
if l.startswith(b"Transfer-Encoding:"):
|
|
if b"chunked" in l:
|
|
raise ValueError("Unsupported " + l)
|
|
elif l.startswith(b"Location:") and not 200 <= status <= 299:
|
|
raise NotImplementedError("Redirects not yet supported")
|
|
if "HTTPS" in l or "HTTP" in l:
|
|
sline = l.split(None, 2)
|
|
resp_code = int(sline[1])
|
|
resp_reason = sline[2].decode().rstrip() if len(sline) > 2 else ""
|
|
continue
|
|
resp_headers.append(l)
|
|
resp_headers = b"\r\n".join(resp_headers)
|
|
content = b"\r\n".join(response)
|
|
except OSError:
|
|
raise
|
|
finally:
|
|
s.close()
|
|
|
|
return Response(resp_code, resp_reason, resp_headers, content)
|
|
|
|
|
|
def head(url, **kw):
|
|
return request("HEAD", url, **kw)
|
|
|
|
|
|
def get(url, **kw):
|
|
return request("GET", url, **kw)
|
|
|
|
|
|
def post(url, **kw):
|
|
return request("POST", url, **kw)
|
|
|
|
|
|
def put(url, **kw):
|
|
return request("PUT", url, **kw)
|
|
|
|
|
|
def patch(url, **kw):
|
|
return request("PATCH", url, **kw)
|
|
|
|
|
|
def delete(url, **kw):
|
|
return request("DELETE", url, **kw)
|