diff options
Diffstat (limited to 'python/dpd/Adapt.py')
| -rw-r--r-- | python/dpd/Adapt.py | 278 |
1 files changed, 97 insertions, 181 deletions
diff --git a/python/dpd/Adapt.py b/python/dpd/Adapt.py index 8108375..9c2c2b6 100644 --- a/python/dpd/Adapt.py +++ b/python/dpd/Adapt.py @@ -9,40 +9,38 @@ This module is used to change settings of ODR-DabMod using the ZMQ remote control socket. """ -import zmq import logging import numpy as np -import os -import datetime +import os.path import pickle +from lib import zmqrc +from typing import List LUT_LEN = 32 FORMAT_POLY = 1 FORMAT_LUT = 2 -def _write_poly_coef_file(coefs_am, coefs_pm, path): - assert (len(coefs_am) == len(coefs_pm)) +def _write_poly_coef_file(coefs_am: List[float], coefs_pm: List[float], path: str) -> None: + assert len(coefs_am) == len(coefs_pm) - f = open(path, 'w') - f.write("{}\n{}\n".format(FORMAT_POLY, len(coefs_am))) - for coef in coefs_am: - f.write("{}\n".format(coef)) - for coef in coefs_pm: - f.write("{}\n".format(coef)) - f.close() + with open(path, 'w') as f: + f.write("{}\n{}\n".format(FORMAT_POLY, len(coefs_am))) + for coef in coefs_am: + f.write("{}\n".format(coef)) + for coef in coefs_pm: + f.write("{}\n".format(coef)) -def _write_lut_file(scalefactor, lut, path): - assert (len(lut) == LUT_LEN) +def _write_lut_file(scalefactor: float, lut: List[complex], path: str) -> None: + assert len(lut) == LUT_LEN - f = open(path, 'w') - f.write("{}\n{}\n".format(FORMAT_LUT, scalefactor)) - for coef in lut: - f.write("{}\n{}\n".format(coef.real, coef.imag)) - f.close() + with open(path, 'w') as f: + f.write("{}\n{}\n".format(FORMAT_LUT, scalefactor)) + for coef in lut: + f.write("{}\n{}\n".format(coef.real, coef.imag)) -def dpddata_to_str(dpddata): +def dpddata_to_str(dpddata) -> str: if dpddata[0] == "poly": coefs_am = dpddata[1] coefs_pm = dpddata[2] @@ -57,215 +55,133 @@ def dpddata_to_str(dpddata): raise ValueError("Unknown dpddata type {}".format(dpddata[0])) class Adapt: - """Uses the ZMQ remote control to change parameters of the DabMod + """Uses the ZMQ remote control to change parameters of the DabMod """ - Parameters - ---------- - port : int - Port at which the ODR-DabMod is listening to connect the - ZMQ remote control. - """ - - def __init__(self, port, coef_path, plot_location): + def __init__(self, port: int, coef_path: str, plot_location: str): logging.debug("Instantiate Adapt object") - self.port = port - self.coef_path = coef_path - self.plot_location = plot_location - self.host = "localhost" - self._context = zmq.Context() - - def _connect(self): - """Establish the connection to ODR-DabMod using - a ZMQ socket that is in request mode (Client). - Returns a socket""" - sock = self._context.socket(zmq.REQ) - poller = zmq.Poller() - poller.register(sock, zmq.POLLIN) - - sock.connect("tcp://%s:%d" % (self.host, self.port)) - - sock.send(b"ping") - - socks = dict(poller.poll(1000)) - if socks: - if socks.get(sock) == zmq.POLLIN: - data = [el.decode() for el in sock.recv_multipart()] - - if data != ['ok']: - raise RuntimeError( - "Invalid ZMQ RC answer to 'ping' at %s %d: %s" % - (self.host, self.port, data)) - else: - sock.close(linger=10) - raise RuntimeError( - "ZMQ RC does not respond to 'ping' at %s %d" % - (self.host, self.port)) - - return sock - - def send_receive(self, message): - """Send a message to ODR-DabMod. It always - returns the answer ODR-DabMod sends back. - - An example message could be - "get sdr txgain" or "set sdr txgain 50" - - Parameter - --------- - message : str - The message string that will be sent to the receiver. - """ - sock = self._connect() - logging.debug("Send message: %s" % message) - msg_parts = message.split(" ") - for i, part in enumerate(msg_parts): - if i == len(msg_parts) - 1: - f = 0 - else: - f = zmq.SNDMORE - - sock.send(part.encode(), flags=f) - - data = [el.decode() for el in sock.recv_multipart()] - logging.debug("Received message: %s" % message) - return data - - def set_txgain(self, gain): - """Set a new txgain for the ODR-DabMod. + self._port = port + self._coef_path = coef_path + self._plot_location = plot_location + self._host = "localhost" + self._mod_rc = zmqrc.ModRemoteControl(self._host, self._port) - Parameters - ---------- - gain : int - new TX gain, in the same format as ODR-DabMod's config file - """ + def set_txgain(self, gain : float) -> None: # TODO this is specific to the B200 if gain < 0 or gain > 89: raise ValueError("Gain has to be in [0,89]") - return self.send_receive("set sdr txgain %.4f" % float(gain)) - - def get_txgain(self): - """Get the txgain value in dB for the ODR-DabMod.""" - # TODO handle failure - return float(self.send_receive("get sdr txgain")[0]) + self._mod_rc.set_param_value("sdr", "txgain", "%.4f" % float(gain)) - def set_rxgain(self, gain): - """Set a new rxgain for the ODR-DabMod. + def get_txgain(self) -> float: + """Get the txgain value in dB, or -1 in case of error""" + try: + return float(self._mod_rc.get_param_value("sdr", "txgain")) + except ValueError as e: + logging.warning(f"Adapt: get_txgain error: {e}") + return -1.0 - Parameters - ---------- - gain : int - new RX gain, in the same format as ODR-DabMod's config file - """ + def set_rxgain(self, gain: float) -> None: # TODO this is specific to the B200 if gain < 0 or gain > 89: raise ValueError("Gain has to be in [0,89]") - return self.send_receive("set sdr rxgain %.4f" % float(gain)) - - def get_rxgain(self): - """Get the rxgain value in dB for the ODR-DabMod.""" - # TODO handle failure - return float(self.send_receive("get sdr rxgain")[0]) - - def set_digital_gain(self, gain): - """Set a new rxgain for the ODR-DabMod. - - Parameters - ---------- - gain : int - new RX gain, in the same format as ODR-DabMod's config file - """ - msg = "set gain digital %.5f" % gain - return self.send_receive(msg) - - def get_digital_gain(self): - """Get the rxgain value in dB for the ODR-DabMod.""" - # TODO handle failure - return float(self.send_receive("get gain digital")[0]) + self._mod_rc.set_param_value("sdr", "rxgain", "%.4f" % float(gain)) + + def get_rxgain(self) -> float: + """Get the rxgain value in dB, or -1 in case of error""" + try: + return float(self._mod_rc.get_param_value("sdr", "rxgain")) + except ValueError as e: + logging.warning(f"Adapt: get_rxgain error: {e}") + return -1.0 + + def set_digital_gain(self, gain: float) -> None: + self._mod_rc.set_param_value("gain", "digital", "%.5f" % float(gain)) + + def get_digital_gain(self) -> float: + """Get the digital gain value in linear scale, or -1 in case + of error""" + try: + return float(self._mod_rc.get_param_value("gain", "digital")) + except ValueError as e: + logging.warning(f"Adapt: get_digital_gain error: {e}") + return -1.0 def get_predistorter(self): """Load the coefficients from the file in the format given in the README, return ("poly", [AM coef], [PM coef]) or ("lut", scalefactor, [LUT entries]) """ - f = open(self.coef_path, 'r') - lines = f.readlines() - predistorter_format = int(lines[0]) - if predistorter_format == FORMAT_POLY: - coefs_am_out = [] - coefs_pm_out = [] - n_coefs = int(lines[1]) - coefs = [float(l) for l in lines[2:]] - i = 0 - for c in coefs: - if i < n_coefs: - coefs_am_out.append(c) - elif i < 2 * n_coefs: - coefs_pm_out.append(c) - else: - raise ValueError( - 'Incorrect coef file format: too many' - ' coefficients in {}, should be {}, coefs are {}' - .format(self.coef_path, n_coefs, coefs)) - i += 1 - f.close() - return 'poly', coefs_am_out, coefs_pm_out - elif predistorter_format == FORMAT_LUT: - scalefactor = int(lines[1]) - coefs = np.array([float(l) for l in lines[2:]], dtype=np.float32) - coefs = coefs.reshape((-1, 2)) - lut = coefs[..., 0] + 1j * coefs[..., 1] - if len(lut) != LUT_LEN: - raise ValueError("Incorrect number of LUT entries ({} expected {})".format(len(lut), LUT_LEN)) - return 'lut', scalefactor, lut - else: - raise ValueError("Unknown predistorter format {}".format(predistorter_format)) + with open(self._coef_path, 'r') as f: + lines = f.readlines() + predistorter_format = int(lines[0]) + if predistorter_format == FORMAT_POLY: + coefs_am_out = [] + coefs_pm_out = [] + n_coefs = int(lines[1]) + coefs = [float(l) for l in lines[2:]] + for i, c in enumerate(coefs): + if i < n_coefs: + coefs_am_out.append(c) + elif i < 2 * n_coefs: + coefs_pm_out.append(c) + else: + raise ValueError( + 'Incorrect coef file format: too many' + ' coefficients in {}, should be {}, coefs are {}' + .format(self._coef_path, n_coefs, coefs)) + return 'poly', coefs_am_out, coefs_pm_out + elif predistorter_format == FORMAT_LUT: + scalefactor = int(lines[1]) + coefs = np.array([float(l) for l in lines[2:]], dtype=np.float32) + coefs = coefs.reshape((-1, 2)) + lut = coefs[..., 0] + 1j * coefs[..., 1] + if len(lut) != LUT_LEN: + raise ValueError("Incorrect number of LUT entries ({} expected {})".format(len(lut), LUT_LEN)) + return 'lut', scalefactor, lut + else: + raise ValueError("Unknown predistorter format {}".format(predistorter_format)) - def set_predistorter(self, dpddata): + def set_predistorter(self, dpddata) -> None: """Update the predistorter data in the modulator. Takes the same tuple format as argument than the one returned get_predistorter()""" if dpddata[0] == "poly": coefs_am = dpddata[1] coefs_pm = dpddata[2] - _write_poly_coef_file(coefs_am, coefs_pm, self.coef_path) + _write_poly_coef_file(coefs_am, coefs_pm, self._coef_path) elif dpddata[0] == "lut": scalefactor = dpddata[1] lut = dpddata[2] - _write_lut_file(scalefactor, lut, self.coef_path) + _write_lut_file(scalefactor, lut, self._coef_path) else: raise ValueError("Unknown predistorter '{}'".format(dpddata[0])) - return self.send_receive("set memlesspoly coeffile {}".format(self.coef_path)) + self._mod_rc.set_param_value("memlesspoly", "coefffile", self._coef_path) - def dump(self, path=None): + def dump(self, path: str) -> None: """Backup current settings to a file""" - dt = datetime.datetime.now().isoformat() - if path is None: - if self.plot_location is not None: - path = self.plot_location + "/" + dt + "_adapt.pkl" - else: - raise Exception("Cannot dump Adapt without either plot_location or path set") + d = { "txgain": self.get_txgain(), "rxgain": self.get_rxgain(), "digital_gain": self.get_digital_gain(), "predistorter": self.get_predistorter() } + with open(path, "wb") as f: pickle.dump(d, f) - return path - - def load(self, path): + def load(self, path: str) -> None: """Restore settings from a file""" with open(path, "rb") as f: d = pickle.load(f) - self.set_txgain(d["txgain"]) + self.set_txgain(0) self.set_digital_gain(d["digital_gain"]) self.set_rxgain(d["rxgain"]) self.set_predistorter(d["predistorter"]) + self.set_txgain(d["txgain"]) # The MIT License (MIT) # -# Copyright (c) 2017 Andreas Steger, Matthias P. Braendli +# Copyright (c) 2018 Matthias P. Braendli +# Copyright (c) 2017 Andreas Steger # # Permission is hereby granted, free of charge, to any person obtaining a copy # of this software and associated documentation files (the "Software"), to deal |