Code cleanup (#5)

Code cleanup and codacy configuration

Signed-off-by: Travis F. Collins <[email protected]>

.codacy.yml

@@ -0,0 +1,7 @@
+exclude_paths:
+  - 'examples/**'
+  - 'test/**'
+  - 'doc/**'
+  - '**.md'
+  - '**.yml'
+  - '**.yaml'

adi/ad9361.py

@@ -165,4 +165,3 @@ class Pluto(ad9364):
 
     _device_name = "PlutoSDR"
     _uri_auto = "ip:pluto.local"
-    pass

adi/context_manager.py

@@ -53,7 +53,7 @@ def __init__(self, uri="", _device_name=""):
                 if not self._ctx and self._uri_auto != "":
                     self._ctx = iio.Context(self._uri_auto)
                 if not self._ctx:
-                    raise
+                    raise Exception("No device found")
             else:
                 self._ctx = iio.Context(self.uri)
         except BaseException:

adi/dds.py

@@ -40,11 +40,6 @@ class dds(attribute):
         this allows for two complex tones to be generated per complex channel.
     """
 
-    dds_frequencies = []
-    dds_scales = []
-    dds_phases = []
-    dds_enabled = []
-
     def __init__(self):
         self.dds_frequencies = np.zeros(self.num_tx_channels * 2)
         self.dds_scales = np.zeros(self.num_tx_channels * 2)

adi/rx_tx.py

@@ -87,29 +87,29 @@ def __del__(self):
         self.__rxbuf = []
         self._rxadc = []
 
-    def _init_channels(self):
+    def _rx_init_channels(self):
         if self._complex_data:
-            for map in self.rx_enabled_channels:
-                v = self._rxadc.find_channel(self._rx_channel_names[map * 2])
+            for m in self.rx_enabled_channels:
+                v = self._rxadc.find_channel(self._rx_channel_names[m * 2])
                 v.enabled = True
-                v = self._rxadc.find_channel(self._rx_channel_names[map * 2 + 1])
+                v = self._rxadc.find_channel(self._rx_channel_names[m * 2 + 1])
                 v.enabled = True
         else:
-            for map in self.rx_enabled_channels:
-                v = self._rxadc.find_channel(self._rx_channel_names[map])
+            for m in self.rx_enabled_channels:
+                v = self._rxadc.find_channel(self._rx_channel_names[m])
                 v.enabled = True
         self.__rxbuf = iio.Buffer(self._rxadc, self.__rx_buffer_size, False)
 
     def __rx_complex(self):
         if not self.__rxbuf:
-            self._init_channels(False)
+            self._rx_init_channels()
         self.__rxbuf.refill()
         data = self.__rxbuf.read()
         x = np.frombuffer(data, dtype=np.int16)
         indx = 0
         sig = []
         l = len(self.rx_enabled_channels) * 2
-        for c in range(l // 2):
+        for _ in range(l // 2):
             sig.append(x[indx::l] + 1j * x[indx + 1 :: l])
             indx = indx + 2
         # Don't return list if a single channel
@@ -119,11 +119,10 @@ def __rx_complex(self):
 
     def __rx_non_complex(self):
         if not self.__rxbuf:
-            self._init_channels(False)
+            self._rx_init_channels()
         self.__rxbuf.refill()
         data = self.__rxbuf.read()
         x = np.frombuffer(data, dtype=np.int16)
-        indx = 0
         sig = []
         l = len(self.__rx_enabled_channels)
         for c in range(l):
@@ -184,16 +183,16 @@ def tx_enabled_channels(self, value):
                 raise Exception("TX mapping exceeds available channels")
         self.__tx_enabled_channels = value
 
-    def _init_channels(self):
+    def _tx_init_channels(self):
         if self._complex_data:
-            for map in self.tx_enabled_channels:
-                v = self._txdac.find_channel(self._tx_channel_names[map * 2], True)
+            for m in self.tx_enabled_channels:
+                v = self._txdac.find_channel(self._tx_channel_names[m * 2], True)
                 v.enabled = True
-                v = self._txdac.find_channel(self._tx_channel_names[map * 2 + 1], True)
+                v = self._txdac.find_channel(self._tx_channel_names[m * 2 + 1], True)
                 v.enabled = True
         else:
-            for map in self.tx_enabled_channels:
-                v = self._txdac.find_channel(self._tx_channel_names[map])
+            for m in self.tx_enabled_channels:
+                v = self._txdac.find_channel(self._tx_channel_names[m])
                 v.enabled = True
         self.__txbuf = iio.Buffer(
             self._txdac, self.tx_buffer_size, self.__tx_cyclic_buffer
@@ -221,18 +220,20 @@ def tx(self, data_np):
                 data_np = [data_np]
             indx = 0
             l = self.num_tx_channels_enabled
-            iq = np.empty(l * len(data_np), dtype=np.int16)
+            data = np.empty(l * len(data_np), dtype=np.int16)
             for chan in data_np:
                 data[indx::l] = chan.astype(int)
                 indx = indx + 1
 
         if not self.__txbuf:
             self.disable_dds()
             self.tx_buff_length = len(data)
-            self._init_channels(True)
+            self._tx_init_channels()
 
         if len(data) != self.tx_buff_length:
-            raise
+            raise Exception(
+                "Buffer length different than data length. Cannot change buffer length on the fly"
+            )
 
         # Send data to buffer
         self.__txbuf.write(bytearray(data))
@@ -253,9 +254,3 @@ def __del__(self):
         rx.__del__(self)
         tx.__del__(self)
         phy.__del__(self)
-
-    def _init_channels(self, istx=False):
-        if istx:
-            tx._init_channels(self)
-        else:
-            rx._init_channels(self)

examples/waterfall.py

@@ -5,7 +5,6 @@
 from matplotlib import mlab as mlab
 import numpy as np
 from PIL import Image
-import time
 import pygame
 
 DISPLAY_WIDTH = 256
@@ -94,4 +93,4 @@ def mymap(x, in_min, in_max, out_min, out_max):
 except KeyboardInterrupt:
     pass
 finally:
-    sdr.close()
+    sdr = []

test/test_pluto.py

@@ -33,8 +33,6 @@
 
 from __future__ import print_function
 
-import logging
-
 import unittest
 import numpy as np
 from adi import Pluto
@@ -58,20 +56,20 @@ def check_pluto():
 
 class TestPluto(unittest.TestCase):
 
-    do_plots = False
+    # do_plots = False
 
     def freq_est(self, y, fs):
         N = len(y)
         T = 1.0 / fs
         yf = np.fft.fft(y)
         yf = np.fft.fftshift(yf)
         xf = np.linspace(-1.0 / (2.0 * T), 1.0 / (2.0 * T), N)
-        if self.do_plots:
-            import matplotlib.pyplot as plt
-
-            fig, ax = plt.subplots()
-            ax.plot(xf, 2.0 / N * np.abs(yf))
-            plt.show()
+        # if self.do_plots:
+        #     import matplotlib.pyplot as plt
+        #
+        #     fig, ax = plt.subplots()
+        #     ax.plot(xf, 2.0 / N * np.abs(yf))
+        #     plt.show()
         indx = np.argmax(np.abs(yf))
         return xf[indx]
 
@@ -110,21 +108,21 @@ def testPlutoDAC(self):
         iq = i + 1j * q
         # Pass through SDR
         sdr.tx(iq)
-        for k in range(5):
+        for _ in range(5):
             data = sdr.rx()
 
         tone_freq = self.freq_est(data, RXFS)
 
-        if self.do_plots:
-            import matplotlib.pyplot as plt
-
-            reals = np.real(data)
-            plt.plot(reals)
-            imags = np.imag(data)
-            plt.plot(imags)
-            plt.xlabel("Samples")
-            plt.ylabel("Amplitude [dbFS]")
-            plt.show()
+        # if self.do_plots:
+        #     import matplotlib.pyplot as plt
+        #
+        #     reals = np.real(data)
+        #     plt.plot(reals)
+        #     imags = np.imag(data)
+        #     plt.plot(imags)
+        #     plt.xlabel("Samples")
+        #     plt.ylabel("Amplitude [dbFS]")
+        #     plt.show()
 
         diff = np.abs(tone_freq - fc)
         self.assertGreater(fc * 0.01, diff, "Frequency offset")