path: root/_receiver/krakenSDR_receiver.py

# KrakenSDR Receiver

# Copyright (C) 2018-2021  Carl Laufer, Tamás Pető
#
#
#   This program is free software: you can redistribute it and/or modify
#   it under the terms of the GNU General Public License as published by
#   the Free Software Foundation, either version 3 of the License, or
#   any later version.
#
#   This program is distributed in the hope that it will be useful,
#   but WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
#   GNU General Public License for more details.
#
#   You should have received a copy of the GNU General Public License
#   along with this program.  If not, see <https://www.gnu.org/licenses/>.
#

# -*- coding: utf-8 -*-

# Import built-in modules
import sys
import os
import time
from struct import pack, unpack
import socket
import _thread
from threading import Lock
import queue 
import logging
#import copy

# Import third party modules
import numpy as np
from scipy import signal
from iq_header import IQHeader
from shmemIface import inShmemIface
class ReceiverRTLSDR():
    
    def __init__(self, data_que, data_interface = "eth", logging_level=10):     
        """
        Parameter:
        ----------        
            :param: data_que: Que to communicate with the UI (web iface/Qt GUI)        
            :param: data_interface: This field is configured by the GUI during instantiation. 
                                    Valid values are the followings:
                                    "eth"  : The module will receiver IQ frames through an Ethernet connection
                                    "shmem": The module will receiver IQ frames through a shared memory interface
            :type : data_interface: string
        """
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging_level)

        # DAQ parameters
        # These values are used by default to configure the DAQ through the configuration interface
        # Values are configured externally upon configuration request
        self.daq_center_freq   = 100 # MHz
        self.daq_rx_gain       = 0   # [dB]
        self.daq_squelch_th_dB = 0

        # UI interface
        self.data_que = data_que

        # IQ data interface
        self.data_interface = data_interface

        # -> Ethernet
        self.receiver_connection_status = False
        self.port = 5000
        self.rec_ip_addr = "127.0.0.1" # Configured by the GUI prior to connection request
        self.socket_inst = socket.socket()
        self.receiverBufferSize = 2 ** 18  # Size of the Ethernet receiver buffer measured in bytes

        # -> Shared memory
        root_path          = os.path.dirname(os.path.dirname(os.path.realpath(__file__)))
        daq_path           = os.path.join(os.path.dirname(root_path),"heimdall_daq_fw")
        self.daq_shmem_control_path = os.path.join(os.path.join(daq_path,"Firmware"),"_data_control/")
        self.init_data_iface()

        # Control interface
        self.ctr_iface_socket = socket.socket()
        self.ctr_iface_port = 5001
        self.ctr_iface_thread_lock = Lock() # Used to synchronize the operation of the ctr_iface thread

        self.iq_frame_bytes = None
        self.iq_samples = None
        self.iq_header = IQHeader()
        self.M = 0 # Number of receiver channels, updated after establishing connection

    def init_data_iface(self):
        if self.data_interface == "shmem":
            # Open shared memory interface to capture the DAQ firmware output
            self.in_shmem_iface = inShmemIface("delay_sync_iq", self.daq_shmem_control_path)
            if not self.in_shmem_iface.init_ok:
                self.logger.critical("Shared memory initialization failed")
                self.in_shmem_iface.destory_sm_buffer()
                return -1
        return 0


    def eth_connect(self):
        """
            Compatible only with DAQ firmwares that has the IQ streaming mode.
            HeIMDALL DAQ Firmware version: 1.0 or later
        """
        try:
            if not self.receiver_connection_status:
                if self.data_interface == "eth":
                    # Establlish IQ data interface connection
                    self.socket_inst.connect((self.rec_ip_addr, self.port))
                    self.socket_inst.sendall(str.encode('streaming'))
                    test_iq = self.receive_iq_frame()

                    self.M = self.iq_header.active_ant_chs

                # Establish control interface connection
                self.ctr_iface_socket.connect((self.rec_ip_addr, self.ctr_iface_port))
                self.receiver_connection_status = True
                self.ctr_iface_init()
                self.logger.info("CTR INIT Center freq: {0}".format(self.daq_center_freq))
                self.set_center_freq(self.daq_center_freq)
                self.set_if_gain(self.daq_rx_gain)
                self.set_squelch_threshold(self.daq_squelch_th_dB)
        except:
            errorMsg = sys.exc_info()[0]
            self.logger.error("Error message: "+str(errorMsg))
            self.receiver_connection_status = False
            self.logger.error("Unexpected error: {0}".format(sys.exc_info()[0]))

            # Re-instantiating sockets
            self.socket_inst = socket.socket()
            self.ctr_iface_socket = socket.socket()
            return -1

        self.logger.info("Connection established")
        que_data_packet = []
        que_data_packet.append(['conn-ok',])
        self.data_que.put(que_data_packet)

    def eth_close(self):
        """
            Close Ethernet conenctions including the IQ data and the control interfaces
        """
        try:
            if self.receiver_connection_status:
                if self.data_interface == "eth":
                    self.socket_inst.sendall(str.encode('q')) # Send exit message
                    self.socket_inst.close()
                    self.socket_inst = socket.socket() # Re-instantiating socket

                # Close control interface connection
                exit_message_bytes=("EXIT".encode()+bytearray(124))
                self.ctr_iface_socket.send(exit_message_bytes)
                self.ctr_iface_socket.close()
                self.ctr_iface_socket = socket.socket()

            self.receiver_connection_status = False
            que_data_packet = []
            que_data_packet.append(['disconn-ok',])
            self.data_que.put(que_data_packet)
        except:
            errorMsg = sys.exc_info()[0]
            self.logger.error("Error message: {0}".format(errorMsg))
            return -1

        if self.data_interface == "shmem":
            self.in_shmem_iface.destory_sm_buffer()

        return 0

    def get_iq_online(self):
        """
            This function obtains a new IQ data frame through the Ethernet IQ data or the shared memory interface
        """

        # Check connection
        if not self.receiver_connection_status:
            fail = self.eth_connect()
            if fail:
                return -1

        if self.data_interface == "eth":
            self.socket_inst.sendall(str.encode("IQDownload")) # Send iq request command
            self.iq_samples = self.receive_iq_frame()

        elif self.data_interface == "shmem":
            active_buff_index = self.in_shmem_iface.wait_buff_free()
            if active_buff_index < 0 or active_buff_index > 1:
                self.logger.info("Terminating.., signal: {:d}".format(active_buff_index))
                return -1

            buffer = self.in_shmem_iface.buffers[active_buff_index]
            iq_header_bytes = buffer[0:1024].tobytes()
            self.iq_header.decode_header(iq_header_bytes)

            # Inititalization from header - Set channel numbers
            if self.M == 0:
                self.M = self.iq_header.active_ant_chs

            incoming_payload_size = self.iq_header.cpi_length*self.iq_header.active_ant_chs*2*int(self.iq_header.sample_bit_depth/8)
            if incoming_payload_size > 0:
                iq_samples_in = (buffer[1024:1024 + incoming_payload_size].view(dtype=np.complex64))\
                                .reshape(self.iq_header.active_ant_chs, self.iq_header.cpi_length)
                self.iq_samples = iq_samples_in.copy() # Must be .copy

            self.in_shmem_iface.send_ctr_buff_ready(active_buff_index)

    def receive_iq_frame(self):
        """
            Called by the get_iq_online function. Receives IQ samples over the establed Ethernet connection
        """
        total_received_bytes = 0
        recv_bytes_count = 0
        iq_header_bytes = bytearray(self.iq_header.header_size)  # allocate array
        view = memoryview(iq_header_bytes)  # Get buffer

        self.logger.debug("Starting IQ header reception")

        while total_received_bytes < self.iq_header.header_size:
            # Receive into buffer
            recv_bytes_count = self.socket_inst.recv_into(view, self.iq_header.header_size-total_received_bytes)
            view = view[recv_bytes_count:]  # reset memory region
            total_received_bytes += recv_bytes_count

        self.iq_header.decode_header(iq_header_bytes)
        # Uncomment to check the content of the IQ header
        #self.iq_header.dump_header()

        incoming_payload_size = self.iq_header.cpi_length*self.iq_header.active_ant_chs*2*int(self.iq_header.sample_bit_depth/8)
        if incoming_payload_size > 0:
            # Calculate total bytes to receive from the iq header data
            total_bytes_to_receive = incoming_payload_size
            receiver_buffer_size = 2**18

            self.logger.debug("Total bytes to receive: {:d}".format(total_bytes_to_receive))

            total_received_bytes = 0
            recv_bytes_count = 0
            iq_data_bytes = bytearray(total_bytes_to_receive + receiver_buffer_size)  # allocate array
            view = memoryview(iq_data_bytes)  # Get buffer

            while total_received_bytes < total_bytes_to_receive:
                # Receive into buffer
                recv_bytes_count = self.socket_inst.recv_into(view, receiver_buffer_size)
                view = view[recv_bytes_count:]  # reset memory region
                total_received_bytes += recv_bytes_count

            self.logger.debug(" IQ data succesfully received")

            # Convert raw bytes to Complex float64 IQ samples
            self.iq_samples = np.frombuffer(iq_data_bytes[0:total_bytes_to_receive], dtype=np.complex64).reshape(self.iq_header.active_ant_chs, self.iq_header.cpi_length)

            self.iq_frame_bytes =  bytearray()+iq_header_bytes+iq_data_bytes
            return self.iq_samples
        else:
             return 0

    def set_squelch_threshold(self, threshold_dB):
        """
            Configures the threshold level of the squelch module in the DAQ FW through the control interface
        """
        if self.receiver_connection_status: # Check connection
            self.daq_squelch_th_dB      = threshold_dB
            if threshold_dB == -80: threshold = 0
            else: threshold = 10**(threshold_dB/20)

            # Assembling message
            cmd="STHU"
            th_bytes=pack("f",threshold)
            msg_bytes=(cmd.encode()+th_bytes+bytearray(120))
            try:
                _thread.start_new_thread(self.ctr_iface_communication, (msg_bytes,))
            except:
                errorMsg = sys.exc_info()[0]
                self.logger.error("Unable to start communication thread")
                self.logger.error("Error message: {:s}".format(errorMsg))

    def ctr_iface_init(self):
        """
            Initialize connection with the DAQ FW through the control interface
        """
        if self.receiver_connection_status: # Check connection
            # Assembling message            
            cmd="INIT"            
            msg_bytes=(cmd.encode()+bytearray(124))           
            try:                
                _thread.start_new_thread(self.ctr_iface_communication, (msg_bytes,))
            except:                
                errorMsg = sys.exc_info()[0]        
                self.logger.error("Unable to start communication thread")
                self.logger.error("Error message: {:s}".format(errorMsg))

    def ctr_iface_communication(self, msg_bytes):
        """
            Handles communication on the control interface with the DAQ FW

            Parameters:
            -----------

                :param: msg: Message bytes, that will be sent ont the control interface
                :type:  msg: Byte array                
        """ 
        self.ctr_iface_thread_lock.acquire()
        self.logger.debug("Sending control message")        
        self.ctr_iface_socket.send(msg_bytes)
        
        # Waiting for the command to take effect
        reply_msg_bytes = self.ctr_iface_socket.recv(128)    

        self.logger.debug("Control interface communication finished")
        self.ctr_iface_thread_lock.release()
        
        status = reply_msg_bytes[0:4].decode()
        if status == "FNSD":
            self.logger.info("Reconfiguration succesfully finished")
            que_data_packet = []
            que_data_packet.append(['config-ok',])
            self.data_que.put(que_data_packet)
            
        else:
            self.logger.error("Failed to set the requested parameter, reply: {0}".format(status))
        
    def set_center_freq(self, center_freq):
        """
            Configures the RF center frequency of the receiver through the control interface
        
            Paramters:
            ----------
                :param: center_freq: Required center frequency to set [Hz]
                :type:  center_freq: float
        """
        if self.receiver_connection_status: # Check connection
            self.daq_center_freq = int(center_freq)
            # Set center frequency
            cmd="FREQ"
            freq_bytes=pack("Q",int(center_freq))
            msg_bytes=(cmd.encode()+freq_bytes+bytearray(116))           
            try:                
                _thread.start_new_thread(self.ctr_iface_communication, (msg_bytes,))            
            except:                
                errorMsg = sys.exc_info()[0]        
                self.logger.error("Unable to start communication thread")
                self.logger.error("Error message: {:s}".format(errorMsg))
    
    def set_if_gain(self, gain):
        """
            Configures the IF gain of the receiver through the control interface
        
            Paramters:
            ----------
                :param: gain: IF gain value [dB]
                :type:  gain: int
        """                
        if self.receiver_connection_status: # Check connection
            self.daq_rx_gain     = gain
            
            # Set center frequency
            cmd="GAIN"
            gain_list=[297, 37] #[int(gain*10)]*self.M
            gain_bytes=pack("I"*self.M, *gain_list)
            msg_bytes=(cmd.encode()+gain_bytes+bytearray(128-(self.M+1)*4))  
            try:                
                _thread.start_new_thread(self.ctr_iface_communication, (msg_bytes,))            
            except:                
                errorMsg = sys.exc_info()[0]        
                self.logger.error("Unable to start communication thread")
                self.logger.error("Error message: {:s}".format(errorMsg))

    def close(self):
        """
            Disconnet the receiver module and the DAQ FW
        """
        self.eth_close()