From 412a5eb9d0bf6bcc9924bb98afd6d51dcb7b2a13 Mon Sep 17 00:00:00 2001 From: Averon <77922089+TheAveron@users.noreply.github.com> Date: Sat, 31 Aug 2024 18:01:34 +0200 Subject: [PATCH] feat: typing and docstrings --- app.py | 141 ++++++++++++++++++++++++++---------- modules/database.py | 13 +++- modules/iss_info.py | 18 ++++- modules/iss_tracker.py | 80 ++++++++++++++++---- modules/reminder_service.py | 31 ++++++-- modules/tle_fetcher.py | 39 ++++++++-- modules/user_service.py | 42 ++++++++--- 7 files changed, 286 insertions(+), 78 deletions(-) diff --git a/app.py b/app.py index ce0940a..0be39f1 100644 --- a/app.py +++ b/app.py @@ -1,16 +1,18 @@ import os from datetime import datetime, timezone +from typing import Tuple, Union + from dotenv import load_dotenv -from flask import ( - Flask, flash, jsonify, redirect, render_template, request, - session, url_for -) +from flask import (Flask, flash, jsonify, redirect, render_template, request, + url_for) from flask_socketio import SocketIO -from modules import ( - init_db, start_reminder_checker, fetch_tle_data, get_logged_in_user, - register_user, verify_user, login_user, logout_user, get_current_position, - get_iss_crew, get_future_positions, get_iss_info, get_next_passes, add_reminder -) +from werkzeug import Response + +from modules import (add_reminder, fetch_tle_data, get_current_position, + get_future_positions, get_iss_crew, get_iss_info, + get_logged_in_user, get_next_passes, init_db, login_user, + logout_user, register_user, start_reminder_checker, + verify_user) # Load environment variables load_dotenv() @@ -25,15 +27,27 @@ start_reminder_checker() TLE = fetch_tle_data() + @app.route("/") -def index(): - """Render the home page with the current user's ID.""" +def index() -> str: + """ + Renders the home page with the current user's ID. + + Returns: + str: Rendered HTML template for the index page. + """ user_id = get_logged_in_user() return render_template("index.html", user_id=user_id) + @app.route("/register", methods=["GET", "POST"]) -def register(): - """Handle user registration.""" +def register() -> Union[str, Response]: + """ + Handles user registration. + + Returns: + Union[str, Response]: Rendered HTML template for the registration page or redirect to the login page. + """ if request.method == "POST": username = request.form["username"] password = request.form["password"] @@ -46,9 +60,15 @@ def register(): return render_template("register.html") + @app.route("/login", methods=["GET", "POST"]) -def login(): - """Handle user login.""" +def login() -> Union[str, Response]: + """ + Handles user login. + + Returns: + Union[str, Response]: Rendered HTML template for the login page or redirect to the index page. + """ if request.method == "POST": username = request.form["username"] password = request.form["password"] @@ -63,59 +83,106 @@ def login(): return render_template("login.html") + @app.route("/logout") -def logout(): - """Log out the current user.""" +def logout() -> Response: + """ + Logs out the current user. + + Returns: + str: Redirect to the index page. + """ logout_user() flash("You have been logged out.", "info") return redirect(url_for("index")) + @app.route("/iss-now") -def iss_now(): - """Return the current position of the ISS.""" +def iss_now() -> Tuple[Response, int]: + """ + Returns the current position of the ISS. + + Returns: + Tuple[Response, int]: JSON response containing the ISS position and HTTP status code. + """ current_position = get_current_position(TLE) - return jsonify(current_position) + return jsonify(current_position), 200 + @app.route("/iss-crew") -def iss_crew(): - """Return the current crew members on the ISS.""" +def iss_crew() -> Tuple[Response, int]: + """ + Returns the current crew members on the ISS. + + Returns: + Tuple[Response, int]: JSON response containing the ISS crew and HTTP status code. + """ iss_crew_data, status_code = get_iss_crew() return jsonify(iss_crew_data), status_code + @app.route("/future-trajectory") -def future_trajectory(): - """Return the future trajectory of the ISS.""" +def future_trajectory() -> Tuple[Response, int]: + """ + Returns the future trajectory of the ISS. + + Returns: + Tuple[Response, int]: JSON response containing the future positions and HTTP status code. + """ start_time = datetime.now(timezone.utc) - duration = float(request.args.get("duration", 3600)) # Default to 1 hour + duration = int(request.args.get("duration", 3600)) # Default to 1 hour interval = 60 # Calculate every 60 seconds future_positions = get_future_positions(TLE, start_time, duration, interval) - return jsonify(future_positions) + return jsonify(future_positions), 200 + @app.route("/iss-info") -def iss_info(): - """Return detailed information about the ISS.""" +def iss_info() -> Tuple[Response, int]: + """ + Returns detailed information about the ISS. + + Returns: + Tuple[Response, int]: JSON response containing the ISS information and HTTP status code. + """ info = get_iss_info(TLE) - return jsonify(info) + return jsonify(info), 200 + @app.route("/next-passes", methods=["GET"]) -def next_passes(): - """Return the next passes of the ISS over a given location.""" - lat = request.args.get("lat", type=float) - lon = request.args.get("lon", type=float) +def next_passes() -> Tuple[Response, int]: + """ + Returns the next passes of the ISS over a given location. + + Returns: + Tuple[Response, int]: JSON response containing the next passes and HTTP status code. + """ + lat = request.args.get("lat", type=str) + lon = request.args.get("lon", type=str) if lat is None or lon is None: return jsonify({"error": "Invalid coordinates"}), 400 passes = get_next_passes(TLE, lat, lon, num_passes=3) - return jsonify(passes) + return jsonify(passes), 200 + @app.route("/add-reminder", methods=["POST"]) -def add_reminder_route(): - """Add a reminder for a specific ISS pass time.""" +def add_reminder_route() -> Tuple[Response, int]: + """ + Adds a reminder for a specific ISS pass time. + + Returns: + Tuple[Response, int]: JSON response with status and HTTP status code. + """ user_id = request.form["user_id"] - pass_time = request.form["pass_time"] # ISO format: '2024-08-30T10:15:00Z' + pass_time_str = request.form["pass_time"] # ISO format: '2024-08-30T10:15:00Z' + + # Convert ISO 8601 string to datetime object + pass_time = datetime.fromisoformat(pass_time_str.rstrip("Z")) + add_reminder(user_id, pass_time) return jsonify({"status": "success"}), 200 + if __name__ == "__main__": app.run(debug=False) socketio.run(app) diff --git a/modules/database.py b/modules/database.py index 05a95b3..8707d1f 100644 --- a/modules/database.py +++ b/modules/database.py @@ -1,10 +1,14 @@ import sqlite3 +from typing import Optional DATABASE_FILE = "database.db" -def init_db(): + +def init_db() -> None: """ Initializes the database by creating the necessary tables if they do not exist. + + This function creates the `iss_reminders` and `users` tables if they are not already present. """ with sqlite3.connect(DATABASE_FILE) as conn: cursor = conn.cursor() @@ -35,8 +39,11 @@ def init_db(): conn.commit() -def get_db_connection(): +def get_db_connection() -> sqlite3.Connection: """ - Returns a new connection to the database. + Creates and returns a new connection to the database. + + Returns: + sqlite3.Connection: A new database connection. """ return sqlite3.connect(DATABASE_FILE) diff --git a/modules/iss_info.py b/modules/iss_info.py index e44efd5..6330c3b 100644 --- a/modules/iss_info.py +++ b/modules/iss_info.py @@ -1,10 +1,20 @@ +from typing import Any, Dict, List, Tuple + import requests -from flask import Flask, jsonify -def get_iss_crew(): +def get_iss_crew() -> Tuple[Dict[str, Any], int]: """ Fetch and return the list of crew members currently aboard the ISS. + + Returns: + Tuple[Dict[str, Any], int]: A tuple where the first element is a dictionary + containing the list of crew members with the key "crew", and the second + element is an HTTP status code. The HTTP status code is 200 for successful + requests and 500 for errors. + + Raises: + requests.RequestException: If there is an error fetching data from the API. """ try: response = requests.get("http://api.open-notify.org/astros.json") @@ -12,7 +22,9 @@ def get_iss_crew(): data = response.json() # Extract crew members aboard the ISS - iss_crew = [person for person in data.get("people", []) if person.get("craft") == "ISS"] + iss_crew: List[Dict[str, Any]] = [ + person for person in data.get("people", []) if person.get("craft") == "ISS" + ] return {"crew": iss_crew}, 200 diff --git a/modules/iss_tracker.py b/modules/iss_tracker.py index e47a03b..519cbc9 100644 --- a/modules/iss_tracker.py +++ b/modules/iss_tracker.py @@ -1,10 +1,22 @@ from datetime import datetime, timedelta, timezone +from typing import Dict, List, Tuple, Union + import ephem -def observer_setup(lat="40.7128", lon="-74.0060", elevation=10): +def observer_setup( + lat: str = "40.7128", lon: str = "-74.0060", elevation: int = 10 +) -> ephem.Observer: """ Set up an observer with given latitude, longitude, and elevation. + + Args: + lat (str): Latitude of the observer in decimal degrees. + lon (str): Longitude of the observer in decimal degrees. + elevation (int): Elevation of the observer in meters. + + Returns: + ephem.Observer: An observer object configured with the given parameters. """ observer = ephem.Observer() observer.lat = lat @@ -14,9 +26,15 @@ def observer_setup(lat="40.7128", lon="-74.0060", elevation=10): return observer -def get_current_position(tle): +def get_current_position(tle: Tuple[str, str, str]) -> Dict[str, float]: """ Get the current latitude and longitude of the satellite. + + Args: + tle (Tuple[str, str, str]): Tuple containing the TLE (Two-Line Element) data. + + Returns: + Dict[str, float]: A dictionary containing the latitude and longitude of the satellite. """ observer = observer_setup() satellite = ephem.readtle(*tle) @@ -27,9 +45,20 @@ def get_current_position(tle): } -def get_future_positions(tle, start_time, duration, interval): +def get_future_positions( + tle: Tuple[str, str, str], start_time: datetime, duration: int, interval: int +) -> List[Dict[str, float]]: """ Get satellite positions at regular intervals over a specified duration. + + Args: + tle (Tuple[str, str, str]): Tuple containing the TLE data. + start_time (datetime): The starting time for position calculations. + duration (int): Duration for which to calculate positions, in seconds. + interval (int): Interval between position calculations, in seconds. + + Returns: + List[Dict[str, float]]: A list of dictionaries, each containing latitude and longitude of the satellite. """ satellite = ephem.readtle(*tle) positions = [] @@ -37,18 +66,26 @@ def get_future_positions(tle, start_time, duration, interval): while current_time <= start_time + timedelta(seconds=duration): satellite.compute(current_time) - positions.append({ - "lat": satellite.sublat * 180.0 / ephem.pi, - "lon": satellite.sublong * 180.0 / ephem.pi, - }) + positions.append( + { + "lat": satellite.sublat * 180.0 / ephem.pi, + "lon": satellite.sublong * 180.0 / ephem.pi, + } + ) current_time += timedelta(seconds=interval) return positions -def get_iss_info(tle): +def get_iss_info(tle: Tuple[str, str, str]) -> Dict[str, Union[float, str]]: """ Get current information about the ISS including position, altitude, and speed. + + Args: + tle (Tuple[str, str, str]): Tuple containing the TLE data. + + Returns: + Dict[str, Union[float, str]]: A dictionary containing latitude, longitude, altitude, speed, and timestamp of the ISS. """ observer = observer_setup() satellite = ephem.readtle(*tle) @@ -58,14 +95,27 @@ def get_iss_info(tle): "latitude": satellite.sublat * 180.0 / ephem.pi, "longitude": satellite.sublong * 180.0 / ephem.pi, "altitude": round(satellite.elevation / 1000.0, 2), # Convert to km - "speed": round(abs(satellite.range_velocity) / 1000.0 * 3600.0, 2), # Convert to km/h + "speed": round( + abs(satellite.range_velocity) / 1000.0 * 3600.0, 2 + ), # Convert to km/h "timestamp": datetime.utcnow().isoformat() + "Z", } -def get_next_passes(tle, observer_lat, observer_lon, num_passes=3): +def get_next_passes( + tle: Tuple[str, str, str], observer_lat: str, observer_lon: str, num_passes: int = 3 +) -> List[Dict[str, str]]: """ Get the next passes of the ISS over a specified observer location. + + Args: + tle (Tuple[str, str, str]): Tuple containing the TLE data. + observer_lat (str): Latitude of the observer in decimal degrees. + observer_lon (str): Longitude of the observer in decimal degrees. + num_passes (int): Number of passes to return. + + Returns: + List[Dict[str, str]]: A list of dictionaries, each containing rise and set times of the ISS. """ observer = observer_setup(lat=observer_lat, lon=observer_lon) satellite = ephem.readtle(*tle) @@ -73,10 +123,12 @@ def get_next_passes(tle, observer_lat, observer_lon, num_passes=3): for _ in range(num_passes): next_pass = observer.next_pass(satellite) - passes.append({ - "rise_time": next_pass[0].datetime().strftime("%Y-%m-%d %H:%M:%S UTC"), - "set_time": next_pass[4].datetime().strftime("%Y-%m-%d %H:%M:%S UTC"), - }) + passes.append( + { + "rise_time": next_pass[0].datetime().strftime("%Y-%m-%d %H:%M:%S UTC"), + "set_time": next_pass[4].datetime().strftime("%Y-%m-%d %H:%M:%S UTC"), + } + ) observer.date = next_pass[4] + ephem.minute # Move time forward return passes diff --git a/modules/reminder_service.py b/modules/reminder_service.py index cf34746..b902fc6 100644 --- a/modules/reminder_service.py +++ b/modules/reminder_service.py @@ -1,13 +1,21 @@ import sqlite3 from datetime import datetime, timezone from threading import Timer +from typing import List, Tuple + from flask_socketio import emit + from .database import get_db_connection -def add_reminder(user_id, pass_time): + +def add_reminder(user_id: str, pass_time: datetime) -> None: """ Adds a reminder for a user at a specified pass time. Ensures that duplicate reminders are not added. + + Args: + user_id (str): The ID of the user to whom the reminder will be added. + pass_time (datetime): The time when the ISS pass is scheduled. """ query = """ INSERT INTO iss_reminders (user_id, pass_time) @@ -20,7 +28,8 @@ def add_reminder(user_id, pass_time): conn.execute(query, (user_id, pass_time, user_id, pass_time)) conn.commit() -def reminder_checker(): + +def reminder_checker() -> None: """ Checks for reminders that need to be triggered and notifies users. Deletes the reminder after notifying the user. @@ -34,7 +43,7 @@ def reminder_checker(): try: with get_db_connection() as conn: cursor = conn.execute(query, (now,)) - reminders = cursor.fetchall() + reminders: List[Tuple[int, str, datetime]] = cursor.fetchall() for reminder_id, user_id, pass_time in reminders: notify_user(user_id, pass_time) @@ -47,15 +56,25 @@ def reminder_checker(): # Schedule the next check after 60 seconds Timer(60, reminder_checker).start() -def start_reminder_checker(): + +def start_reminder_checker() -> None: """ Initiates the reminder checking loop. """ reminder_checker() -def notify_user(user_id, pass_time): + +def notify_user(user_id: str, pass_time: datetime) -> None: """ Sends a WebSocket notification to the user about the ISS pass time. + + Args: + user_id (str): The ID of the user to be notified. + pass_time (datetime): The time when the ISS pass is scheduled. """ - emit("reminder", {"user_id": user_id, "pass_time": pass_time}, namespace="/notifications") + emit( + "reminder", + {"user_id": user_id, "pass_time": pass_time.isoformat()}, + namespace="/notifications", + ) print(f"Notification sent for user {user_id} at {pass_time}") diff --git a/modules/tle_fetcher.py b/modules/tle_fetcher.py index 34e82a6..95d9927 100644 --- a/modules/tle_fetcher.py +++ b/modules/tle_fetcher.py @@ -1,6 +1,8 @@ import os import time from datetime import datetime, timedelta +from typing import List, Tuple + import requests # Local backup TLE data @@ -15,11 +17,24 @@ def fetch_tle_data( - url="https://www.celestrak.com/NORAD/elements/stations.txt", retries=3, delay=5 -): + url: str = "https://www.celestrak.com/NORAD/elements/stations.txt", + retries: int = 3, + delay: int = 5, +) -> Tuple[str, str, str]: """ Fetch the latest TLE data from the specified URL or use cached data if available and valid. If all fetch attempts fail, fallback to local TLE data. + + Args: + url (str): The URL from which to fetch the TLE data. + retries (int): The number of retry attempts for fetching the data. + delay (int): The delay in seconds between retry attempts. + + Returns: + Tuple[str, str, str]: A tuple containing the TLE name, line1, and line2. + + Raises: + requests.exceptions.RequestException: If there is an error fetching data from the API. """ if is_cache_valid(): return read_cached_tle() @@ -34,7 +49,10 @@ def fetch_tle_data( cache_tle_data(tle_data) return tle_data[0].strip(), tle_data[1].strip(), tle_data[2].strip() - except (requests.exceptions.HTTPError, requests.exceptions.ConnectionError) as e: + except ( + requests.exceptions.HTTPError, + requests.exceptions.ConnectionError, + ) as e: print(f"Attempt {attempt + 1} failed: {e}") if attempt < retries - 1: time.sleep(delay) @@ -43,9 +61,12 @@ def fetch_tle_data( return LOCAL_TLE["name"], LOCAL_TLE["line1"], LOCAL_TLE["line2"] -def is_cache_valid(): +def is_cache_valid() -> bool: """ Check if the cache file exists and is within the expiration period. + + Returns: + bool: True if the cache file exists and is still valid, False otherwise. """ if os.path.exists(CACHE_FILE): file_mod_time = datetime.fromtimestamp(os.path.getmtime(CACHE_FILE)) @@ -53,9 +74,12 @@ def is_cache_valid(): return False -def read_cached_tle(): +def read_cached_tle() -> Tuple[str, str, str]: """ Read the TLE data from the cache file. + + Returns: + Tuple[str, str, str]: A tuple containing the TLE name, line1, and line2. """ with open(CACHE_FILE, "r") as file: tle_data = file.readlines() @@ -64,9 +88,12 @@ def read_cached_tle(): return LOCAL_TLE["name"], LOCAL_TLE["line1"], LOCAL_TLE["line2"] -def cache_tle_data(tle_data): +def cache_tle_data(tle_data: List[str]) -> None: """ Cache the TLE data to a file. + + Args: + tle_data (List[str]): A list containing the TLE data lines. """ with open(CACHE_FILE, "w") as file: file.write("\n".join(tle_data[:3])) diff --git a/modules/user_service.py b/modules/user_service.py index 74cbca2..6dbe3f6 100644 --- a/modules/user_service.py +++ b/modules/user_service.py @@ -1,18 +1,29 @@ import sqlite3 +from typing import Optional + from flask import session from werkzeug.security import check_password_hash, generate_password_hash + from .database import get_db_connection -def register_user(username, password): + +def register_user(username: str, password: str) -> bool: """ - Registers a new user with a hashed password. Returns True if successful, False if the username already exists. + Registers a new user with a hashed password. + + Args: + username (str): The username of the new user. + password (str): The password for the new user. + + Returns: + bool: True if registration is successful, False if the username already exists. """ try: with get_db_connection() as conn: hashed_password = generate_password_hash(password) conn.execute( "INSERT INTO users (username, password) VALUES (?, ?)", - (username, hashed_password) + (username, hashed_password), ) conn.commit() return True @@ -20,9 +31,16 @@ def register_user(username, password): return False # Username already exists -def verify_user(username, password): +def verify_user(username: str, password: str) -> Optional[int]: """ - Verifies a user's credentials. Returns the user_id if successful, None otherwise. + Verifies a user's credentials. + + Args: + username (str): The username of the user. + password (str): The password for the user. + + Returns: + Optional[int]: The user ID if authentication is successful, None otherwise. """ query = "SELECT id, password FROM users WHERE username = ?" with get_db_connection() as conn: @@ -33,22 +51,28 @@ def verify_user(username, password): return None -def login_user(user_id): +def login_user(user_id: int) -> None: """ Logs in a user by setting the session user_id. + + Args: + user_id (int): The ID of the user to log in. """ session["user_id"] = user_id -def logout_user(): +def logout_user() -> None: """ Logs out the current user by removing the user_id from the session. """ session.pop("user_id", None) -def get_logged_in_user(): +def get_logged_in_user() -> Optional[int]: """ - Returns the currently logged-in user_id from the session, or None if no user is logged in. + Returns the currently logged-in user_id from the session. + + Returns: + Optional[int]: The user ID if logged in, None otherwise. """ return session.get("user_id")