Add type hints

lsl/catalog.py

@@ -19,6 +19,7 @@
 from lsl.misc import telemetry
 telemetry.track_module()
 
+from typing import Dict, List, Union
 
 __version__   = '0.2'
 __all__ = ['CatalogEntry', 'Catalog', 'LWA_Catalog', 'PSR_Catalog', 'PKS_Catalog', 
@@ -43,14 +44,14 @@
     # limit the class attributes
     __slots__ = ('name', 'position', 'alias_list')
 
-    def __init__(self, name, position):
+    def __init__(self, name: str, position: transform.CelestialPosition):
         """
         Create a catalog entry.
         """
 
         self.name = name
         self.position = position
-        self.alias_list = []
+        self.alias_list: List[str] = []
 
     def __repr__(self):
         """
@@ -74,15 +75,15 @@
 
     __metaclass__ = abc.ABCMeta
 
-    def __init__(self, name):
+    def __init__(self, name: str):
         """
         Create a source catalog.
         """
 
         # initialize catalog data structures
         self.name = name
-        self.source_map = {}
-        self.alias_map = {}
+        self.source_map: Dict[str,CatalogEntry] = {}
+        self.alias_map: Dict[str,CatalogEntry] = {}
 
         # parse_file() is an abstract method which must be defined in
         # a concrete implementation for a particular catalog  
@@ -95,10 +96,10 @@
         structures.
         """
 
-        pass
+        raise NotImplementedError
 
     @staticmethod
-    def get_directory():
+    def get_directory() -> str:
         """
         Returns the path to the catalog data file directory.
         """
@@ -119,7 +120,7 @@
 
         return len(self.source_map)
 
-    def __getitem__(self, key):
+    def __getitem__(self, key: str) -> CatalogEntry:
         """
         Access source by subscript name.  Raises KeyError if the source
         is not in the catalog.
@@ -137,7 +138,7 @@
 
         return iter(self.source_map.keys())
 
-    def lookup(self, name):
+    def lookup(self, name: str) -> Union[CatalogEntry,None]:
         """
         Lookup a source in the catalog.
 
@@ -592,7 +593,7 @@
     Base definition for the Fermi LAT point source catalogs.
     """
 
-    def __init__(self, name, filename):
+    def __init__(self, name: str, filename: str):
         """
         Create a Fermi LAT catalog instance.
         """
@@ -694,10 +695,10 @@
                     }
 
     # a mapping of catalog names to instances
-    catalog_instance_map = {}
+    catalog_instance_map: Dict[str,Catalog] = {}
 
     @classmethod
-    def get_catalog(klass, name):
+    def get_catalog(klass, name: str) -> Catalog:
         """
         Returns a Catalog object representing the catalog
         given by name.
@@ -714,13 +715,13 @@
             catalog = klass.catalog_instance_map[name]
         except KeyError:
             catalogClass = klass.catalog_class_map[name]
-            catalog = catalogClass()
+            catalog = catalogClass()    # type: ignore
             klass.catalog_instance_map[name] = catalog
 
         return catalog
 
     @classmethod
-    def get_names(klass):
+    def get_names(klass) -> List[str]:
         """
         Return a list of known catalog names.
         """

lsl/common/_fir.pyi

@@ -0,0 +1,7 @@
+import numpy as np
+
+from typing import Tuple
+
+def integer16(signals: np.ndarray, filter: np.ndarray) -> np.ndarray: ...
+def integer16Delayed(signals: np.ndarray, filter: np.ndarray, delay: int) -> np.ndarray: ...
+def integerBeamformer(signals: np.ndarray, filters: np.ndarray, courses: np.ndarray, fines: np.ndarray, gains: np.ndarray) -> Tuple[np.ndarray,np.ndarray]: ...

lsl/common/adp.py

@@ -19,9 +19,13 @@
 from scipy.signal import freqz
 from scipy.interpolate import interp1d
 
+from lsl.common.data_access import DataAccess
+
 from lsl.misc import telemetry
 telemetry.track_module()
 
+from typing import Callable
+
 
 __version__ = '0.4'
 __all__ = ['fS', 'fC', 'T', 'T2', 'N_MAX', 'TBN_TUNING_WORD_MIN', 'TBN_TUNING_WORD_MAX',
@@ -54,42 +58,41 @@
 #: Maximum number of beams
 DRX_BEAMS_MAX = 3
 
-# FIR Filters
-## TBN
-_TBN_FIR = [-3.22350e-05, -0.00021433,  0.0017756, -0.0044913,  0.0040327, 
-             0.00735870,  -0.03218100,  0.0553980, -0.0398360, -0.0748920, 
-             0.58308000,   0.58308000, -0.0748920, -0.0398360,  0.0553980,
-            -0.03218100,   0.00735870,  0.0040327, -0.0044913,  0.0017756,
-            -0.00021433,  -3.2235e-05]
-## DRX
-_DRX_FIR = [ 0.0111580, -0.0074330,  0.0085684, -0.0085984,  0.0070656, -0.0035905, 
-            -0.0020837,  0.0099858, -0.0199800,  0.0316360, -0.0443470,  0.0573270, 
-            -0.0696630,  0.0804420, -0.0888320,  0.0941650,  0.9040000,  0.0941650, 
-            -0.0888320,  0.0804420, -0.0696630,  0.0573270, -0.0443470,  0.0316360, 
-            -0.0199800,  0.0099858, -0.0020837, -0.0035905,  0.0070656, -0.0085984,  
-             0.0085684, -0.0074330,  0.0111580]
-
-
+with DataAccess.open('digital/adp_coeffs.npz', 'rb') as fh:
+    dataDict = np.load(fh)
+
+    # FIR Filters
+    ## TBN
+    _TBN_FIR = dataDict['TBN_FIR'][...]
+
+    ## DRX
+    _DRX_FIR = dataDict['DRX_FIR'][...]
+
+    try:
+        dataDict.close()
+    except AttributeError:
+        pass
+
 _N_PTS = 1000 # Number of points to use in calculating the bandpasses
 
 
-def freq_to_word(freq):
+def freq_to_word(freq: float) -> int:
     """
     Given a frequency in Hz, convert it to the closest DP tuning word.
     """
 
     return int(round(freq*2**32 / fS))
 
 
-def word_to_freq(word):
+def word_to_freq(word: int) -> float:
     """
     Given a DP tuning word, convert it to a frequncy in Hz.
     """
 
     return word*fS / 2**32
 
 
-def delay_to_dpd(delay):
+def delay_to_dpd(delay: float) -> int:
     """
     Given a delay in ns, convert it to a course and fine portion and into the 
     final format expected by ADP (big endian 16.12 unsigned integer)
@@ -110,7 +113,7 @@
     return combined
 
 
-def dpd_to_delay(combined):
+def dpd_to_delay(combined: int) -> float:
     """
     Given a delay value in the final format expect by ADP, return the delay in ns.
     """
@@ -128,7 +131,7 @@
     return delay
 
 
-def gain_to_dpg(gain):
+def gain_to_dpg(gain: float) -> int:
     """
     Given a gain (between 0 and 1), convert it to a gain in the final form 
     expected by ADP (big endian 16.1 signed integer).
@@ -143,7 +146,7 @@
     return combined
 
 
-def dpg_to_gain(combined):
+def dpg_to_gain(combined: int) -> float:
     """
     Given a gain value in the final format expected by ADP, return the gain
     as a decimal value (0 to 1).
@@ -158,7 +161,7 @@
     return gain
 
 
-def tbn_filter(sample_rate=1e5, npts=_N_PTS):
+def tbn_filter(sample_rate: float=1e5, npts: int=_N_PTS) -> Callable:
     """
     Return a function that will generate the shape of a TBN filter for a given sample
     rate.
@@ -180,7 +183,7 @@
     return interp1d(h, w/w.max(), kind='cubic', bounds_error=False, fill_value=0.0)
 
 
-def drx_filter(sample_rate=19.6e6, npts=_N_PTS):
+def drx_filter(sample_rate: float=19.6e6, npts: int=_N_PTS) -> Callable:
     """
     Return a function that will generate the shape of a DRX filter for a given sample
     rate.