st-ten-1/src/components/multicomp_730424.py

import sys
import traceback

if "--sim-serial" in sys.argv:
    from components.dummies.serial import serial
else:
    import serial

from .component import Component


class Multicomp730424(Component):
    def __init__(self, config=None, name=None, period=1, lazy=True, paused=False, threaded=True):
        self.conn = None
        super().__init__(config=config, name=name, period=period, lazy=lazy, paused=paused, threaded=threaded)

    def config_changed(self):
        self.port = self.config[self.name]["port"]
        self.baudrate = int(self.config[self.name]["baudrate"])
        self.stopbits = getattr(serial, self.config[self.name].get("stopbits", "stopbits_one").upper())
        self.parity = getattr(serial, self.config[self.name].get("parity", "parity_none").upper())
        self.bytesize = getattr(serial, self.config[self.name].get("bytesize", "eightbits").upper())
        if self.conn is not None:
            self.conn.close()
        self.conn = serial.Serial(
            self.port,
            baudrate=self.baudrate,
            stopbits=self.stopbits,
            parity=self.parity,
            bytesize=self.bytesize,
        )

    def set_resistance_scale(self, resistance):
        self.write(f"CONF:RES {resistance}")

    def read(self):
        try:
            response = self.conn.readline()
        except serial.serialutil.SerialException:
            self.log.exception(traceback.format_exc())
            return None
        try:
            decoded = response.decode("ascii")
        except UnicodeDecodeError:
            return response
        try:
            value = float(decoded)
        except ValueError:
            return decoded
        return value

    def write(self, command):
        try:
            self.conn.write(f"{command}\n".encode("ascii"))
        except serial.serialutil.SerialException:
            self.log.exception(traceback.format_exc())

    def _get(self):
        info = {}
        try:
            for value_name, command in {
                "scale": "CONF:RES?",
                "measure": "MEAS?",
            }.items():
                self.write(command)
                info[value_name] = self.read()
        except serial.serialutil.SerialException:
            self.log.exception(traceback.format_exc())
        self.log.debug(str(info))
        super()._get([info])