plc-sim/server/instrument_controller.py

import copy
import os
import random
import threading
import time
import typing
from collections import deque
from enum import Enum
from typing import Optional

import cv2
import numpy as np
import pyvisa
from numpy import ndarray as mat
from pyvisa import ResourceManager as scpiManager
from pyvisa.resources import MessageBasedResource as scpiConnection
from flask import Flask, Response
from flask import Flask, jsonify
from flask_cors import CORS

from program_public_tools import ProgramPublicTools


class ScpiInstrument(object):
    connection: scpiConnection
    name: str
    response: str


class InstrumentResponseMark(object):
    def __init__(self):
        self.digital_multimeter: str = "DM3068"
        self.digital_oscilloscope: str = "DHO1204"
        self.waveform_generator: str = "DG4202"
        self.analog_electronic_load: str = "DL3021A"
        self.__check_env_value()

    def __check_env_value(self):
        config_digital_multimeter = self.__get_env("PLC_SIM_SERVER_DIGITAL_MULTIMETER")
        config_digital_oscilloscope = self.__get_env("PLC_SIM_SERVER_DIGITAL_OSCILLOSCOPE")
        config_waveform_generator = self.__get_env("PLC_SIM_SERVER_WAVEFORM_GENERATOR")
        config_analog_electronic_load = self.__get_env("PLC_SIM_SERVER_ANALOG_ELECTRONIC_LOAD")
        print(f"[Instrument Controller] Using config: PLC_SIM_SERVER_DIGITAL_MULTIMETER = {config_digital_multimeter}")
        print(f"[Instrument Controller] Using config: PLC_SIM_SERVER_DIGITAL_OSCILLOSCOPE = {config_digital_oscilloscope}")
        print(f"[Instrument Controller] Using config: PLC_SIM_SERVER_WAVEFORM_GENERATOR = {config_waveform_generator}")
        print(f"[Instrument Controller] Using config: PLC_SIM_SERVER_ANALOG_ELECTRONIC_LOAD = {config_analog_electronic_load}")
        if config_digital_multimeter is not None:
            self.digital_multimeter = config_digital_multimeter
        if config_digital_oscilloscope is not None:
            self.digital_oscilloscope = config_digital_oscilloscope
        if config_waveform_generator is not None:
            self.waveform_generator = config_waveform_generator
        if config_analog_electronic_load is not None:
            self.analog_electronic_load = config_analog_electronic_load

    @staticmethod
    def __get_env(name: str) -> Optional[str]:
        return os.environ.get(name, None)


class InstrumentControllerConfig(object):
    connection_timeout_ms: int = 5000
    response_mark: InstrumentResponseMark = InstrumentResponseMark()
    buffer_path: str = "./temp/"


class InstrumentServices(object):
    def __init__(self):
        self.digital_multimeter: Optional[DigitalMultimeterService] = None
        self.digital_oscilloscope: Optional[DigitalOscilloscopeService] = None
        self.waveform_generator: Optional[WaveformGeneratorService] = None
        self.analog_electronic_load: Optional[AnalogElectronicLoadService] = None

    def check(self) -> list[str]:
        not_connect_list: list[str] = []
        for attr_name, attr_value in vars(self).items():
            if attr_value is None:
                not_connect_list.append(attr_name)
        return not_connect_list


