init code

celestrak tles.create_tables.sql

@@ -0,0 +1,30 @@
+create table
+tles (
+    sat varchar(70) not null primary key,
+    tle varchar(140) not null
+);
+
+create table
+sat_groups (
+    id mediumint not null auto_increment primary key,
+    sat varchar(70) not null,
+    group_name varchar(20),
+    foreign key (group_name) references groups(group_name)
+);
+
+create table
+groups (
+    group_name varchar(20) primary key
+)
+
+insert into groups (group_name) values ('weather'), ('noaa'), ('goes'), ('resource'), ('sarsat'), ('dmc'), ('tdrss'), ('argos'), ('planet'), ('spire'), ('geo'), ('intelsat'), ('ses'), ('iridium'), ('iridium-NEXT'), ('starlink'), ('oneweb'), ('orbcomm'), ('globalstar'), ('swarm'), ('amateur'), ('x-comm'), ('other-comm'), ('satnogs'), ('gorizont'), ('raduga'), ('molniya'), ('gnss'), ('gps-ops'), ('glo-ops'), ('galileo'), ('beidou'), ('sbas'), ('nnss'), ('musson'), ('science'), ('geodetic'), ('education'), ('engineering'), ('military'), ('radar'), ('cubesat'), ('other');
+
+select * from groups;
+select * from sat_groups;
+select * from tles;
+
+drop table sat_groups;
+delete from groups;
+
+select group_name from sat_groups
+where sat like 'NOAA 15%';

tle_database.py

@@ -0,0 +1,58 @@
+import mariadb
+import logging
+import sys
+
+logger = logging.getLogger('tle_fetcher.' + __name__)
+
+class TleDatabase: 
+
+    def __init__(self):
+        self.conn = None
+
+    def connect_db(self):
+
+        # Connect to MariaDB
+        try:
+            conn = mariadb.connect(
+                user='tle_getter',
+                host='localhost',
+                port=3306,
+                database='celestrak_tles'
+            )
+        except mariadb.Error as e:
+            logger.error("Error connecting to MariaDB Platform: %s", e.msg)
+            sys.exit(1)
+
+        self.conn = conn
+
+    def delete_rows(self):
+        cur = self.conn.cursor()
+
+        try:
+            cur.execute('delete from sat_groups')
+            cur.execute('delete from tles')     
+        except mariadb.Error as e: 
+            logger.error("Error clearing tables: %s", e.msg)
+        self.conn.commit()
+
+    def update_tles(self, tles: dict):
+        cur = self.conn.cursor()
+
+        try: 
+            cur.executemany('INSERT INTO tles (sat, tle) values (?,?)', list(tles.items()))
+        except mariadb.Error as e: 
+            logger.error("Error updating TLEs: %s", e.msg)
+        self.conn.commit() 
+
+    def update_sat_group(self, group: list):
+        cur = self.conn.cursor()
+
+        try:    
+            cur.executemany('INSERT INTO sat_groups (sat, group_name) values (?,?)', group)
+        except mariadb.Error as e: 
+            logger.error("Error updating groups: %s", e.msg)
+        self.conn.commit()
+
+    def close_db(self):
+        self.conn.close()
+        

tle_fetcher.py

@@ -0,0 +1,92 @@
+#!/usr/bin/python3
+
+from tle_database import TleDatabase
+from tle_reader import TleReader
+import logging
+
+CELESTRAK_ROOT = 'https://www.celestrak.com/NORAD/elements/'
+
+groups = [
+    'weather',
+    'noaa',
+    'goes',
+    'resource',
+    'sarsat',
+    'dmc',
+    'tdrss',
+    'argos',
+    'planet',
+    'spire',
+    'geo',
+    'intelsat',
+    'ses',
+    'iridium',
+    'iridium-NEXT',
+    'starlink',
+    'oneweb',
+    'orbcomm',
+    'globalstar',
+    'swarm',
+    'amateur',
+    'x-comm',
+    'other-comm',
+    'satnogs',
+    'gorizont',
+    'raduga',
+    'molniya',
+    'gnss',
+    'gps-ops',
+    'glo-ops',
+    'galileo',
+    'beidou',
+    'sbas',
+    'nnss',
+    'musson',
+    'science',
+    'geodetic',
+    'engineering',
+    'education',
+    'military',
+    'radar',
+    'cubesat',
+    'other'
+    ]
+
+log = logging.getLogger('tle_fetcher')
+log.setLevel(logging.INFO)
+
+time_string = "%m/%d/%Y %H:%M:%S %Z"
+formatter = logging.Formatter(
+    fmt='{asctime} [{module}] {message}',
+    datefmt=time_string,
+    style='{')
+
+f_handler = logging.FileHandler('/home/bryon/dev/tle_fetcher/tle_fetcher.log', mode='a')
+f_handler.setLevel(logging.INFO)
+f_handler.setFormatter(formatter)
+log.addHandler(f_handler)
+
+c_handler = logging.StreamHandler()
+c_handler.setLevel(logging.INFO)
+c_handler.setFormatter(formatter)
+log.addHandler(c_handler)
+
+log.info("TLE fetcher started")
+
+
+def main(tle_rd: TleReader, tle_db: TleDatabase) -> None:
+
+    tles = tle_rd.get_tles()
+
+    tle_db.connect_db()
+    tle_db.delete_rows()
+    tle_db.update_tles(tles)
+    for group in groups:
+        gp = tle_rd.get_group(group)
+        tle_db.update_sat_group(gp)
+    tle_db.close_db()
+
+    log.info("TLE fetcher completed")
+
+if __name__ == "__main__":
+    main(TleReader(CELESTRAK_ROOT), TleDatabase())

tle_reader.py

@@ -0,0 +1,52 @@
+import urllib.request
+import sys
+import logging
+
+logger = logging.getLogger('tle_fetcher.' + __name__)
+
+class TleReader():
+    def __init__(self, url_root: str):
+        self.url_root = url_root   
+
+    def get_tles(self) -> dict:
+        url = self.url_root + 'active.txt'
+        tles = {}
+
+        try:
+            with urllib.request.urlopen(url) as txt:
+                i = 0  
+                key = ""
+                value = ""
+                for line in txt:
+                    line = line.strip().decode('utf-8')
+                    if i % 3 == 0:
+                        key = line
+                    elif i % 3 == 1:
+                        value = line
+                    else:
+                        value += "\n" + line
+                        tles[key] = value
+                    i += 1
+        except Exception as ex:
+            logger.error("Error getting TLEs: %s", ex.msg)
+            sys.exit(1)
+        
+        return tles
+
+    def get_group(self, group_name: str) -> list:
+        url = self.url_root + group_name + '.txt'
+        group = []
+        
+        try:    
+            with urllib.request.urlopen(url) as txt:
+                i = 0  
+                for line in txt:
+                    line = line.strip().decode('utf-8')
+                    if i % 3 == 0:
+                        group.append((line, group_name))
+                    i += 1
+        except Exception as ex:
+            logger.error("Error getting groups: %s", ex.msg)
+            sys.exit(1)
+        
+        return group