Source code for irfpy.cy1orb.Cy1OrbitNr

''' Orbit number handling.

Setup:

1. Prepare the orbit number file. It is named ``ch1_orbit_number.out_v2``.
   The original file is existing at http://sara.irf.se/archive/raw/orbnum/ch1_orbit_number.out
2. Setup the ${HOME}/.irfpyrc file.  In the section of ``cy1orb``, the prepared file
   should be specified in the entry , ``orbnruri``.

   ::

       [cy1orb]
       orbnruri = file:///path/to/the/file/ch1_orbit_number.out_v2

Usage:

3. Use either of these functions

    - :func:`get_first_orbit_number`
    - :func:`get_first_time`
    - :func:`get_last_orbit_number`
    - :func:`get_last_time`
    - :func:`get_start_time`
    - :func:`get_stop_time`
    - :func:`get_orbit_number`


Sample:

>>> from irfpy.cy1orb import Cy1OrbitNr as cy1onr
>>> t0 = cy1onr.get_first_time()   # The time that the orbit file starts
>>> print(t0)
2008-10-22 01:10:19.081000

>>> t1 = cy1onr.get_last_time()   # The time that the orbit file ends
>>> print(t1)
2009-08-29 10:13:45.081000


>>> orb0 = cy1onr.get_first_orbit_number()   # The start orbit of the orbit file
>>> print(orb0)
0

>>> orb1 = cy1onr.get_last_orbit_number()   # The start orbit of the orbit file
>>> print(orb1)
3475

>>> print(cy1onr.get_start_time(1000))   # Start time of the orbit 1000
2009-01-30 13:29:33.081000

>>> print(cy1onr.get_stop_time(1000))   # Stop time of the orbit 1000
2009-01-30 15:27:39.081000

>>> import datetime
>>> tt = datetime.datetime(2009, 1, 30, 14, 0, 0)
>>> print(cy1onr.get_orbit_number(tt))
1000



.. note::

    In early 2011, we found that the orbit number file had been changed.
    This may introduce a big confusion.
    It has been handled transparently, and in case one uses old (v1) file
    the exception of :class:`OldOrbnumFileError` is thrown

    See :ref:`ref-orbitnr` for details.

'''
import urllib.request
import urllib.parse
import urllib.error
import gzip

import logging
import calendar
import datetime

import bisect

from irfpy.util.irfpyrc import Rc
from irfpy.util import utc

_logger = logging.getLogger(__name__)


