Changed Base_URL to "https://api.wigle.net/api/v2/cell/search"

cellular.py

@@ -1 +1,223 @@
-# TODO: Create Cellular API version
+import datetime
+import os
+import re
+import time
+
+import pandas as pd
+import requests
+
+# Constants for coordinate validation
+MIN_LATITUDE, MAX_LATITUDE = -90, 90
+MIN_LONGITUDE, MAX_LONGITUDE = -180, 180
+
+# Constants for API requests
+BASE_URL = "https://api.wigle.net/api/v2/cell/search"
+REQUEST_DELAY = 5
+MAX_RETRIES = 3
+
+
+def is_valid_token(token):
+    """
+    Check if a token is valid.
+
+    Parameters:
+        token (str): The token to check.
+
+    Returns:
+        bool: True if the token is valid, False otherwise.
+    """
+    return bool(re.fullmatch(r"[\w\-_=]*", token))
+
+
+def get_input(prompt, validation_func):
+    """
+    Get validated user input.
+
+    Parameters:
+        prompt (str): The input prompt to display.
+        validation_func (function): The validation function to use on the input.
+
+    Returns:
+        str: The user input if it is valid.
+    """
+    while True:
+        value = input(prompt).strip()
+        if validation_func(value):
+            return value
+        print("Invalid input. Please try again.")
+
+
+def get_coordinate(prompt, min_val, max_val):
+    """
+    Get a coordinate from the user.
+
+    Parameters:
+        prompt (str): The input prompt to display.
+        min_val (float): The minimum valid value for the coordinate.
+        max_val (float): The maximum valid value for the coordinate.
+
+    Returns:
+        float: The coordinate if it is valid.
+    """
+    while True:
+        value = input(prompt)
+        try:
+            value = float(value)
+            if min_val <= value <= max_val:
+                return value
+            else:
+                print(
+                    f"Invalid input. Value should be between {min_val} and {max_val}."
+                )
+        except ValueError:
+            print("Invalid input. Please enter a number.")
+
+
+def get_coordinate_range(lower_prompt, upper_prompt, min_val, max_val):
+    """
+    Get a coordinate range from the user.
+
+    Parameters:
+        lower_prompt (str): The input prompt for the lower value of the range.
+        upper_prompt (str): The input prompt for the upper value of the range.
+        min_val (float): The minimum valid value for the coordinate.
+        max_val (float): The maximum valid value for the coordinate.
+
+    Returns:
+        tuple: The coordinate range if it is valid.
+    """
+    while True:
+        lower = get_coordinate(lower_prompt, min_val, max_val)
+        upper = get_coordinate(upper_prompt, min_val, max_val)
+        if lower < upper:
+            return lower, upper
+        print("Invalid range. The lower value should be less than the upper value.")
+
+
+def get_token(prompt):
+    """
+    Get an API token from the user.
+
+    Parameters:
+        prompt (str): The input prompt to display.
+
+    Returns:
+        dict: The Authorization header if the token is valid.
+    """
+    token = get_input(prompt, is_valid_token)
+    return {"Authorization": f"Basic {token}"}
+
+
+def send_request(headers, latrange1, latrange2, longrange1, longrange2):
+    """
+    Send a request to the WiGLE API and save the results.
+
+    Parameters:
+        headers (dict): The headers to use for the request.
+        latrange1 (float): The lower latitude of the range to request.
+        latrange2 (float): The upper latitude of the range to request.
+        longrange1 (float): The lower longitude of the range to request.
+        longrange2 (float): The upper longitude of the range to request.
+
+    Returns:
+        None
+    """
+    timestamp = datetime.datetime.now().strftime("%Y%m%d_%H%M%S")
+    filename = os.path.join("tests", f"{latrange1}_{longrange1}_{timestamp}.csv")
+
+    current_count = 0
+    total_count = None
+    retries = 0
+
+    query_params = {
+        "latrange1": latrange1,
+        "latrange2": latrange2,
+        "longrange1": longrange1,
+        "longrange2": longrange2,
+        "first": 100,
+    }
+
+    print("Retrieving results...")
+
+    with open(filename, "a") as f:
+        while retries < MAX_RETRIES:
+            try:
+                response = requests.get(BASE_URL, params=query_params, headers=headers)
+                status_code = response.status_code
+
+                if status_code == 200:
+                    data = response.json()
+                    results = data["results"]
+                    current_count += len(results)
+
+                    if total_count is None:
+                        total_count = data["totalResults"]
+
+                    print(f"Retrieved {current_count} out of {total_count} results")
+
+                    df = pd.DataFrame(results)
+                    df.to_csv(f, index=False, header=f.tell() == 0, escapechar="\\")
+
+                    if current_count >= total_count:
+                        break
+
+                    query_params["start"] = current_count
+                    time.sleep(REQUEST_DELAY)
+
+                elif status_code == 401:
+                    print("Unauthorized. Check your API token.")
+                    break
+
+                elif status_code == 429:
+                    print("Too many requests. Saving the current results.")
+                    break
+
+                else:
+                    print(f"Error {status_code}: Failed to retrieve results.")
+                    break
+
+            except requests.exceptions.Timeout:
+                print("Request timed out. Retrying...")
+                retries += 1
+                time.sleep(REQUEST_DELAY)
+
+            except requests.exceptions.TooManyRedirects:
+                print("Too many redirects. Terminating request.")
+                break
+
+            except requests.exceptions.ConnectionError:
+                print("Connection Error. Retrying...")
+                retries += 1
+                time.sleep(REQUEST_DELAY)
+
+            except Exception as e:
+                print(f"An error occurred: {e}")
+                break
+
+        print(f"All results retrieved. Total results: {current_count}")
+
+
+def main():
+    """
+    Get the latitude and longitude ranges and API token from the user,
+    and send a request to the WiGLE API.
+
+    Returns:
+        None
+    """
+    lat_range_1, lat_range_2 = get_coordinate_range(
+        "Enter lower latitude: ", "Enter upper latitude: ", MIN_LATITUDE, MAX_LATITUDE
+    )
+    long_range_1, long_range_2 = get_coordinate_range(
+        "Enter lower longitude: ",
+        "Enter upper longitude: ",
+        MIN_LONGITUDE,
+        MAX_LONGITUDE,
+    )
+    headers = get_token("Enter your API token: ")
+
+    send_request(headers, lat_range_1, lat_range_2, long_range_1, long_range_2)
+
+
+if __name__ == "__main__":
+    main()