-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
refactor: refacoring for cleaner code
- Loading branch information
Showing
10 changed files
with
248 additions
and
234 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,33 +1,42 @@ | ||
import sqlite3 | ||
|
||
DATABASE_FILE = "database.db" | ||
|
||
def init_db(): | ||
conn = sqlite3.connect("database.db") | ||
cursor = conn.cursor() | ||
cursor.execute( | ||
""" | ||
CREATE TABLE IF NOT EXISTS iss_reminders ( | ||
id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
user_id INTEGER, | ||
pass_time DATETIME, | ||
notified BOOLEAN DEFAULT 0 | ||
) | ||
""" | ||
) | ||
Initializes the database by creating the necessary tables if they do not exist. | ||
""" | ||
with sqlite3.connect(DATABASE_FILE) as conn: | ||
cursor = conn.cursor() | ||
|
||
cursor.execute( | ||
""" | ||
CREATE TABLE IF NOT EXISTS users ( | ||
id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
username TEXT UNIQUE NOT NULL, | ||
password TEXT NOT NULL | ||
# Create iss_reminders table | ||
cursor.execute( | ||
""" | ||
CREATE TABLE IF NOT EXISTS iss_reminders ( | ||
id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
user_id INTEGER, | ||
pass_time DATETIME, | ||
notified BOOLEAN DEFAULT 0 | ||
) | ||
""" | ||
) | ||
""" | ||
) | ||
conn.commit() | ||
conn.close() | ||
|
||
# Create users table | ||
cursor.execute( | ||
""" | ||
CREATE TABLE IF NOT EXISTS users ( | ||
id INTEGER PRIMARY KEY AUTOINCREMENT, | ||
username TEXT UNIQUE NOT NULL, | ||
password TEXT NOT NULL | ||
) | ||
""" | ||
) | ||
|
||
conn.commit() | ||
|
||
|
||
def get_db_connection(): | ||
conn = sqlite3.connect("database.db") | ||
return conn | ||
""" | ||
Returns a new connection to the database. | ||
""" | ||
return sqlite3.connect(DATABASE_FILE) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -1,97 +1,82 @@ | ||
from datetime import UTC, datetime, timedelta | ||
|
||
from datetime import datetime, timedelta, timezone | ||
import ephem | ||
|
||
|
||
def observer_setup(lat="40.7128", lon="-74.0060"): | ||
# Create an Observer object | ||
def observer_setup(lat="40.7128", lon="-74.0060", elevation=10): | ||
""" | ||
Set up an observer with given latitude, longitude, and elevation. | ||
""" | ||
observer = ephem.Observer() | ||
|
||
# Set the observer's location | ||
observer.lat = lat | ||
observer.lon = lon | ||
observer.elevation = 10 | ||
|
||
observer.date = datetime.now(UTC) | ||
|
||
observer.elevation = elevation | ||
observer.date = datetime.now(timezone.utc) | ||
return observer | ||
|
||
|
||
def get_current_position(tle): | ||
satellite = ephem.readtle(tle[0], tle[1], tle[2]) | ||
satellite.compute(observer_setup()) | ||
""" | ||
Get the current latitude and longitude of the satellite. | ||
""" | ||
observer = observer_setup() | ||
satellite = ephem.readtle(*tle) | ||
satellite.compute(observer) | ||
return { | ||
"lat": satellite.sublat * 180.0 / ephem.pi, | ||
"lon": satellite.sublong * 180.0 / ephem.pi, | ||
} | ||
|
||
|
||
def get_future_positions(tle, start_time, duration, interval): | ||
satellite = ephem.readtle(tle[0], tle[1], tle[2]) | ||
""" | ||
Get satellite positions at regular intervals over a specified duration. | ||
""" | ||
satellite = ephem.readtle(*tle) | ||
positions = [] | ||
current_time = start_time | ||
|
||
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): | ||
# Create a satellite object from the TLE data | ||
satellite = ephem.readtle(tle[0], tle[1], tle[2]) | ||
|
||
# Compute the current position of the satellite | ||
satellite.compute(observer_setup()) # Use the current UTC time | ||
|
||
# Get latitude and longitude of the ISS | ||
latitude = satellite.sublat * 180.0 / ephem.pi # Convert from radians to degrees | ||
longitude = satellite.sublong * 180.0 / ephem.pi # Convert from radians to degrees | ||
""" | ||
Get current information about the ISS including position, altitude, and speed. | ||
""" | ||
observer = observer_setup() | ||
satellite = ephem.readtle(*tle) | ||
satellite.compute(observer) | ||
|
||
# Calculate altitude in kilometers | ||
altitude_km = satellite.elevation / 1000.0 # ephem returns elevation in meters | ||
|
||
# Calculate speed in km/h | ||
speed_kmh = abs( | ||
satellite.range_velocity / 1000.0 * 3600.0 | ||
) # ephem returns range_velocity in m/s | ||
|
||
# Return the data as JSON | ||
return { | ||
"latitude": latitude, | ||
"longitude": longitude, | ||
"altitude": round(altitude_km, 2), | ||
"speed": round(speed_kmh, 2), | ||
"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 | ||
"timestamp": datetime.utcnow().isoformat() + "Z", | ||
} | ||
|
||
|
||
def get_next_passes(tle, observer_lat, observer_lon, num_passes=3): | ||
iss = ephem.readtle(tle[0], tle[1], tle[2]) | ||
|
||
Paris = { | ||
"lat": "48.864716", | ||
"Lon": "2.349014", | ||
} | ||
observer = observer_setup(Paris["lat"], Paris["Lon"]) | ||
|
||
""" | ||
Get the next passes of the ISS over a specified observer location. | ||
""" | ||
observer = observer_setup(lat=observer_lat, lon=observer_lon) | ||
satellite = ephem.readtle(*tle) | ||
passes = [] | ||
|
||
for _ in range(num_passes): | ||
next_pass = observer.next_pass(iss) | ||
print(next_pass) | ||
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"), | ||
} | ||
) | ||
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"), | ||
}) | ||
observer.date = next_pass[4] + ephem.minute # Move time forward | ||
|
||
return passes |
Oops, something went wrong.