feat: HTTP header_mode implement, add example (#8)

Author: electricalgorithm

core/modules/http.py

@@ -251,16 +251,22 @@ def set_server_url(self, url=None ,timeout=5):
             return result
         return {"response": "Missing arguments : url", "status": Status.ERROR}
 
-    def get(self, header_mode=0, timeout=60):
+    def get(self, data="", header_mode=0, input_timeout=5, timeout=60):
         """
         Function for sending HTTP GET request
 
         Parameters
         ----------
+        data : str
+            The full header for the GET request if header_mode is set.
         header_mode : int
             Customization of HTTP(S) request header)(default=0)
                 0 --> Disable
                 1 --> Enable
+        input_timeout : int
+            Input timeout (default=5)
+        timeout : int
+            Timeout (default=60)
 
         Returns
         -------
@@ -276,11 +282,23 @@ def get(self, header_mode=0, timeout=60):
             status : int
                 Status of the command.
         """
-        if header_mode == 1:
-            return {"response": "Not implemented yet!", "status": Status.ERROR}
-
-        command = f'AT+QHTTPGET={timeout}'
-        return self.atcom.send_at_comm(command,"OK")
+        # Set the request header config.
+        result = self.set_request_header_status(status=header_mode)
+        if result["status"] == Status.SUCCESS:
+            if header_mode == 1:
+                # Send a GET request to the modem.
+                command = f'AT+QHTTPGET={timeout},{len(data)},{input_timeout}'
+                result =  self.atcom.send_at_comm(command, "CONNECT", timeout=60)
+                if result["status"] == Status.SUCCESS:
+                    # Send the request header.
+                    return self.atcom.send_at_comm(data, "OK", line_end=False)
+            else:
+                # Send a GET request without header.
+                command = f'AT+QHTTPGET={timeout}'
+                return self.atcom.send_at_comm(command,"OK")
+
+        # Return the result of request header if there is no SUCCESS.
+        return result
 
     def post(self, data, header_mode=0, input_timeout=5, timeout=60):
         """
@@ -314,14 +332,15 @@ def post(self, data, header_mode=0, input_timeout=5, timeout=60):
             status : int
                 Status of the command.
         """
-        if header_mode == 1:
-            return {"response": "Not implemented yet!", "status": Status.ERROR}
-
-        command = f'AT+QHTTPPOST={len(data)},{input_timeout},{timeout}'
-        result =  self.atcom.send_at_comm(command,"CONNECT", timeout=timeout)
-
+        # Set the request header config.
+        result = self.set_request_header_status(status=header_mode)
         if result["status"] == Status.SUCCESS:
-            result = self.atcom.send_at_comm(data, "OK", line_end=False) # send data
+            # Send a POST request to the modem.
+            command = f'AT+QHTTPPOST={len(data)},{input_timeout},{timeout}'
+            result =  self.atcom.send_at_comm(command,"CONNECT", timeout=timeout)
+            if result["status"] == Status.SUCCESS:
+                # Send the request (header and) body.
+                result = self.atcom.send_at_comm(data, "OK", line_end=False)
         return result
 
     def post_from_file(self, file_path, header_mode=0, timeout=60):
@@ -390,14 +409,15 @@ def put(self, data, header_mode=0, input_timeout=5, timeout=60):
             status : int
                 Status of the command.
         """
-        if header_mode == 1:
-            return {"response": "Not implemented yet!", "status": Status.ERROR}
-
-        command = f'AT+QHTTPPUT={len(data)},{input_timeout},{timeout}'
-        result = self.atcom.send_at_comm(command,"CONNECT")
-
+        # Set the request header config.
+        result = self.set_request_header_status(status=header_mode)
         if result["status"] == Status.SUCCESS:
-            result = self.atcom.send_at_comm(data, "OK", line_end=False) # send data
+            # Send a PUT request to the modem.
+            command = f'AT+QHTTPPUT={len(data)},{input_timeout},{timeout}'
+            result = self.atcom.send_at_comm(command,"CONNECT")
+            if result["status"] == Status.SUCCESS:
+                # Send the request (header and) body.
+                result = self.atcom.send_at_comm(data, "OK", line_end=False)
         return result
 
     def put_from_file(self, file_path, file_type=0, header_mode=0, timeout=60):

examples/http_custom_header_example.py

@@ -0,0 +1,91 @@
+import json
+import time
+from core.modem import Modem
+from core.utils.debug import Debug
+from core.utils.status import Status
+
+
+# Provide your server details to send HTTP
+# requests. For testing purposes, you may
+# use "webhook.site".
+# Example:
+#     Host: https://webhook.site
+#     Query: /6112efa6-4f8c-4bdb-8fa3-5787cc02f80a?query_param=DATA
+SERVER_DETAILS = {
+    'HOST': '[YOUR_SERVER_DOMAIN_ADDRESS]',
+    'QUERY': '[YOUR_REQUEST_QUERY_ADDRESS]'
+}
+
+# This messages are arbitrary, you can change
+# both attribute and data names.
+DATA_TO_POST = {
+    'messageType': 'picocell_status',
+    'message': 'Picocell can connect to the server.'
+}
+
+DATA_TO_PUT = {
+    'messageType': 'picocell_battery',
+    'message': '90%'
+}
+
+
+# Initilaze the instances for the Picocell SDK.
+modem = Modem()
+debug = Debug()
+
+# Send APN details.
+modem.network.set_apn()
+
+# Check the network registration.
+result = modem.network.check_network_registration()
+if result["status"] != Status.SUCCESS:
+    debug.error("Could not connected to the cellular network.")
+    
+
+# Set the first HTTP context.
+modem.http.set_context_id()
+
+# Activate PDP.
+modem.network.deactivate_pdp_context()
+modem.network.activate_pdp_context()
+
+# Set server URL.
+debug.info("Set Server URL: ", modem.http.set_server_url(SERVER_DETAILS["HOST"]))
+
+# Send a POST request.
+data_post_json = json.dumps(DATA_TO_POST)
+header =    "POST " + SERVER_DETAILS["QUERY"] + " HTTP/1.1\n" + \
+            "Host: " + SERVER_DETAILS["HOST"][8:] + "\n" + \
+            "Custom-Header-Name: Custom-Data\n" + \
+            "Content-Type: application/json\n" + \
+            "Content-Length: " + str(len(data_post_json) + 1) + "\n" + \
+            "\n\n"
+
+debug.info("Send POST Request: ", modem.http.post(data=header+data_post_json, header_mode=1))
+
+# Wait for the modem to be ready.
+time.sleep(5)
+
+# Send a GET request.
+header =    "GET " + SERVER_DETAILS["QUERY"] + " HTTP/1.1\n" + \
+            "Host: " + SERVER_DETAILS["HOST"][8:] + "\n" + \
+            "Content-Type: text/plain\n" + \
+            "Content-Length: 0\n" + \
+            "Custom-Header-Name: Custom-Data\n" + \
+            "\n\n"
+
+debug.info("Send GET Request: ", modem.http.get(data=header, header_mode=1))
+
+# Wait for the modem to be ready.
+time.sleep(5)
+
+# Send a PUT request.
+data_put_json = json.dumps(DATA_TO_PUT)
+header =    "PUT " + SERVER_DETAILS["QUERY"] + " HTTP/1.1\n" + \
+            "Host: " + SERVER_DETAILS["HOST"][8:] + "\n" + \
+            "Custom-Header-Name: Custom-Data\n" + \
+            "Content-Type: application/json\n" + \
+            "Content-Length: " + str(len(data_put_json) + 1) + "\n" + \
+            "\n\n"
+
+debug.info("Send PUT Request: ", modem.http.put(data=header+data_put_json, header_mode=1))