ruff

Author: bgeiser

python/art-tracker-mock/main.py

@@ -7,6 +7,7 @@
 
 from scapy.all import rdpcap, IP, UDP
 
+
 @dataclass
 class Packet:
     payload: bytes
@@ -34,7 +35,7 @@ def _read_packets(pcap_file: Path, pcap_ip: ip_address, pcap_ports: list[int]) -
                 if last_time is not None:
                     delay = float(pkt.time) - last_time
                 else:
-                    delay = 0.
+                    delay = 0.0
                 res.append(Packet(payload, udp_layer.dport, delay))
                 last_time = float(pkt.time)
     return res

python/ayf/audioio/ayfaudiobuf.py

@@ -1,39 +1,40 @@
-
 import numpy as np
 from enum import Enum
 
 # ======================================================================
 # ======================================================================
 
+
 class ProcessingType(Enum):
     UNDEFINED = 0
-    I_SINGLEBUF = 1 # interleaved, singlebuffer [0 0 0 1 1 1]
-    NOI_SINGLEBUF = 2 # non-interleaved, singlebuffer [0 0 0 1 1 1]
-    NOI_BUFARR = 3 # non-interleaved, buffer array [[0 0 0][1 1 1]]
+    I_SINGLEBUF = 1  # interleaved, singlebuffer [0 0 0 1 1 1]
+    NOI_SINGLEBUF = 2  # non-interleaved, singlebuffer [0 0 0 1 1 1]
+    NOI_BUFARR = 3  # non-interleaved, buffer array [[0 0 0][1 1 1]]
+
 
 # ======================================================================
 # ======================================================================
- 
-class ayfaudio_buf_param:
 
+
+class ayfaudio_buf_param:
     def __init__(self):
         self.tp = ProcessingType.UNDEFINED
         self.chans = 0
         self.bs = 0
         self.sr = 0
 
-    def params_set(self, operation = ProcessingType.I_SINGLEBUF, chans = 1, bs = 1024, sr = 48000, arith = 'double'):
-        
+    def params_set(self, operation=ProcessingType.I_SINGLEBUF, chans=1, bs=1024, sr=48000, arith="double"):
+
         self.tp = operation
         self.chans = chans
         self.bs = bs
         self.sr = sr
         self.arith = arith
 
-class ayfaudio_buf:
 
+class ayfaudio_buf:
     def __init__(self):
-        
+
         self.params = ayfaudio_buf_param()
 
         self.c = 0
@@ -43,8 +44,8 @@ def __init__(self):
 
     def allocate(self, operation, nChans, fsize, srate, arith):
 
-        self.params.params_set(chans = nChans, operation = operation, bs = fsize, sr = srate, arith = arith)
-        
+        self.params.params_set(chans=nChans, operation=operation, bs=fsize, sr=srate, arith=arith)
+
         match self.params.tp:
             case ProcessingType.I_SINGLEBUF:
                 self.c = 1
@@ -65,8 +66,7 @@ def allocate(self, operation, nChans, fsize, srate, arith):
                 self.fld = np.zeros(self.l, dtype=arith)
 
             case _:
-                assert(False)
-
+                assert False
 
     def deallocate(self):
         self.params.tp = ProcessingType.UNDEFINED
@@ -78,23 +78,26 @@ def deallocate(self):
         self.fld = None
         self.fIdx = 0
 
+
 # ======================================================================
 # ======================================================================
 
+
 class ayfaudio_data:
-    def __init__(self, srate = 48000, nChans = 2):
+    def __init__(self, srate=48000, nChans=2):
         self.sampling_rate = srate
         self.nChans = nChans
-  
+
+
 # ======================================================================
 # ======================================================================
 
-class ayfaudio_outbuf:
 
-    def __init__(self, maxLength = 48000, nChans = 2, fsize = 128):
+class ayfaudio_outbuf:
+    def __init__(self, maxLength=48000, nChans=2, fsize=128):
         self.buf = None
         self.data = ayfaudio_data()
-        self.duration_samples = 0        
+        self.duration_samples = 0
         self.maxLength = maxLength
         self.fsize = fsize
         self.nChans = nChans
@@ -103,40 +106,40 @@ def __init__(self, maxLength = 48000, nChans = 2, fsize = 128):
     def postprocess_oneBuf_noi_singlebuf(self, outBufHdl):
         cpToStart = 0
         idxStart = outBufHdl.fIdx * outBufHdl.bs
-        idxStop =  (outBufHdl.fIdx+1) * outBufHdl.bs
+        idxStop = (outBufHdl.fIdx + 1) * outBufHdl.bs
         idxStart = min(idxStart, self.duration_samples)
         idxStop = min(idxStop, self.duration_samples)
         cpNum = idxStop - idxStart
-        if(cpNum > 0):
+        if cpNum > 0:
             for idxChan in range(0, self.nChans):