class InstrumentController(object):
    def __init__(self, pubtools: ProgramPublicTools):
        self.__scpi_manager: scpiManager = self.__init_scpi_manager()
        self.config: InstrumentControllerConfig = InstrumentControllerConfig()
        self.__pubtools: ProgramPublicTools = pubtools
        self.__scpi_instrument_list: list[ScpiInstrument] = []
        self.services: InstrumentServices = InstrumentServices()
        self.__retry_times: int = 1
        self.__max_connection_number: int = 4
        self.__connection_timeout_signal: int = 0

    @staticmethod
    def __init_scpi_manager() -> scpiManager:
        return pyvisa.ResourceManager()

    def auto_connect(self) -> tuple[list[ScpiInstrument], list[str]]:
        self.__auto_disconnect()
        self.__scpi_instrument_list = self.__scan_and_connect_scpi_all()
        for instrument in self.__scpi_instrument_list:
            if self.config.response_mark.digital_multimeter in instrument.response:
                self.services.digital_multimeter = DigitalMultimeterService(instrument, self.__pubtools, self.config)
            elif self.config.response_mark.digital_oscilloscope in instrument.response:
                self.services.digital_oscilloscope = DigitalOscilloscopeService(instrument, self.__pubtools,
                                                                                self.config)
            elif self.config.response_mark.waveform_generator in instrument.response:
                self.services.waveform_generator = WaveformGeneratorService(instrument, self.__pubtools, self.config)
            elif self.config.response_mark.analog_electronic_load in instrument.response:
                self.services.analog_electronic_load = AnalogElectronicLoadService(instrument, self.__pubtools,
                                                                                   self.config)
            else:
                pass
        not_connect_list = self.services.check()
        if len(not_connect_list) == 0:
            self.__pubtools.debug_output("All SCPI instruments connect successfully.")
        else:
            for instrument_name in not_connect_list:
                self.__pubtools.debug_output(f'Error: Cannot connect to instrument "{instrument_name}".')
        return self.__scpi_instrument_list, not_connect_list

    def __scan_and_connect_scpi_all(self) -> list[ScpiInstrument]:
        self.__pubtools.debug_output("Start to connect instruments...")
        device_name_list: tuple[str] = self.__scpi_manager.list_resources()
        scpi_instrument_list: list[ScpiInstrument] = []
        connect_number: int = 0
        if len(device_name_list) != 0:
            for device_name in device_name_list:
                scpi_connection, scpi_response = self.__scpi_connect(device_name)
                if scpi_connection is not None:
                    scpi_instrument: ScpiInstrument = ScpiInstrument()
                    scpi_instrument.connection = scpi_connection
                    scpi_instrument.name = device_name
                    scpi_instrument.response = scpi_response
                    scpi_instrument_list.append(scpi_instrument)
                    connect_number += 1
                if self.__connection_timeout_signal >= 2:
                    self.__connection_timeout_signal = 0
                    break
                if connect_number >= self.__max_connection_number:
                    break
        self.__pubtools.debug_output(f"Connect completed! Connected to [{connect_number}] instruments.")
        return scpi_instrument_list

    def __auto_disconnect(self) -> list[ScpiInstrument]:
        new_scpi_instrument_list: list[ScpiInstrument] = []
        if len(self.__scpi_instrument_list) != 0:
            for scpi_instrument in self.__scpi_instrument_list:
                scpi_instrument.connection.close()
        self.__scpi_instrument_list = new_scpi_instrument_list
        return new_scpi_instrument_list

    def __scpi_connect(self, device_name: str) -> tuple[Optional[scpiConnection], Optional[str]]:
        if (device_name.count("::") < 2):
            self.__connection_timeout_signal += 1
            self.__pubtools.debug_output(f'[InstrumentController] Warning: Maybe connecting a invalid virtual SCPI object "{device_name}". Auto ignore.')
        else:
            for _ in range(0, self.__retry_times):
                try:
                    scpi_connection = typing.cast(scpiConnection, self.__scpi_manager.open_resource(device_name))
                    scpi_connection.timeout = self.config.connection_timeout_ms
                    scpi_response: str = scpi_connection.query('*IDN?')
                    scpi_response = scpi_response.strip()
                    self.__pubtools.debug_output(f'Connect to SCPI device "{device_name}".')
                    self.__pubtools.debug_output(f'Device response: "{scpi_response}"')
                except Exception as e:
                    self.__pubtools.debug_output(f'Error when attempting to connect scpi device "{device_name}".')
                    info = f'Error information: {e}'
                    self.__pubtools.debug_output(info)
                    if ("timeout" in info) or ("Timeout" in info):
                        self.__connection_timeout_signal += 1
                        self.__pubtools.debug_output("[InstrumentController] Warning: One timeout expection when connecting.")
                    self.__pubtools.debug_output(f'Start to retry......')
                else:
                    return scpi_connection, scpi_response
            self.__pubtools.debug_output(f'Retrying failed for {self.__retry_times} times. Give up connecting.')
        return None, None


