018 - ESP32 MicroPython: Thingspeak | RESTful APIs

 1# **************************************#
 2# Author: George Bantique               #
 3#         TechToTinker Youtube Channel  #
 4#         TechToTinker.blogspot.com     #
 5#         tech.to.tinker@gmail.com      #
 6# Date: Nov.13, 2020                    #
 7# Please feel free to modify the code   #
 8# according to your needs.              #
 9# **************************************#
10
11# **************************************
12# Load necessary libraries
13import machine
14import network 
15import wifi_credentials 
16import urequests 
17import dht 
18import time 
19
20# **************************************
21# Create objects:
22led = machine.Pin(2,machine.Pin.OUT) 
23d = dht.DHT22(machine.Pin(23)) 
24
25# *************************************
26# Configure the ESP32 wifi as STAtion
27sta = network.WLAN(network.STA_IF)
28if not sta.isconnected(): 
29  print('connecting to network...') 
30  sta.active(True) 
31  #sta.connect('your wifi ssid', 'your wifi password') 
32  sta.connect(wifi_credentials.ssid, wifi_credentials.password) 
33  while not sta.isconnected(): 
34    pass 
35print('network config:', sta.ifconfig()) 
36
37# **************************************
38# Constants and variables:
39HTTP_HEADERS = {'Content-Type': 'application/json'} 
40THINGSPEAK_WRITE_API_KEY = 'W6ANWJGN2T7507GW' 
41UPDATE_TIME_INTERVAL = 5000  # in ms 
42last_update = time.ticks_ms() 
43# initially there would be some delays 
44# before submitting the first update 
45# but should be enough to stabilize the 
46# the DHT sensor. 
47
48# **************************************
49# Main loop:
50while True: 
51    if time.ticks_ms() - last_update >= UPDATE_TIME_INTERVAL: 
52        d.measure() 
53        t = d.temperature() 
54        h = d.humidity() 
55         
56        dht_readings = {'field1':t, 'field2':h} 
57        request = urequests.post( 
58          'http://api.thingspeak.com/update?api_key=' +
59          THINGSPEAK_WRITE_API_KEY, 
60          json = dht_readings, 
61          headers = HTTP_HEADERS )  
62        request.close() 
63        print(dht_readings) 
64         
65        led.value(not led.value()) 
66        last_update = time.ticks_ms()

1ssid = "your wifi ssid"
2password = "your wifi password"

  1import usocket
  2
  3class Response:
  4    def __init__(self, f):
  5        self.raw = f
  6        self.encoding = "utf-8"
  7        self._cached = None
  8
  9    def close(self):
 10        if self.raw:
 11            self.raw.close()
 12            self.raw = None
 13        self._cached = None
 14
 15    @property
 16    def content(self):
 17        if self._cached is None:
 18            try:
 19                self._cached = self.raw.read()
 20            finally:
 21                self.raw.close()
 22                self.raw = None
 23        return self._cached
 24
 25    @property
 26    def text(self):
 27        return str(self.content, self.encoding)
 28
 29    def json(self):
 30        import ujson
 31        return ujson.loads(self.content)
 32
 33def request(method, url, data=None, json=None, headers={}, stream=None):
 34    try:
 35        proto, dummy, host, path = url.split("/", 3)
 36    except ValueError:
 37        proto, dummy, host = url.split("/", 2)
 38        path = ""
 39    if proto == "http:":
 40        port = 80
 41    elif proto == "https:":
 42        import ussl
 43        port = 443
 44    else:
 45        raise ValueError("Unsupported protocol: " + proto)
 46
 47    if ":" in host:
 48        host, port = host.split(":", 1)
 49        port = int(port)
 50
 51    ai = usocket.getaddrinfo(host, port, 0, usocket.SOCK_STREAM)
 52    ai = ai[0]
 53
 54    s = usocket.socket(ai[0], ai[1], ai[2])
 55    try:
 56        s.connect(ai[-1])
 57        if proto == "https:":
 58            s = ussl.wrap_socket(s, server_hostname=host)
 59        s.write(b"%s /%s HTTP/1.0rn" % (method, path))
 60        if not "Host" in headers:
 61            s.write(b"Host: %srn" % host)
 62        # Iterate over keys to avoid tuple alloc
 63        for k in headers:
 64            s.write(k)
 65            s.write(b": ")
 66            s.write(headers[k])
 67            s.write(b"rn")
 68        if json is not None:
 69            assert data is None
 70            import ujson
 71            data = ujson.dumps(json)
 72            s.write(b"Content-Type: application/jsonrn")
 73        if data:
 74            s.write(b"Content-Length: %drn" % len(data))
 75        s.write(b"rn")
 76        if data:
 77            s.write(data)
 78
 79        l = s.readline()
 80        #print(l)
 81        l = l.split(None, 2)
 82        status = int(l[1])
 83        reason = ""
 84        if len(l) > 2:
 85            reason = l[2].rstrip()
 86        while True:
 87            l = s.readline()
 88            if not l or l == b"rn":
 89                break
 90            #print(l)
 91            if l.startswith(b"Transfer-Encoding:"):
 92                if b"chunked" in l:
 93                    raise ValueError("Unsupported " + l)
 94            elif l.startswith(b"Location:") and not 200 <= status <= 299:
 95                raise NotImplementedError("Redirects not yet supported")
 96    except OSError:
 97        s.close()
 98        raise
 99
100    resp = Response(s)
101    resp.status_code = status
102    resp.reason = reason
103    return resp
104
105def head(url, **kw):
106    return request("HEAD", url, **kw)
107
108def get(url, **kw):
109    return request("GET", url, **kw)
110
111def post(url, **kw):
112    return request("POST", url, **kw)
113
114def put(url, **kw):
115    return request("PUT", url, **kw)
116
117def patch(url, **kw):
118    return request("PATCH", url, **kw)
119
120def delete(url, **kw):
121    return request("DELETE", url, **kw)