-                self.buf[idxChan][idxStart:idxStart + cpNum] = outBufHdl.fld[cpToStart:cpToStart + cpNum]
+                self.buf[idxChan][idxStart : idxStart + cpNum] = outBufHdl.fld[cpToStart : cpToStart + cpNum]
                 cpToStart += outBufHdl.bs
-            
+
     def postprocess_oneBuf_noi_bufarr(self, outBufHdl, idxFrame):
-        assert(0)
+        assert 0
 
     def postprocess_single_frame(self, outBufHdl, idxFrame):
         match outBufHdl.tp:
             case ProcessingType.UNDEFINED:
-                print('Error')
-                assert(0)
+                print("Error")
+                assert 0
 
             case ProcessingType.NOI_SINGLEBUF:
                 self.postprocess_oneBuf_noi_singlebuf(outBufHdl)
 
             case ProcessingType.NOI_BUFARR:
                 self.postprocess_oneBuf_noi_bufarr(outBufHdl, idxFrame)
-        
+
         self.fIdx = outBufHdl.fIdx
 
     def param_get(self):
         return self.data
 
-    def start(self, arith, nFrames = -1):
-        
-        if(nFrames > 0):
+    def start(self, arith, nFrames=-1):
+
+        if nFrames > 0:
             self.duration_samples = nFrames * self.fsize
         else:
             self.duration_samples = self.maxLength * self.fsize
 
-        self.buf = np.empty((self.nChans, int(self.duration_samples)),dtype=arith)
+        self.buf = np.empty((self.nChans, int(self.duration_samples)), dtype=arith)

python/ayf/audioio/ayfaudioif.py

@@ -1,6 +1,6 @@
-import numpy as np
 from . import ayfaudiobuf as ayfbuf
 
+
 class ayfaudio_in:
     # @abstractmethod
     def prepare_data(self, formatSystem) -> None: ...
@@ -26,15 +26,17 @@ def progress(self, inBufHdl) -> float: ...
     # =============================================================================
     # =============================================================================
 
+
 class ayfaudio_processor:
     # @abstractmethod
-    def prepare(self, numIn = 1, numOut = 1, bSize = 128, sRate = 48000) -> int: ...
+    def prepare(self, numIn=1, numOut=1, bSize=128, sRate=48000) -> int: ...
 
     # @abstractmethod
     def process(self, inBuf, outBuf) -> int: ...
 
     # @abstractmethod
     def postprocess(self) -> int: ...
 
+
 # ======================================================================
 # ======================================================================

python/ayf/audioio/ayfaudioinpulse.py

@@ -1,25 +1,22 @@
-import pathlib
 import numpy as np
 import math
-import audiofile as af
 
 from . import ayfaudioif as ayfif
-from . import ayfaudioio as ayfio
 from . import ayfaudiobuf as ayfbuf
 
+
 # ========================================================================
 class ayfaudio_inbuffered(ayfif.ayfaudio_in):
-
     def prepare_data(self, formatSystem):
-            
-        if(formatSystem == "single"):
-            if(self.buf.dtype == "float64"):
+
+        if formatSystem == "single":
+            if self.buf.dtype == "float64":
                 self.buf = np.float32(self.buf)
-        elif(formatSystem == "double"):
-            if(self.buf.dtype == "float32"):
+        elif formatSystem == "double":
+            if self.buf.dtype == "float32":
                 self.buf = np.float64(self.buf)
         else:
-            assert(False)
+            assert False
 
         self.fIdx = 0
 
@@ -31,117 +28,117 @@ def clear_data(self):
 
     def param_get(self):
         return self.data
-        
+
     def prepare_oneBuf_noi_singlebuf(self, inBufHdl):
         inBufHdl.fld.fill(0)
         cpToStart = 0
-        if(inBufHdl.fIdx * inBufHdl.bs >= self.duration_samples):
+        if inBufHdl.fIdx * inBufHdl.bs >= self.duration_samples:
             return False
 
         idxStart = inBufHdl.fIdx * inBufHdl.bs
-        idxStop =  (inBufHdl.fIdx+1) * inBufHdl.bs
-        idxStop = min(idxStop, self.duration_samples)        
+        idxStop = (inBufHdl.fIdx + 1) * inBufHdl.bs
+        idxStop = min(idxStop, self.duration_samples)
         cpNum = idxStop - idxStart
         chansLim = min(self.data.nChans, inBufHdl.chans)
         for idxChan in range(0, chansLim):
-            inBufHdl.fld[cpToStart:cpToStart + cpNum] = self.buf[idxChan][idxStart:idxStart + cpNum]
-            cpToStart += inBufHdl.bs        
+            inBufHdl.fld[cpToStart : cpToStart + cpNum] = self.buf[idxChan][idxStart : idxStart + cpNum]
+            cpToStart += inBufHdl.bs
         return True
 
-
     def prepare_oneBuf_noi_bufarr(self, inBufHdl, frameIdx):
-        assert(0)
+        assert 0
 
     def prepare_oneBuf_i_singlebuf(self, inBufHdl: ayfbuf.ayfaudio_buf):