class FloatServer:
    def __init__(self, name: Optional[str] = None):
        self.name: Optional[str] = name
        self.__app = Flask(__name__)
        CORS(self.__app)
        self.__value = 0.0
        self.__setup_routes()

    def __setup_routes(self):
        if self.name is None:
            self.name = f"/float_default_{time.time()}"
        else:
            self.name = f"/{self.name}"

        @self.__app.route(self.name, methods=['GET'])
        def get_value():
            return jsonify({'value': self.__value})

    def run(self, host, port):
        self.__app.run(host=host, port=port)

    def push(self, new_value):
        self.__value = new_value


class MultimeterValueRange(Enum):
    range_200mv = "0"
    range_2v = "1"
    range_20v = "2"
    range_200v = "3"
    range_1000v = "4"
    default = "3"


class MultimeterValueRangeTools(object):
    @staticmethod
    def get_range_type() -> list[str]:
        type_list: list[str] = ["200mv", "2v", "20v", "200v", "1000v", "default"]
        return type_list

    @staticmethod
    def solve_range(rang_string) -> MultimeterValueRange:
        match rang_string:
            case "200mv":
                return MultimeterValueRange.range_200mv
            case "2v":
                return MultimeterValueRange.range_2v
            case "20v":
                return MultimeterValueRange.range_20v
            case "200v":
                return MultimeterValueRange.range_200v
            case "1000v":
                return MultimeterValueRange.range_1000v
            case "default":
                return MultimeterValueRange.default
            case _:
                return MultimeterValueRange.default


class DigitalMultimeterService(object):
    def __init__(self, scpi_instrument: ScpiInstrument, pubtools: ProgramPublicTools,
                 config: InstrumentControllerConfig):
        self.scpi_instrument: ScpiInstrument = scpi_instrument
        self.__pubtools: ProgramPublicTools = pubtools
        self.__config: InstrumentControllerConfig = config
        self.__range: MultimeterValueRange = MultimeterValueRange.default
        self.realtime_value: float = float(0)
        self.__listening_thread: Optional[threading.Thread] = None
        self.__float_server_thread: Optional[threading.Thread] = None
        self.__whether_range_need_to_set: bool = True
        self.__error_count: int = 0
        self.name: str = "Digital Multimeter"
        self.__instrument_name: str = "multimeter"
        self.__float_server: FloatServer = FloatServer(self.__instrument_name)
        self.__is_listening: bool = False
        self.__server_url: str = ''

    def set_range(self, value_range: MultimeterValueRange):
        self.__range = value_range

    @staticmethod
    def get_range() -> list[str]:
        return MultimeterValueRangeTools.get_range_type()

    @staticmethod
    def solve_range_string(range_string: str) -> MultimeterValueRange:
        return MultimeterValueRangeTools.solve_range(range_string)

    def __float_server_main(self, float_server: FloatServer, server_host, server_port):
        self.__float_server.run(host=server_host, port=server_port)

    def keep_listening(self, realtime_terminal_output: bool = False,
                       server_port: Optional[int] = None, server_host: Optional[str] = "0.0.0.0"):
        if self.__is_listening is False:
            self.__start_server_listening(realtime_terminal_output, server_port, server_host)
            self.__is_listening = True
        return self.__return_server_info()

    def __start_server_listening(self, realtime_terminal_output: bool = False,
                                 server_port: Optional[int] = None, server_host: Optional[str] = "0.0.0.0"):
        if server_port is None:
            server_port = self.__pubtools.find_available_server_port(number=1, address="0.0.0.0", start_port=8600)[0]
        listening_thread_args = (realtime_terminal_output,)
        self.__listening_thread = threading.Thread(target=self.__listening_main, args=listening_thread_args).start()
        float_server_args = (self.__float_server, server_host, server_port)
        self.__float_server_thread = threading.Thread(target=self.__float_server_main, args=float_server_args).start()
        server_root_name = self.__float_server.name
        self.__server_url = f"http://localhost:{server_port}{server_root_name}"
        return self.__return_server_info()

    def __return_server_info(self):
        return self.__listening_thread, self.__float_server_thread, self.__server_url

    def __listening_main(self, realtime_terminal_output: bool):
        threading.Thread(target=self.__listening_auto_clear)
        while True:
            self.realtime_value = self.__get_value()
            self.__float_server.push(self.realtime_value)
            if realtime_terminal_output is True:
                self.__pubtools.debug_output(f"Listening voltage from digital multimeter: {self.realtime_value}.")

    def __listening_auto_clear(self):
        while True:
            self.__whether_range_need_to_set = True
            time.sleep(5)

    def get_value_latest(self) -> float:
        return self.realtime_value

    def __get_value(self, use_range: Optional[MultimeterValueRange] = None):
        if use_range is None:
            use_range = self.__range
        else:
            self.__whether_range_need_to_set = True
        if (self.__whether_range_need_to_set is True) and (self.__error_count <= 3):
            try:
                cmd_set = f":MEASure:VOLTage:DC {use_range.value}"
                self.scpi_instrument.connection.query(cmd_set)
            except Exception as e:
                self.__pubtools.debug_output("Exception found when trying to set range of digital multimeter.")
                self.__pubtools.debug_output(f"Error info: {e}")
                self.__error_count += 1
            else:
                self.__whether_range_need_to_set = False
                self.__error_count = 0
        cmd_query = ":MEASure:VOLTage:DC?"
        multimeter_value = self.scpi_instrument.connection.query(cmd_query)
        return float(multimeter_value)


