feat: add HTTP examples (#2)

electricalgorithm · web-flow · commit 773113682f94 · 2022-07-28T10:07:09.000+03:00
diff --git a/examples/http_requests_example.py b/examples/http_requests_example.py
@@ -0,0 +1,110 @@
+"""
+Example code for publising data to any server with using HTTP methods.
+"""
+
+import json
+import time
+
+from core.modem import Modem
+from core.status import Status
+
+def handle_cellular_network_connection():
+    """This function handles the connection to cellular network connection.
+    """
+    # Ask for return from the modem.
+    modem.check_modem_communication()
+    # Send APN details.
+    modem.set_modem_apn(cid=1, apn="super")
+    # Try to get a operator response for 10 times.
+    for _index in range(0, 10):
+        # Check if there is a connection to any operator.
+        response = modem.get_operator_information()
+        if response["status"] == Status.SUCCESS:
+            return
+        # Wait for 1 second.
+        time.sleep(1)
+
+def prepare_http_connection():
+    """This function prepares HTTP connection for modem.
+    """
+    # Set the first HTTP context.
+    modem.set_modem_http_context_id()
+    # Activate PDP.
+    modem.activate_pdp_context()
+
+def prepare_http_with_query(host_url, query_string=""):
+    """Sets the modem's HTTP server address.
+
+    Args:
+        host_url (str): Server address to send request.
+        query_string (str, optional): Query payload for the server. Defaults to "".
+    """
+    # Add query to the server URL.
+    url_to_post = host_url + "/?" + query_string
+    # Set the HTTP url.
+    modem.set_modem_http_server_url(url=url_to_post)
+
+def send_post_request(server_url, data_dict, query_string=""):
+    """POST request to the HTTP server with a payload and optional query.
+
+    Args:
+        server_url (str): Server address to send request.
+        data_dict (dict): The data for the request as dictionary.
+        query_string (str, optional): Query payload for the server. Defaults to "".
+    """
+    # Configure the HTTP address.
+    prepare_http_with_query(server_url, query_string)
+    # Convert to data to JSON.
+    data_to_post = json.dumps(data_dict)
+    # Send a post request to the URL.
+    print("POST Request: ", modem.http_post_request(data=data_to_post))
+    # Wait for six seconds before the next operation.
+    time.sleep(6)
+
+def send_get_request(server_url, query_string=""):
+    """GET request to the HTTP server with optional query.
+
+    Args:
+        server_url (str): Server address to send request.
+        query_string (str, optional): Query payload for the server. Defaults to "".
+    """
+    # Configure the HTTP address.
+    prepare_http_with_query(server_url, query_string)
+    # Send a GET request.
+    print("GET Request: ", modem.http_get_request())
+    # Wait for six seconds before the next operation.
+    time.sleep(6)
+
+def send_put_request(server_url, data_dict, query_string=""):
+    """PUT request to the HTTP server with a payload and optional query.
+
+    Args:
+        server_url (str): Server address to send request.
+        data_dict (dict): The data for the request as dictionary.
+        query_string (str, optional): Query payload for the server. Defaults to "".
+    """
+    # Configure the HTTP address.
+    prepare_http_with_query(server_url, query_string)
+    # Convert to data to JSON.
+    data_to_post = json.dumps(data_dict)
+    # Send a PUT request.
+    print("PUT Request: ", modem.http_put_request(data_to_post))
+    # Wait for six seconds before the next operation.
+    time.sleep(6)
+
+
+if __name__ == "__main__":
+    # Initilizate the modem.
+    modem = Modem({})
+    # Connect to a cellular network.
+    handle_cellular_network_connection()
+    # Prepare for the HTTP context.
+    prepare_http_connection()
+    # Select a HOST for testing purposes.
+    HOST = "https://webhook.site/6112efa6-4f8c-4bdb-8fa3-5787cc02f80a"
+    # Send a POST request to the server.
+    send_post_request(HOST, {"ProjectTopic": "PicocellHTTPExample"}, "device=Pico&try=0")
+    # Send a GET request to the server.
+    send_get_request(HOST, "company=Sixfab&course=Picocell")
+    # Send PUT request.
+    send_put_request(HOST, {"Temperature(*C)": 28}, "sensor=Garden1")
diff --git a/examples/http_requests_example_with_manager.py b/examples/http_requests_example_with_manager.py
@@ -0,0 +1,265 @@
+"""
+Example code for publising data to any server with using HTTP methods by using manager class.
+"""
+
+import json
+import time
+
+from core.modem import Modem
+from core.status import Status
+from core.manager import StateManager, Step
+
+def step_for_setting_server_url(modem_object, next_function, url=None, query=""):
+    """This function returns a step for setting server's URL with a query.
+
+    Args:
+        modem_object (Modem): The modem instance to call methods.
+        next_function (string): Next step's name if the step got success.
+        url (str, optional): URL for the host. Defaults to None.
+        query (str, optional): Query for the server. Defaults to "".
+
+    Returns:
+        Step: The step instance for setting server URL in the modem.
+    """
+
+    url_with_query = url + "/?" + query
+    # Set-up the steps for state manager.
+    step_set_server_url = Step(
+        function=modem_object.set_modem_http_server_url,
+        name="set_server_url",
+        success=next_function,
+        fail="failure",
+        function_params={"url": url_with_query},
+        cachable=True
+    )
+
+    return step_set_server_url
+
+
+def prepare_communication_for_http(modem_object):
+    """A function to register the cellular network
+    and activate the PDP.
+
+    Args:
+        modem_object (Modem): The modem instance to call methods.
+
+    Returns:
+        dict: A dictionary which has the results inside.
+    """
+
+    # Set-up the steps for state manager.
+    step_network_reg = Step(
+        function=modem_object.register_network,
+        name="register_network",
+        success="pdp_deactivate",
+        fail="failure"
+    )
+
+    step_pdp_deactivate = Step(
+        function=modem_object.deactivate_pdp_context,
+        name="pdp_deactivate",
+        success="pdp_activate",
+        fail="failure"
+    )
+
+    step_pdp_activate = Step(
+        function=modem_object.activate_pdp_context,
+        name="pdp_activate",
+        success="success",
+        fail="failure",
+        cachable=True
+    )
+
+    # Create a state for this function to put
+    # modem's cache.
+    function_name = "prepare_communication_for_http"
+
+    # Create the state manager.
+    state_manager = StateManager(first_step=step_network_reg,
+                                cache=modem_object.cache,
+                                function_name=function_name)
+
+    # Add each step.
+    state_manager.add_step(step_network_reg)
+    state_manager.add_step(step_pdp_deactivate)
+    state_manager.add_step(step_pdp_activate)
+
+    # Run the loop to finish state manager.
+    while True:
+        result = state_manager.run()
+        if result["status"] == Status.SUCCESS:
+            return result
+        elif result["status"] == Status.ERROR:
+            return result
+        time.sleep(result["interval"])
+
+
+def post_message_to_http_server(modem_object, payload, url=None, query=""):
+    """POST request to the HTTP server with a payload and optional query.
+
+    Args:
+        modem_object (Modem): The modem instance to call methods.
+        payload (dict): The dictionary message which will be sent.
+        url (str, optional): URL for the host. Defaults to None.
+        query (str, optional): Query for the server. Defaults to "".
+
+    Returns:
+        dict: A dictionary which has the results inside.
+    """
+
+    # Set-up the steps for state manager.
+    step_set_server_url = step_for_setting_server_url(
+        modem_object, "post_request", url, query)
+
+    step_post_request = Step(
+        function=modem_object.http_post_request,
+        name="post_request",
+        success="success",
+        fail="failure",
+        function_params={"data": payload},
+        cachable=True
+    )
+
+    # Create a state for this function to put
+    # modem's cache.
+    function_name = "post_message_to_http_server"
+
+    # Create the state manager.
+    state_manager = StateManager(first_step=step_set_server_url,
+                                cache=modem_object.cache,
+                                function_name=function_name)
+
+    # Add each step.
+    state_manager.add_step(step_set_server_url)
+    state_manager.add_step(step_post_request)
+
+    # Run the loop to finish state manager.
+    while True:
+        result = state_manager.run()
+        if result["status"] == Status.SUCCESS:
+            return result
+        elif result["status"] == Status.ERROR:
+            return result
+        time.sleep(result["interval"])
+
+
+def get_message_to_http_server(modem_object, url=None, query=""):
+    """GET request to the HTTP server with optional query.
+
+    Args:
+        modem_object (Modem): The modem instance to call methods.
+        url (str, optional): URL for the host. Defaults to None.
+        query (str, optional): Query for the server. Defaults to "".
+
+    Returns:
+        dict: A dictionary which has the results inside.
+    """
+
+    # Set-up the steps for state manager.
+    step_set_server_url = step_for_setting_server_url(
+        modem_object, "get_request", url, query)
+
+    step_get_request = Step(
+        function=modem_object.http_get_request,
+        name="get_request",
+        success="success",
+        fail="failure",
+        cachable=True
+    )
+
+    # Create a state for this function to put
+    # modem's cache.
+    function_name = "get_message_to_http_server"
+
+    # Create the state manager.
+    state_manager = StateManager(first_step=step_set_server_url,
+                                cache=modem_object.cache,
+                                function_name=function_name)
+
+    # Add each step.
+    state_manager.add_step(step_set_server_url)
+    state_manager.add_step(step_get_request)
+
+    # Run the loop to finish state manager.
+    while True:
+        result = state_manager.run()
+        if result["status"] == Status.SUCCESS:
+            return result
+        elif result["status"] == Status.ERROR:
+            return result
+        time.sleep(result["interval"])
+
+
+def put_message_to_http_server(modem_object, payload, url=None, query=""):
+    """PUT request to the HTTP server with a payload and optional query.
+
+    Args:
+        modem_object (Modem): The modem instance to call methods.
+        payload (dict): The dictionary message which will be sent.
+        url (str, optional): URL for the host. Defaults to None.
+        query (str, optional): Query for the server. Defaults to "".
+
+    Returns:
+        dict: A dictionary which has the results inside.
+    """
+
+    # Set-up the steps for state manager.
+    step_set_server_url = step_for_setting_server_url(
+        modem_object, "put_request", url, query)
+
+    step_put_request = Step(
+        function=modem_object.http_put_request,
+        name="put_request",
+        success="success",
+        fail="failure",
+        function_params={"data": payload},
+        cachable=True
+    )
+
+    # Create a state for this function to put
+    # modem's cache.
+    function_name = "put_message_to_http_server"
+
+    # Create the state manager.
+    state_manager = StateManager(first_step=step_set_server_url,
+                                cache=modem_object.cache,
+                                function_name=function_name)
+
+    # Add each step.
+    state_manager.add_step(step_set_server_url)
+    state_manager.add_step(step_put_request)
+
+    # Run the loop to finish state manager.
+    while True:
+        result = state_manager.run()
+        if result["status"] == Status.SUCCESS:
+            return result
+        elif result["status"] == Status.ERROR:
+            return result
+        time.sleep(result["interval"])
+
+
+if __name__ == "__main__":
+    # Initilize the modem object.
+    modem = Modem({})
+    # Prepare communication for HTTP.
+    prepare_communication_for_http(modem)
+    # Server address to send requests.
+    HOST = "https://webhook.site/6112efa6-4f8c-4bdb-8fa3-5787cc02f80a"
+    # Prepare a payload to post into the server.
+    payload_to_post = {"ProjectTopic": "PicocellHTTPExampleWithManager"}
+    payload_to_post_as_json = json.dumps(payload_to_post)
+    # Post it to the server with given query.
+    post_message_to_http_server(
+        modem, payload_to_post_as_json, HOST, "device=Pico&try=0")
+    time.sleep(4)
+    # Get a request from the server.
+    get_message_to_http_server(modem, HOST, "company=Sixfab&course=Picocell")
+    time.sleep(6)
+    # Prepare a payload to put into the server.
+    payload_to_put = {"Temperature(*C)": 28}
+    payload_to_put_as_json = json.dumps(payload_to_put)
+    # Send the request into the server.
+    put_message_to_http_server(
+        modem, payload_to_put_as_json, HOST, "sensor=Garden2")
+    time.sleep(4)
