Huanyang sterowanie przez RS-485

import serial
import time
import HuanyangDev

dev = HuanyangDev.HuanyangDev({"port": "COM5", "rate": 19200, "parity": serial.PARITY_NONE, "address": 1, "timeout": 0.1})
dev.open()

dev.write_function_data(8, 2200)
pd008 = dev.read_function_data(8)
print("PD008 = {}".format(pd008))

dev.write_freq(200)
dev.write_control_data(0x03)  # start

# 0: target frequency,  1: output frequency, 2:output current, 3: rpm, 4: DC voltage, 5: AC voltage,6:cont, 7:temp
f1, f2 = dev.read_control_data(0), dev.read_control_data(1)
print("target frequency = {}   output frequency = {}".format(f1, f2))
time.sleep(5)

dev.write_control_data(0x08)  # stop
dev.close()

import crcmod
import serial
import struct

_crc16 = crcmod.predefined.mkPredefinedCrcFun('modbus')


class HuanyangDev:
    def __init__(self, config):
        self.config = config
        self.conn = None

    def open(self):
        self.conn = serial.Serial(
            port=self.config["port"],
            baudrate=self.config["rate"],
            parity=self.config["parity"],
            timeout=self.config["timeout"]
        )

    def close(self):
        self.conn.close()

    def _build_packet(self, function, data):
        packet = [self.config["address"], function, len(data)]
        packet.extend(data)
        crc = self._hy_crc(packet)
        packet.extend(crc)
        return packet

    @staticmethod
    def _hy_crc(message):
        return list(struct.pack('<H', _crc16(bytes(message))))

    def _check_crc(self, message):
        if len(message) > 5:
            calc_crc = self._hy_crc(message[:-2])
            msg_crc = list(message[-2:])
            return calc_crc == msg_crc
        return False

    def _check_msg(self, message):
        if self._check_crc(message):
            if int(message[1]) & 0xF0 > 0:
                return False
            return int(message[2]) == len(message[3:-2])
        return False

    def read_function_data(self, parameter):
        packet = self._build_packet(0x01, [parameter])
        self.conn.write(bytes(packet))
        ans = self.conn.read(8)
        if not self._check_msg(ans):
            return None

        data = ans[3:-2]
        value = int.from_bytes(bytes(data[1:]), byteorder='big')
        return value

    def write_function_data(self, parameter, value):
        value = int(value)
        value_length = max(1, (value.bit_length() + 7) // 8)
        pdata = [parameter]
        pdata.extend(list(value.to_bytes(value_length, 'big')))
        packet = self._build_packet(0x02, pdata)
        self.conn.write(bytes(packet))
        ans = self.conn.read(8)

        if self._check_msg(ans):
            rdata = ans[3:-2]
            value = int.from_bytes(bytes(rdata[1:]), byteorder='big')
            return value
        return None

    def write_control_data(self, data):
        packet = self._build_packet(0x03, [data])
        self.conn.write(bytes(packet))
        ans = self.conn.read(6)
        if self._check_msg(ans):
            value = int.from_bytes(bytes(ans[3:-2]), byteorder='big')
            return value
        return None

    def read_control_data(self, parameter):
        packet = self._build_packet(0x04, [parameter])
        self.conn.write(bytes(packet))
        ans = self.conn.read(8)
        if not self._check_msg(ans):
            return None

        rdata = ans[3:-2]
        value = int.from_bytes(bytes(rdata[1:]), byteorder='big')
        return value

    def write_freq(self, freq):
        pdata = list(int(freq * 100).to_bytes(2, 'big'))
        packet = self._build_packet(0x05, pdata)
        self.conn.write(bytes(packet))
        ans = self.conn.read(8)
        if self._check_msg(ans):
            value = int.from_bytes(bytes(ans[3:-2]), byteorder='big')
            ret = value / 100
            return ret
        return None

from hycontrol import VFDConf, RegisterMap, VFDDevice
import os

conf = VFDConf(port="COM5", rate=19200, parity="N", address=1, timeout=0.1)

regmap = RegisterMap(
    os.path.join(os.path.dirname(os.path.realpath(__file__)), 'hycontrol', 'registers.yml'))

d = VFDDevice(conf, regmap)

d.connect()

data_to_read = []
data_to_read.extend(range(1, 12))
data_to_read.extend(range(70, 75))

data = {}
for a in data_to_read:
    data[a] = d.read_function_data(a)


for k in data:
    print("PD{:03d} = {}".format(k, data[k]))


d.close()

from hycontrol import VFDConf, RegisterMap, VFDDevice
import os
import _thread

conf = VFDConf(port="COM5", rate=19200, parity="N", address=1, timeout=0.1)

regmap = RegisterMap(
    os.path.join(os.path.dirname(os.path.realpath(__file__)), 'hycontrol', 'registers.yml'))

d = VFDDevice(conf, regmap)
d.connect()

commands = {'status': 0x00, 'run_fwd': 0x03, 'run_rev': 0x11, 'stop': 0x08 }

def get_status(dev, cmd):
    return dev.read_control_data(cmd)

def report_thread(dev):
    import time

    while 1:
        time.sleep(0.5)
        rpm = get_status(dev, 3)
        ocur = get_status(dev, 2)
        print("speed = {} rpm      output current = {:2.2f} A".format(rpm[0], ocur[0]))


def run_spindle(dev):
    dev.write_freq(200)
    dev.write_control_data(commands['run_fwd'])

    _thread.start_new_thread(report_thread, (dev,))
    input('Press ENTER to exit\n')

try:
    run_spindle(d)
finally:
    d.write_control_data(commands['stop'])
    d.close()