class VideoStreamer:
    def __init__(self, buffer_size: int = 10, name: Optional[str] = None):
        self.name: Optional[str] = name
        self.__app = Flask(__name__)
        self.__frame_queue = deque(maxlen=buffer_size)
        self.__lock = threading.Lock()  # 用于确保线程安全
        self.__setup_routes()

    def push(self, frame):
        """
        推送一个帧到流。
        """
        with self.__lock:
            self.__frame_queue.append(frame)

    def __generate(self):
        while True:
            if len(self.__frame_queue) == 0:
                time.sleep(0.05)  # 若队列为空，稍微等待
                continue
            with self.__lock:
                frame = self.__frame_queue.popleft()  # 从队列左侧获取帧
            ret, jpeg = cv2.imencode('.jpg', frame)
            if ret:
                yield (b'--frame\r\n'
                       b'Content-Type: image/jpeg\r\n\r\n' + jpeg.tobytes() + b'\r\n\r\n')

    def __video_feed(self):
        return Response(self.__generate(), mimetype='multipart/x-mixed-replace; boundary=frame')

    def __setup_routes(self):
        if self.name is None:
            self.name = f"/video_default_{time.time()}"
        else:
            self.name = f"/{self.name}"
        self.__app.add_url_rule(self.name, 'video_feed', self.__video_feed)

    def run(self, host, port):
        self.__app.run(threaded=True, host=host, port=port)
        return self.name