-        
+
         inBufHdl.fld.fill(0)
         nSamples = inBufHdl.params.bs * inBufHdl.params.chans
-        
+
         # Only full frames. We need samples from frame 0 to frame 1 - therefore "+1"
-        if((self.fIdx + 1) * nSamples >= self.duration_samples):
+        if (self.fIdx + 1) * nSamples >= self.duration_samples:
             return False
-        
+
         # Extract buffer here!
         idxStart = self.fIdx * nSamples
-        idxStop =  (self.fIdx+1) * nSamples
-        idxStop = min(idxStop, (self.duration_samples * inBufHdl.params.chans)) 
+        idxStop = (self.fIdx + 1) * nSamples
+        idxStop = min(idxStop, (self.duration_samples * inBufHdl.params.chans))
         nSamples = idxStop - idxStart
-                
+
         # Might be a 1D-array or a 2D-array
-        firstBuf = self.buf   
-        if(len(self.buf) == 2):
+        firstBuf = self.buf
+        if len(self.buf) == 2:
             firstBuf = self.buf[0]
 
         # Copy content and the frame
-        inBufHdl.fld[0:nSamples] = firstBuf[idxStart:idxStop]  
+        inBufHdl.fld[0:nSamples] = firstBuf[idxStart:idxStop]
         inBufHdl.fIdx = self.fIdx
         return True
-    
+
     def prepare_single_frame(self, inBufHdl, frameIdx):
         match inBufHdl.tp:
             case ayfbuf.ProcessingType.NOI_SINGLEBUF:
-                retVal = self.prepare_oneBuf_noi_singlebuf(inBufHdl) # No frameindex required as it handles that internally
+                retVal = self.prepare_oneBuf_noi_singlebuf(inBufHdl)  # No frameindex required as it handles that internally
 
             case ayfbuf.ProcessingType.NOI_BUFARR:
                 retVal = self.prepare_oneBuf_noi_bufarr(inBufHdl, frameIdx)
 
             case ayfbuf.ProcessingType.I_SINGLEBUF:
-                retVal = self.prepare_oneBuf_i_singlebuf(inBufHdl) # No frameindex required as it handles that internally
+                retVal = self.prepare_oneBuf_i_singlebuf(inBufHdl)  # No frameindex required as it handles that internally
             case _:
-                print('Error')
-                assert(0)
+                print("Error")
+                assert 0
 
-        if(retVal == True):
+        if retVal:
             inBufHdl.fIdx = self.fIdx
-            self.fIdx = self.fIdx + 1    
+            self.fIdx = self.fIdx + 1
         return retVal
 
     def start(self, inBufHdl):
-        return math.ceil( self.duration_samples/inBufHdl.params.bs)
+        return math.ceil(self.duration_samples / inBufHdl.params.bs)
 
-    def stop(self) -> int: ...
-        # Do nothing                
+    def stop(self) -> int:
+        ...
+        # Do nothing
 
     def progress(self, inBufHdl: ayfbuf.ayfaudio_buf):
-        return (self.fIdx* inBufHdl.params.bs)/ self.duration_samples
-    
+        return (self.fIdx * inBufHdl.params.bs) / self.duration_samples
+
+
 # ===========================================================================
 # ===========================================================================
 
-class ayfaudio_inpulse(ayfaudio_inbuffered):
 
-    def __init__(self, duration_secs = 10, samplerate = 48000, nChans = 1, format = 'double'):
+class ayfaudio_inpulse(ayfaudio_inbuffered):
+    def __init__(self, duration_secs=10, samplerate=48000, nChans=1, format="double"):
 
         self.data = ayfbuf.ayfaudio_data()
         self.data.sampling_rate = samplerate
-        self.duration_samples = duration_secs * self.data.sampling_rate 
+        self.duration_samples = duration_secs * self.data.sampling_rate
         self.fIdx = 0
         self.buf = None
 
         print("### AYFAUDIOINPULSE -- Creating pulse input.")
 
-        if(format == "single"):
-            self.buf = np.zeros(nChans * self.duration_samples, dtype="float32") 
+        if format == "single":
+            self.buf = np.zeros(nChans * self.duration_samples, dtype="float32")
 
-        elif(format == "double"):
-            
-        # self.buf = np.float64(self.buf)
+        elif format == "double":
+            # self.buf = np.float64(self.buf)
             self.buf = np.zeros(nChans * self.duration_samples, dtype="float64")
         else:
-            assert(False)
+            assert False
 
         # Fill very first sample to realize the pulse
         for idxChan in range(0, nChans):
             self.buf[idxChan] = 1.0
-                    
-        szArr = self.buf.shape
-        if(len(szArr) == 2):
 
+        szArr = self.buf.shape
+        if len(szArr) == 2:
             self.duration_samples = szArr[1]
             self.data.nChans = szArr[0]
         else:
             self.duration_samples = szArr[0]
             self.data.nChans = 1
 
+
 # ==========================================================================
 # ==========================================================================