TLE collection

tle_collection.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "import requests\n",
    "from pyorbital.tlefile import Tle\n",
    "from pyorbital.orbital import Orbital\n",
    "from datetime import datetime\n",
    "from functools import partial\n",
    "import zipfile\n",
    "import numpy as np\n",
    "\n",
    "import collections\n",
    "from collections import Iterable"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "class Tle_collection(object):\n",
    "    \n",
    "    def __init__(self, tle_objects=[]):\n",
    "        \n",
    "        if not isinstance(tle_objects, Iterable):\n",
    "            raise TypeError('tle_objects must be an iterable collection of some sort')\n",
    "            \n",
    "        if not all(map(lambda x: isinstance(x, Tle), tle_objects)):\n",
    "            raise TypeError('tle_objects can only contain Tle objects')            \n",
    "            \n",
    "        self.collection = tle_objects\n",
    "        \n",
    "    def __len__(self):\n",
    "        return len(self.collection)\n",
    "        \n",
    "    @classmethod\n",
    "    def from_spacetrack(cls, response, platform='noname'):\n",
    "        \n",
    "        lines = response.splitlines()\n",
    "        \n",
    "        tle_objects = list(map(lambda x: Tle(platform, line1=x[0], line2=x[1]), zip(lines[:-1:2], lines[1::2])))\n",
    "        \n",
    "        return cls(tle_objects=tle_objects)\n",
    "    \n",
    "    @classmethod\n",
    "    def from_celestrak(cls, zip_file):\n",
    "        \n",
    "        with zipfile.ZipFile(zip_file) as zf:\n",
    "            \n",
    "            # only read txt files (?)\n",
    "            txt_files = filter(lambda x: x.endswith('.txt'), zf.namelist())\n",
    "            \n",
    "            contents = b''\n",
    "            for txt_file in txt_files:\n",
    "                contents += zf.read(txt_file)\n",
    "                \n",
    "        contents = contents.decode('utf-8')\n",
    "        \n",
    "        return cls().from_spacetrack(contents)\n",
    "        \n",
    "    def _sort_function(self, target_dt, tle):\n",
    "        \n",
    "        dt_seconds = abs((tle.epoch - target_dt) / np.timedelta64(1, 's'))\n",
    "        \n",
    "        return dt_seconds\n",
    "        \n",
    "    def get_closest_tle(self, target_dt):\n",
    "\n",
    "        sort_func = partial(self._sort_function, target_dt)        \n",
    "        sorted_tle = sorted(self.collection, key=sort_func)\n",
    "\n",
    "        return sorted_tle[0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Space-Track example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Download TLE's "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [],
   "source": [
    "payload = {'identity': '[email protected]',\n",
    "           'password': 'somepassword'}\n",
    "\n",
    "login_url = 'https://www.space-track.org/auth/login'\n",
    "\n",
    "url_template = 'https://www.space-track.org/basicspacedata/query/class/tle/EPOCH/{dt_from}--{dt_till}/NORAD_CAT_ID/{norad_id}/orderby/EPOCH%20ASC/format/tle'\n",
    "\n",
    "dt_from = '2018-01-01'\n",
    "dt_till = '2018-01-05'\n",
    "\n",
    "norad_id = '43013' # NOAA 20 / VIIRS\n",
    "\n",
    "with requests.Session() as s:\n",
    "    p = s.post(login_url, data=payload)\n",
    "    \n",
    "    if p.status_code != 200:\n",
    "        raise Exception('Login failed')\n",
    "\n",
    "    url = url_template.format(dt_from=dt_from, dt_till=dt_till, norad_id=norad_id)\n",
    "    r = s.get(url)\n",
    "\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [],
   "source": [
    "tle_coll = Tle_collection.from_spacetrack(r.text)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "[(numpy.datetime64('2018-01-01T12:32:43.916928'), '18', 1.52273052),\n",
       " (numpy.datetime64('2018-01-01T21:00:13.054176'), '18', 1.87515109),\n",
       " (numpy.datetime64('2018-01-02T12:13:41.501568'), '18', 2.50950812),\n",
       " (numpy.datetime64('2018-01-02T13:55:11.328672'), '18', 2.57999223),\n",
       " (numpy.datetime64('2018-01-02T20:41:10.635360'), '18', 2.86192865),\n",
       " (numpy.datetime64('2018-01-03T13:36:08.908128'), '18', 3.56676977),\n",
       " (numpy.datetime64('2018-01-03T18:40:38.386848'), '18', 3.77822207),\n",
       " (numpy.datetime64('2018-01-04T13:17:06.480672'), '18', 4.55354723),\n",
       " (numpy.datetime64('2018-01-04T20:03:05.785632'), '18', 4.83548363),\n",
       " (numpy.datetime64('2018-01-04T21:44:35.609280'), '18', 4.9059677)]"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "# show the last 10 TLE's \n",
    "[(x.epoch, x.epoch_year, x.epoch_day) for x in tle_coll.collection][-10:]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Find closest TLE to target_dt"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(numpy.datetime64('2018-01-03T13:36:08.908128'), pyorbital.tlefile.Tle)"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "target_dt = np.datetime64('2018-01-03 12:00:00')\n",
    "\n",
    "closest = tle_coll.get_closest_tle(target_dt)\n",
    "\n",
    "closest.epoch, type(closest)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Celestrak example"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Build collection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "5424"
      ]
     },
     "execution_count": 7,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "terra_archive = 'D:/Data/TLE/Celestrak/terra.zip'\n",
    "\n",
    "tle_coll = Tle_collection.from_celestrak(terra_archive)\n",
    "\n",
    "len(tle_coll)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "1 25994U 99068A   02155.22316851  .00000826  00000-0  19331-3 0  9134\n",
      "2 25994  98.2059 230.2895 0000801 106.7960 253.3331 14.57122284130898\n"
     ]
    }
   ],
   "source": [
    "target_dt = np.datetime64('2002-06-04 09:50:00')\n",
    "\n",
    "closest = tle_coll.get_closest_tle(target_dt)\n",
    "\n",
    "print(closest.line1, closest.line2, sep='\\n')"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.4"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}