class DigitalOscilloscopeService(object):
    def __init__(self, scpi_instrument: ScpiInstrument, pubtools: ProgramPublicTools,
                 config: InstrumentControllerConfig):
        self.scpi_instrument: ScpiInstrument = scpi_instrument
        self.__pubtools: ProgramPublicTools = pubtools
        self.__config: InstrumentControllerConfig = config
        self.__instrument_name: str = "oscilloscope"
        self.realtime_screenshot: Optional[mat] = None
        self.__listening_thread: Optional[threading.Thread] = None
        self.__streamer_thread: Optional[threading.Thread] = None
        self.__windows_name: str = "Realtime Screenshot From Digital Oscilloscope"
        self.__video_streamer: VideoStreamer = VideoStreamer(name="oscilloscope")
        self.name: str = "Digital Oscilloscope"
        self.__is_listening: bool = False
        self.__stream_port: Optional[int] = None
        self.__stream_url: Optional[str] = None
        self.__screenshot_frame = None
        self.__img_empty = np.zeros((500, 500, 3), dtype=np.uint8)
        self.__img_empty[:, :] = [255, 255, 255]  # BGR

    def keep_listening(self, realtime_terminal_output: bool = False,
                       stream_port: Optional[int] = None, stream_host: str = "0.0.0.0"):
        if self.__is_listening is False:
            self.__start_listening_server(realtime_terminal_output, stream_port, stream_host)
            self.__is_listening = True
        return self.__return_stream_info()

    def __return_stream_info(self):
        return self.__listening_thread, self.__stream_port, self.__stream_url

    def __start_listening_server(self, realtime_terminal_output: bool = False,
                                 stream_port: Optional[int] = None, stream_host: str = "0.0.0.0"):
        if stream_port is None:
            stream_port = self.__pubtools.find_available_server_port(number=1, address="0.0.0.0", start_port=9000)[0]
        self.__stream_port = stream_port
        listening_thread_args = (realtime_terminal_output, True)
        self.__listening_thread = threading.Thread(target=self.__listening_main, args=listening_thread_args).start()
        streamer_thread_args = (self.__video_streamer, stream_port, stream_host)
        self.__streamer_thread = threading.Thread(target=self.__streamer_main, args=streamer_thread_args).start()
        stream_root_name = self.__video_streamer.name
        self.__stream_url = f"http://localhost:{stream_port}{stream_root_name}"
        return self.__return_stream_info()

    @staticmethod
    def __streamer_main(video_streamer: VideoStreamer,
                        stream_port: int = None, stream_host: str = "0.0.0.0"):
        video_streamer.run(host=stream_host, port=stream_port)

    def save_img(self):
        get_time: str = time.strftime("%H:%M:%S")
        current_file_path = os.path.realpath(__file__)
        current_dir = os.path.dirname(current_file_path)
        parent_dir = os.path.dirname(current_dir)
        file_path = os.path.join(parent_dir, "export", f"OSC.{get_time.replace(':', '.')}.png")
        if self.__screenshot_frame is None:
            cv2.imwrite(file_path, self.__img_empty)
        else:
            cv2.imwrite(file_path, self.__screenshot_frame)
        return get_time

    def __listening_main(self, realtime_terminal_output: bool, enable_saving_file: bool = True):
        video_recode_service = None
        height, width = 512, 512
        screenshot_frame_last = np.zeros((height, width, 3), dtype=np.uint8)
        screenshot_frame_last[:, :] = [255, 255, 255]  # BGR
        if enable_saving_file is True:
            folder_path = os.path.join(self.__config.buffer_path, self.__instrument_name)
            os.makedirs(folder_path, exist_ok=True)
            img_path = os.path.join(folder_path, f"{self.__instrument_name}.avi")
            fourcc = cv2.VideoWriter_fourcc('I', '4', '2', '0')
            video_recode_service = cv2.VideoWriter(img_path, fourcc, 25, (640, 480))
        if realtime_terminal_output is True:
            cv2.namedWindow(self.__windows_name, cv2.WINDOW_NORMAL)
        while True:
            try:
                screenshot_frame = self.get_screenshot()
            except Exception as e:
                self.__pubtools.debug_output(e)
                screenshot_frame = screenshot_frame_last
            else:
                screenshot_frame_last = screenshot_frame
            self.__screenshot_frame = screenshot_frame
            self.__video_streamer.push(screenshot_frame)
            if (enable_saving_file is True) and (video_recode_service is not None):
                video_recode_service.write(screenshot_frame)
            if realtime_terminal_output is True:
                cv2.imshow(self.__windows_name, self.realtime_screenshot)
                cv2.waitKey(1)

    def get_screenshot(self) -> mat:
        self.scpi_instrument.connection.write(':DISP:DATA? JPG')
        raw_data = self.scpi_instrument.connection.read_raw()
        image_data = self.__parse_data(raw_data)
        img_raw = np.asarray(bytearray(image_data), dtype="uint8")
        img_raw = cv2.imdecode(img_raw, cv2.IMREAD_COLOR)
        return img_raw

    @staticmethod
    def __parse_data(raw_data: bytes) -> bytes:
        start_index = raw_data.find(b'#')
        if start_index == -1:
            raise ValueError("Invalid data: TMC header not found.")
        length_digits = int(raw_data[start_index + 1:start_index + 2])
        data_length = int(raw_data[start_index + 2:start_index + 2 + length_digits])
        data_start = start_index + 2 + length_digits
        data_end = data_start + data_length
        image_data = raw_data[data_start:data_end]
        return image_data