[docs]class Cy1OrbitNr: ''' Orbit number class. Backend. Class that treats the orbit number. Even though orbit number is not spice-base, the class is considered as a part of spice related class. The input file is normally provided through the ISSDC public server. The file is saved at ssh://butler.irf.se/home/sara/saradds/archive/incoming/issdc/orbit_no_and_eph/ch1_orbit_number.out.gz. The file should be 'well behaved', i.e., the format is strict (14 columns: 13 decimal int + 1 floating point), in addition to the orbit number is sequencial and continuous. The interface is the same as C++ implementation in https://butler.irf.se/sara-trac/file/ISSDC/spice/module/SaraSpice/Cy1OrbitNr.h ''' DEFAULT_URI = None # It is already deprecated. def __init__(self, rcfile=None): ''' Constructor of Cy1OrbitNr class. Constructor do nothing. You have to call setFromDefaultUri() or setFromFile() before using. ''' self._data = {} """Internal expression of the loaded data. orbit_number => (start_dt, stop_dt)""" self._cache = {'starttime': [], # Cache. 'starttime' is for starttime. 'time2orb': {}, # 'time2orb' is for conversion from time to orbit. 'firstorbit': None, 'lastorbit': None, } rc = Rc() if rcfile is not None: rc.append(rcfile) self.orbnruri = rc.get('cy1orb', 'orbnruri') def _isdataset(self): return len(self._data) != 0 def _parse(self, buffer): bspl = buffer.split('\n') # print(len(bspl)) for line in bspl: es = line.split() if len(es) != 16: continue e = [int(es[i]) for i in range(15)] start = datetime.datetime(e[1], e[2], e[3], e[4], e[5], e[6], e[7] * 1000) stop = datetime.datetime(e[8], e[9], e[10], e[11], e[12], e[13], e[14] * 1000) self._data[e[0]] = (start, stop) # For caching. orbs = list(self._data.keys()) self._cache['starttime'] = [self._data[orb][0] for orb in orbs] self._cache['starttime'].sort() for orb in orbs: self._cache['time2orb'][self._data[orb][0]] = orb self._cache['firstorbit'] = min(orbs) self._cache['lastorbit'] = max(orbs) def _file_version(self): ''' File version is now 1 and 2. Use 2. ''' if not self._isdataset(): raise RuntimeError('Orbit number data not set; Use setFromXXX() method to load the data.') t = self.getStartTime(10) t = utc.convert(t, datetime.datetime) t_v1 = datetime.datetime(2008, 10, 26, 0o1, 20, 43) t_v2 = datetime.datetime(2008, 10, 29, 0o2, 13, 55) if t == t_v1: return 1 elif t == t_v2: return 2 else: _logger.error('Unknown version of ch1_orbit_number file.') raise RuntimeError("Unknown orbit file; Contact to the developer with the loaded file.")
[docs] def setFromFile(self, filename): ''' Set the orbit information data from the specified file. ''' if filename.endswith('.gz'): f = gzip.GzipFile(filename, 'r') else: f = open(filename, 'r') buf = f.read() f.close() self._parse(buf) v = self._file_version() if v == 1: raise OldOrbnumFileError('Version 1 orbnum file loaded. Use version 2 file.')
[docs] def setFromUri(self, uriname): ''' Set the orbit information data from the specified URI. ''' _logger.debug('URI specified as %s' % uriname) try: ff = urllib.request.urlretrieve(uriname) except IOError as e: _logger.error('Failed to load the orbit number file: %s' % uriname) raise e self.setFromFile(ff[0]) v = self._file_version() if v == 1: raise OldOrbnumFileError('Version 1 orbnum file loaded. Use version 2 file.')
[docs] def setFromDefaultUri(self): ''' Set the orbit information data from the URI specified in the ``.irfpyrc``. ''' self.setFromUri(self.orbnruri) v = self._file_version() if v == 1: raise OldOrbnumFileError('Version 1 orbnum file loaded. Use version 2 file.')
[docs] def getFirstOrbitNr(self): ''' Returns the first orbit number ''' if not self._isdataset(): raise RuntimeError("Orbit number data not set. Use setFromXXX() method.\n") return self._cache['firstorbit']
[docs] def getFirstStarttime(self): onr = self.getFirstOrbitNr() return self.getStartDatetime(onr)
[docs] def getLastStoptime(self): onr = self.getLastOrbitNr() return self.getStopDatetime(onr)
[docs] def getLastOrbitNr(self): ''' Returns the last orbit number ''' if not self._isdataset(): raise RuntimeError("Orbit number data not set. Use setFromXXX() method.\n") return self._cache['lastorbit']
[docs] def getStartTime(self, orbitNr): ''' Returns the start time of the specified orbit number ''' if not self._isdataset(): raise RuntimeError("Orbit number data not set. Use setFromXXX() method.\n") if orbitNr not in self._data: raise RuntimeError("Orbit %d not found." % orbitNr) t = self._data[orbitNr][0] return calendar.timegm(t.utctimetuple()) + t.microsecond / 1e6
[docs] def getStartDatetime(self, orbitNr): ''' Return the start time of the specified orbit number as ``datetime.datetime``. ''' t = self.getStartTime(orbitNr) return utc.convert(t, datetime.datetime)
[docs] def getStopTime(self, orbitNr): ''' Returns the stop time of the specified orbit number ''' if not self._isdataset(): raise RuntimeError("Orbit number data not set. Use setFromXXX() method.\n") if orbitNr not in self._data: raise RuntimeError("Orbit %d not found." % orbitNr) t = self._data[orbitNr][1] return calendar.timegm(t.utctimetuple()) + t.microsecond / 1e6
[docs] def getStopDatetime(self, orbitNr): ''' Return the stop time of the specified orbit number as ``datetime.datetime``. ''' t = self.getStopTime(orbitNr) return utc.convert(t, datetime.datetime)
[docs] def getOrbitNr(self, t): '''Get the orbit number of the specified time. :param t: Time of inquiry. :type t: ``datetime.datetime`` ''' # print len(self._data) if not self._isdataset(): raise RuntimeError("Orbit number data not set. Use setFromXXX() method.\n") dt = utc.convert(t, datetime.datetime) _logger.debug('Time specified = %s' % str(dt)) t0 = self.getFirstStarttime() t1 = self.getLastStoptime() if dt < t0: raise RuntimeError('Orbit number not defined. Before database') elif dt >= t1: raise RuntimeError('Orbit number not defined. After database') onr = bisect.bisect(self._cache['starttime'], dt) - 1 return onr
try: _onr = Cy1OrbitNr() """ Default Cy1OrbitNr object """ _onr.setFromDefaultUri() except BaseException as _e: _logger.warning("Not successful loading of default orbit number object") _logger.warning("Warning message is as follows") _logger.warning(str(_e)) _logger.warning("Some functionality regarding orbit number would not work") _onr = None
[docs]def get_first_orbit_number(): ''' Returns the first orbit number, usually, 0 ''' return _onr.getFirstOrbitNr()
[docs]def get_first_time(): """ Return the start time supported by the orbit number file :return: Time of the start of the orbit number file :rtype: ``datetime.datetime`` """ return _onr.getFirstStarttime()
[docs]def get_last_orbit_number(): ''' Returns the first orbit number, usually, 3475 ''' return _onr.getLastOrbitNr()
[docs]def get_last_time(): """ Return the time of the orbit number file supported. :returns: Time of the end of the orbit number file :rtype: ``datetime.datetime`` """ return _onr.getLastStoptime()
[docs]def get_start_time(orbit_number): """ Return the start time of the given orbit number. :param orbit_number: Orbit number. :returns: The start time of the given orbit number. """ return _onr.getStartDatetime(orbit_number)
[docs]def get_stop_time(orbit_number): """ Return the stop time of the given orbit number. :param orbit_number: Orbit number. :returns: The stop time of the given orbit number. """ return _onr.getStopDatetime(orbit_number)
[docs]def get_orbit_number(t): """ Return the orbit number of the given time :param t: Time :type t: ``datetime.datetime`` :return: Orbit number """ return _onr.getOrbitNr(t)
[docs]class OldOrbnumFileError(RuntimeError): ''' Exception raised if v1 version of the orbnum file is loaded. >>> try: ... raise OldOrbnumFileError("Old orbnum file loaded.") ... print("If you reach this line, something wrong.") ... except OldOrbnumFileError as e: ... print(e) 'Old orbnum file loaded.' ''' def __init__(self, value): self.value = value def __str__(self): return repr(self.value)