Skip to content
This repository has been archived by the owner on Jan 27, 2023. It is now read-only.

Commit

Permalink
Update linter to ignore auto-generate kaitai class
Browse files Browse the repository at this point in the history 
Update module init to not import any submodules
Update package details
Update main to only work as a standalone run-once information fetcher
Update submodule import patterns to not implicitly import global cosi
Add OreFlat0 kaitastruct and autogenerated kaitai class decoder
Remove old and misplaced fsk test data
  • Loading branch information
@dmitri-mcguckin
dmitri-mcguckin committed Mar 10, 2021
1 parent ceee05d commit d0ef508
Show file tree
Hide file tree
Showing 12 changed files with 339 additions and 249 deletions.
2 changes: 1 addition & 1 deletion .github/workflows/unit-tests.yaml
View file
Original file line number Diff line number Diff line change
Expand Up @@ -27,7 +27,7 @@ jobs:
- name: Lint with flake8
run: |
flake8 cosi --count --select=E9,F63,F7,F82 --show-source --statistics
flake8 cosi --count --exclude csim.py --exit-zero --max-complexity=30 --max-line-length=127 --statistics
flake8 cosi --count --exclude 'cosi/structs/*' --exit-zero --max-complexity=30 --max-line-length=127 --statistics
- name: Run unit tests
run: pytest tests/*
42 changes: 19 additions & 23 deletions cosi/__init__.py
View file
Original file line number Diff line number Diff line change
@@ -1,45 +1,41 @@
import os
import datetime
from datetime import timedelta

# Version Details
MAJOR = 0
MAJOR = 1
MINOR = 0
PATCH = 2
PATCH = 0

# Application details
# Application information
APP_NAME = 'uniclogs-cosi'
APP_VERSION = f'{MAJOR}.{MINOR}.{PATCH}'
APP_AUTHOR = 'Dmitri McGuckin'
APP_LICENSE = 'GPLv3'
APP_DESCRIPTION = 'Cosmos Satnogs/SpaceTrack Interface'
APP_DESCRIPTION = 'Cosmos Satnogs/SpaceTrack Interface. An application for' \
' fetching the latest relevant satellite metadata and' \
' telemetry from SatNOGSs\' and Space-Tracks\' API\'s'
APP_URL = 'https://github.com/oresat/uniclogs-cosi'

# Dart DB constants
DB_HOST = os.getenv('DART_HOST')
try:
DB_PORT = int(os.getenv('DART_PORT'))
except TypeError:
DB_PORT = 5432
DB_PORT = int(os.getenv('DART_PORT', '5432'))
DB_NAME = os.getenv('DART_DB')
DB_USERNAME = os.getenv('DART_USERNAME')
DB_PASSWORD = os.getenv('DART_PASSWORD')

# Satnogs constants
SATNOGS_TOKEN = os.getenv('SATNOGS_TOKEN')

SATNOGS_URL = 'https://db.satnogs.org'
SATNOGS_SATELITE = SATNOGS_URL + '/api/satellites/{}/?format=json'
SATNOGS_TELEMETRY = SATNOGS_URL + '/api/telemetry/?satellite={}&format=json'
SATNOGS_TOKEN = os.getenv('SATNOGS_DB_TOKEN')
SATNOGS_API = 'https://db-dev.satnogs.org/api'
SATNOGS_SATELITE_ENDPOINT = f'{SATNOGS_API}/satellites/''{}/?format=json'
SATNOGS_TELEMETRY_ENDPOINT = f'{SATNOGS_API}/telemetry/?satellite=' \
'{}&format=json'

# Spacetrack (18th-Space) constants
SPACETRACK_USERNAME = os.getenv('SPACETRACK_USERNAME')
SPACETRACK_PASSWORD = os.getenv('SPACETRACK_PASSWORD')

SPACETRACK_URL = "https://www.space-track.org"
SPACETRACK_LOGIN = SPACETRACK_URL + "/ajaxauth/login"
SPACETRACK_TLE = SPACETRACK_URL \
+ "/basicspacedata/query/class/tle/NORAD_CAT_ID/{}\
/predicates/TLE_LINE0,TLE_LINE1,TLE_LINE2/limit/1/format/json"


STALE_TLE_DURATION = datetime.timedelta(days=1)
SPACETRACK_API = 'https://www.space-track.org'
SPACETRACK_LOGIN_ENDPOINT = f'{SPACETRACK_API}/ajaxauth/login'
SPACETRACK_TLE_ENDPOINT = f'{SPACETRACK_API}/basicspacedata/query/class/tle/' \
'NORAD_CAT_ID/{}/predicates/TLE_LINE0,TLE_LINE1,' \
'TLE_LINE2/limit/1/format/json'
STALE_TLE_TIMEOUT = timedelta(days=1)
251 changes: 73 additions & 178 deletions cosi/__main__.py
View file
Original file line number Diff line number Diff line change
@@ -1,193 +1,88 @@
import re
import time
import datetime
import json
import argparse
import cosi as c
import sqlalchemy
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
import cosi.models as m
import cosi.satnogs as satnogs
import cosi.spacetrack as spacetrack

Base = declarative_base()
db_url = 'postgresql://{}:{}@{}:{}/{}'.format(c.DB_USERNAME,
c.DB_PASSWORD,
c.DB_HOST,
c.DB_PORT,
c.DB_NAME)
engine = create_engine(db_url)
DartSession = sqlalchemy.orm.sessionmaker(bind=engine)


def parse_poll_interval(poll_interval: str) -> int:
"""Takes the arguments list and returns the total seconds CoSI needs to
wait before polling spacetrack and satnogs again.
Parameters
----------
* args: `list` A python list of arguments pre-parsed by arg-parse
Returns
-------
`int`: Time in seconds for CoSI to sleep
Raises
------
`ValueError`: Raises this if the user-input for `--poll-interval`
does not match the patter `[0-9]+(d|h|m|s)`
"""
if(re.match('[0-9]+d', poll_interval) is not None):
timedelta = datetime.timedelta(days=int(poll_interval[:-1]))
elif(re.match('[0-9]+h', poll_interval) is not None):
timedelta = datetime.timedelta(hours=int(poll_interval[:-1]))
elif(re.match('[0-9]+m', poll_interval) is not None):
timedelta = datetime.timedelta(minutes=int(poll_interval[:-1]))
elif(re.match('[0-9]+s', poll_interval) is not None):
timedelta = datetime.timedelta(seconds=int(poll_interval[:-1]))
else:
raise ValueError(f'Bad poll interval value: {poll_interval}.\n\tPoll'
' interval must follow pattern: [0-9]+(d|h|m|s)'
'\n\tEx: --poll-interval 12h or --poll-interval 30m')
return timedelta.total_seconds()


def fetch_tle(args: list):
tle = spacetrack.request_tle(args.norad_id)
tle = m.TLE(header_text=tle.get('TLE_LINE0'),
first_line=tle.get('TLE_LINE1'),
second_line=tle.get('TLE_LINE2'))
if(args.debug):
print('\nInserting {} into db...'.format(tle))
m.inject_tle(tle)


def fetch_satellite(args: list):
satellite = satnogs.request_satellite(args.norad_id)
if(args.debug):
print('\nSatellite info: {}'.format(satellite))


def fetch_telemetry(args: list):
# Get latest telemetry
encoded = satnogs.request_telemetry(args.norad_id)
frame = bytearray.fromhex(encoded.get('frame'))
if(args.debug):
print('\nEncoded telemetry: {}'.format(encoded))

# Decode telemetry
decoded = satnogs.decode_telemetry_frame(frame)
if(args.debug):
print('\nDecoded telemetry: {}'.format(decoded))

# Send data to the db
try:
magentometer = m.Telemetry(invalid_count=decoded.mag_invalid_count,
sensor_used=decoded.mag_sensor_used,
vector_body_1=decoded.mag_vector_body1,
vector_body_2=decoded.mag_vector_body2,
vector_body_3=decoded.mag_vector_body3,
vector_valid=decoded.mag_vector_valid)
except AttributeError as e:
print('\nFailed to decode telemetry: {}'.format(e))
return

try:
if(args.debug):
print('Injecting telemetry into the DB: {}'
.format(magentometer))
status = m.inject_telemetry(magentometer)
if(args.debug and status):
print('\nSuccesfully injected!')
else:
raise Exception('Failed to inject telemetry')
except Exception:
if(args.debug):
print('\nFailed to inject telemetry frame into DB: {}'
.format(magentometer))


def cycle(args):
if(args.debug):
print('\n\n[{}]: Starting a new poll-cycle for CoSI...'
.format(time.ctime()))

if(not args.no_tle):
fetch_tle(args)
if(not args.no_satellite):
fetch_satellite(args)
if(not args.no_telemetry):
fetch_telemetry(args)

if(args.latest_tle is not None):
try:
norad_id = int(args.latest_tle[0])
tle = m.get_latest_tle_by_id(norad_id)
except ValueError:
satellite_name = args.latest_tle[0]
tle = m.get_latest_tle_by_name(satellite_name)
print('Latest TLE: {}'.format(tle))
from . import APP_NAME, APP_DESCRIPTION
from .satnogs import request_satellite, \
request_telemetry, \
decode_telemetry_frame
from .spacetrack import request_tle


def main():
# Parse cmd line arguments
parser = argparse.ArgumentParser(prog='cosi-runner',
description='Daemon for CoSI.',
parser = argparse.ArgumentParser(prog=APP_NAME,
description=APP_DESCRIPTION,
allow_abbrev=False)
parser.add_argument('--latest-tle',
dest='latest_tle',
nargs=1,
metavar=('(NORAD ID|SATELLITE NAME)'),
help='[Default: 43793 (CSim)] Displays the latest TLE'
' stored in DART DB either by Norad ID or by'
' satellite name')
parser.add_argument('-n', '--norad-id',
dest='norad_id',
parser.add_argument('norad_id',
type=int,
default=43793,
help='[Default: 43793 (CSim)] Norad Satellite ID to'
' use when fetching TLE and Telemetry')
parser.add_argument('--no-satellite',
dest='no_satellite',
help='The satellite NORAD ID to search by when'
' fetching TLE\'s or telemetry.')
parser.add_argument('--get-satellite',
dest='get_satellite',
action='store_true',
default=False,
help='Disables fetching the latest satellite info from'
' https://db.satnogs.org')
parser.add_argument('--no-telemetry',
dest='no_telemetry',
help='Fetch the latest satellite metadata from SatNOGS'
' DB and save it as a JSON file.')
parser.add_argument('--get-telemetry',
dest='get_telemetry',
action='store_true',
default=False,
help='Disables fetching the latest telemetry from'
' https://db.satnogs.org')
parser.add_argument('--no-tle',
dest='no_tle',
help='Fetch the latest telemetry from SatNOGS DB and'
' save it as a binary file.')
parser.add_argument('--get-tle',
dest='get_tle',
action='store_true',
default=False,
help='Disables fetching the latest Two Line Element'
' (TLE) from https://space-track.org')
parser.add_argument('-o', '--once',
dest='once',
action='store_true',
default=False,
help='Runs a cycle once instead of looping every poll'
' interval')
parser.add_argument('-p', '--poll-interval',
dest='poll_interval',
help='Fetch the latest TLE from Space-Track and save it to file.')
parser.add_argument('--no-decode',
dest='decode',
action='store_false',
default=True,
help='Disable decoding the raw frames if fetching'
' satellite telemetry.')
parser.add_argument('--satellite-path',
dest='satellite_path',
type=str,
default='30m',
help='[Default: 30m] Time interval at which CoSI polls'
' spacetrack and satnogs for data')
parser.add_argument('-v', '--verbose',
dest='debug',
action='store_true',
default=False,
help='Enable additional debug information')
default='./satellite.json',
help='Specify the location to write the satellite'
' metadata to. Used in conjunction with'
' `--get-satellite`.')
parser.add_argument('--telemetry-path',
dest='telemetry_path',
type=str,
default='./telemetry.raw',
help='Specify the location to write the raw telemetry'
' to. Used in conjunction with `--get-telemetry`.')
parser.add_argument('--tle-path',
dest='tle_path',
type=str,
default='./latest.tle',
help='Specify the location to write the latest TLE to.'
' Used in conjunction with `--get-tle`.')
args = parser.parse_args()
poll_interval = parse_poll_interval(args.poll_interval)

if(args.once):
cycle(args)
else:
while True:
cycle(args)
time.sleep(poll_interval)
if(args.get_satellite):
satellite = request_satellite(args.norad_id)
with open(args.satellite_path, 'w+') as file:
json.dump(satellite, file, indent=4, sort_keys=True)

if(args.get_telemetry):
telemetry = request_telemetry(args.norad_id)
frame = telemetry.get('frame', None)
if(frame is None):
raise ValueError(f'Did not find a raw Frame for Satellite #{args.norad_id}: {telemetry}')
with open(args.telemetry_path, 'wb+') as file:
file.write(bytearray.fromhex(frame))

if(args.decode):
decoded_frame = decode_telemetry_frame(frame)
print(f'Decoded frame: {decoded_frame}')
# TODO: Write to DART database

if(args.get_tle):
tle_data = request_tle(args.norad_id)
tle = '\n'.join(tle_data.values())
with open(args.tle_path, 'w+') as file:
file.write(tle)


if __name__ == '__main__':
main()
Loading

0 comments on commit d0ef508

Please sign in to comment.