class WaveformType(Enum):
    Sine = "SIN"
    Square = "SQU"
    Ramp = "RAMP"
    Pulse = "PULS"
    Noise = "NOIS"


class WaveformTypeTools(object):
    @staticmethod
    def get_all_waveform_types() -> str:
        return " ".join([waveform_type.name for waveform_type in WaveformType])

    @staticmethod
    def match_waveform_type(waveform_name: str) -> WaveformType:
        try:
            return WaveformType[waveform_name.strip().capitalize()]
        except KeyError:
            return WaveformType.Sine


class WaveformFreqUnit(Enum):
    MHz = "MHz"
    kHz = "kHz"
    Hz = "Hz"
    mHz = "mHz"
    uHz = "uHz"


class WaveformLevelUnit(Enum):
    V = "Vpp"
    mV = "mVpp"


class WaveformConfig(object):
    def __init__(self):
        self.type: WaveformType = WaveformType.Sine
        self.type_string: str = "SIN"
        self.freq: float = 100
        self.freq_unit: WaveformFreqUnit = WaveformFreqUnit.kHz
        self.high_level: float = 3
        self.high_level_unit: WaveformLevelUnit = WaveformLevelUnit.V
        self.low_level: float = 3
        self.low_level_unit: WaveformLevelUnit = WaveformLevelUnit.mV

    @staticmethod
    def get_type_string() -> str:
        return WaveformTypeTools.get_all_waveform_types()

    @staticmethod
    def match_type_string(type_string: str) -> WaveformType:
        return WaveformTypeTools.match_waveform_type(type_string)

    def get_freq_unit_type_string(self) -> str:
        return self.__get_type_value_string(WaveformFreqUnit)

    def match_freq_unit_type_string(self, freq_unit_string: str) -> WaveformFreqUnit:
        return self.__match_type_value_string(WaveformFreqUnit, WaveformFreqUnit.kHz, freq_unit_string)

    def get_level_unit_type_string(self) -> str:
        return self.__get_type_value_string(WaveformLevelUnit)

    def match_level_unit_type_string(self, level_unit_string: str) -> WaveformLevelUnit:
        return self.__match_type_value_string(WaveformLevelUnit, WaveformLevelUnit.V, level_unit_string)

    @staticmethod
    def __get_type_value_string(type_enum) -> str:
        type_value_string: str = ""
        for type_enum_item in type_enum:
            type_value_string = type_value_string + " " + type_enum_item.value()
        return type_value_string

    @staticmethod
    def __match_type_value_string(type_enum, type_default, type_value_string: str):
        for item in type_enum:
            if item.value == type_value_string:
                return item
        return type_default

    def to_str(self) -> str:
        config = copy.deepcopy(self)
        parts = [
            config.type.value.lower(),
            str(config.freq),
            config.freq_unit.value,
            str(config.high_level),
            config.high_level_unit.value.lower(),
            str(config.low_level),
            config.low_level_unit.value.lower()
        ]
        if parts[2] == "MHz":
            parts[2] = "LMHz"
        return "|".join(parts)

    def read_str(self, config_str: str):
        parts = config_str.split("|")
        self.type = WaveformType(parts[0].capitalize())
        self.freq = float(parts[1])
        if parts[2].lower() == "lmhz":
            self.freq_unit = WaveformFreqUnit.MHz
        else:
            self.freq_unit = WaveformFreqUnit(parts[2].capitalize())
        self.high_level = float(parts[3])
        self.high_level_unit = WaveformLevelUnit(parts[4].upper())
        self.low_level = float(parts[5])
        self.low_level_unit = WaveformLevelUnit(parts[6].upper())


