Small aiohttp server application for TLE storage
Project description
TLEStorageService
Small aiohttp server application for TLE storage
Requirements
aio-nasa-tle-loader >= 1.0.0
aio-space-track-api >= 2.1.1
aiohttp >= 2.1.0
aiopg >= 0.13.0
alembic >= 0.9.2
pyaml >=16.12.2
sqlalchemy >= 1.1.10
ujson >= 1.35
Installing
pip install tle-storage-service
Getting started
Start aiohttp application:
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import logging
import os
from aiohttp import web
from tle_storage_service import TleStorageService, load_cfg, LOG_FORMAT
DIR = os.path.abspath(os.path.dirname(__file__))
if __name__ == '__main__':
logging.basicConfig(level=logging.INFO, format=LOG_FORMAT)
app = TleStorageService(config=load_cfg(path=os.path.join(DIR, 'config', 'server.yaml')))
web.run_app(app, port=8080)
Create periodic script for refresh db
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
import asyncio
import logging
import os
from aio_nasa_tle_loader import AsyncNasaTLELoader
from aio_space_track_api import AsyncSpaceTrackApi
from aiopg.sa import create_engine
from tle_storage_service import periodic_task, load_cfg, nasa, space_track, LOG_FORMAT
DIR = os.path.abspath(os.path.dirname(__file__))
CFG = load_cfg(path=os.path.join(DIR, 'config', 'periodic.yaml'))
logger = logging.getLogger(__name__)
@periodic_task(CFG['sync'].get('space-track', 4) * 3600)
async def call_space_track(engine, api_params, satellites, loop):
async with AsyncSpaceTrackApi(loop=loop, **api_params) as api:
await space_track(engine, api, satellites)
@periodic_task(CFG['sync'].get('nasa', 0.5) * 3600)
async def call_nasa(engine, loop):
async with AsyncNasaTLELoader(loop=loop) as loader:
await nasa(engine, loader)
async def run(db_params, api_params, satellites, loop):
async with create_engine(loop=loop, **db_params) as engine:
await asyncio.gather(
call_nasa(engine, loop),
call_space_track(engine, api_params, satellites, loop),
loop=loop
)
def notify_callback(msg):
logger.info('[%s] Receive msg <- "%s"', msg.channel, msg.payload)
if __name__ == '__main__':
logging.basicConfig(format=LOG_FORMAT, level=logging.DEBUG)
loop = asyncio.get_event_loop()
loop.create_task(run(CFG['db'],
CFG['space-track'],
CFG['sync']['satellites'],
loop))
try:
loop.run_forever()
finally:
loop.close()
More code examples into examples directory
Source code
The latest developer version is available in a github repository: https://github.com/nkoshell/tle-storage-service
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
tle-storage-service-1.0.4.tar.gz
(14.8 kB
view hashes)
Built Distribution
Close
Hashes for tle-storage-service-1.0.4.tar.gz
Algorithm | Hash digest | |
---|---|---|
SHA256 | ba4a90918cab5ccbde052751301f466917248f38170b9bd1b440dbe5c70f945d |
|
MD5 | 2bbf246c29b4dcf5c123154d952df942 |
|
BLAKE2b-256 | 6f4266a0967d52a64e2d0f476c3b71bcdbfbe796503dbfc0be98751069587e7a |
Close
Hashes for tle_storage_service-1.0.4-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 877f5414bf44e39f12ff10d7583d446a311bd808890e89fa3b416ad906a907d1 |
|
MD5 | 8f031c19e29ce237f17ea6c27a65a755 |
|
BLAKE2b-256 | adc756fd96f621a44ad16e16c10d8c1dea9c9ddb7a29a4bc6152059aec110cf3 |