class WaveformGeneratorService(object):
    def __init__(self, scpi_instrument: ScpiInstrument, pubtools: ProgramPublicTools,
                 config: InstrumentControllerConfig):
        self.scpi_instrument: ScpiInstrument = scpi_instrument
        self.__pubtools: ProgramPublicTools = pubtools
        self.__config: InstrumentControllerConfig = config
        self.enable_channel_1: bool = True
        self.enable_channel_2: bool = True
        self.config_channel_1: WaveformConfig = WaveformConfig()
        self.config_channel_2: WaveformConfig = WaveformConfig()
        self.name: str = "Waveform Generator"

    def output_restart(self):
        self.output_pause()
        self.output_start()

    def output_start(self):
        if self.enable_channel_1:
            self.__output_start_channel(1)
        if self.enable_channel_2:
            self.__output_start_channel(2)

    def __output_start_channel(self, channel_number: int):
        cmd = f":OUTPut{channel_number} ON"
        self.scpi_instrument.connection.write(cmd)

    def output_pause(self):
        self.__output_pause_channel(1)
        self.__output_pause_channel(2)

    def __output_pause_channel(self, channel_number: int):
        cmd = f":OUTPut{channel_number} OFF"
        self.scpi_instrument.connection.write(cmd)

    def get_config(self, channel: str) -> tuple[int, WaveformConfig]:
        if channel == "2":
            return 2, self.config_channel_2
        else:
            return 1, self.config_channel_1

    def set_enable(self, channel: str, enable: bool) -> int:
        if channel == "2":
            self.enable_channel_2 = enable
            return 2
        else:
            self.enable_channel_1 = enable
            return 1

    def apply_config(self, immediate_start: bool = False):
        self.__apply_config_channel(1, self.enable_channel_1, self.config_channel_1, immediate_start)
        self.__apply_config_channel(2, self.enable_channel_2, self.config_channel_2, immediate_start)

    def __apply_config_channel(self, channel_number: int, channel_enable: bool, channel_config: WaveformConfig,
                               immediate_start: bool = False):
        if channel_config.type_string != WaveformType.Noise.value:
            cmd = (f":SOURce{channel_number}:APPLy:{channel_config.type_string} " +
                   f"{channel_config.freq}{channel_config.freq_unit.value}," +
                   f"{channel_config.high_level}{channel_config.high_level_unit.value}")
        else:
            cmd = (f":SOURce{channel_number}:APPLy:{channel_config.type_string} " +
                   f"{channel_config.high_level}{channel_config.high_level_unit.value}")
        self.scpi_instrument.connection.write(cmd)
        if (immediate_start is False) or (channel_enable is False):
            self.output_pause()
        else:
            self.output_start()


class AnalogElectronicLoadService(object):
    def __init__(self, scpi_instrument: ScpiInstrument, pubtools: ProgramPublicTools,
                 config: InstrumentControllerConfig):
        self.scpi_instrument: ScpiInstrument = scpi_instrument
        self.__pubtools: ProgramPublicTools = pubtools
        self.__config: InstrumentControllerConfig = config
        self.name: str = "Analog Electronic Load"
        self.__mode_set_time: int = 0

    def set_mode(self):
        if self.__mode_set_time <= 2:
            self.__mode_set_time += 1
            cmd_set = f":SOUR:FUNC RES"
            self.scpi_instrument.connection.write(cmd_set)

    def set_resistance(self, value: str):
        self.set_mode()
        cmd_set = f":SOUR:RES:LEV:IMM {value}"
        self.scpi_instrument.connection.write(cmd_set)
        self.__set_open()

    def __set_open(self):
        self.scpi_instrument.connection.write(":SOUR:INP:STAT 1")

    def __set_close(self):
        self.scpi_instrument.connection.write(":SOUR:INP:STAT 0")


def run_as_main():
    program_pubtools: ProgramPublicTools = ProgramPublicTools()
    instrument_controller: InstrumentController = InstrumentController(program_pubtools)
    instrument_controller.auto_connect()
    # _, _, server_url_multimeter = instrument_controller.services.digital_multimeter.keep_listening()
    # print("digital_multimeter", server_url_multimeter)
    # _, _, server_url_oscilloscope = instrument_controller.services.digital_oscilloscope.keep_listening()
    # print("digital_oscilloscope", server_url_oscilloscope)
    while True:
        input_value = input("input: ")
        instrument_controller.services.analog_electronic_load.set_resistance(input_value)


if __name__ == "__main__":
    run_as_main()