modbus_test/client.py

print("modbus test")

# Настройки клиента Modbus
MODBUS_SERVER_HOST = "192.168.70.55"  # IP-адрес Modbus-сервера
MODBUS_SERVER_PORT = 502
MODBUS_SLAVE_ID = 11

indent = 21100
s = 800
k = 2

from pymodbus.client import ModbusTcpClient
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadDecoder
import time
import struct


def two_byte_convert(result):
    # print(result)
    v0 = result.registers[0]
    value = (v0 << 16) + result.registers[1]
    if result.registers[0] > (65535 * 0.5):
        value = value - (1 << 32)
    return value


def bulb(addr):
    try:
        coils = client.read_coils(addr, 1, MODBUS_SLAVE_ID)
        print("читаем начальное состояние", coils.bits)

        new_bits = coils.bits
        new_bits[0] = not new_bits[0]
        print("пытаемся записать", new_bits, coils.slave_id)
        result = client.write_coil(addr, new_bits[0], coils.slave_id)
        print(result)

        print(
            "читаем после попытки",
            client.read_coils(0, len(coils.bits), coils.slave_id).bits,
        )
    except Exception as e:
        raise e


def get_coordinates():
    try:
        data = [
            ("j1", 2268),
            ("j2", 2270),
            ("j3", 2272),
            ("j4", 2274),
            ("j5", 2276),
            ("j6", 2278),
            ("x", 2332),
            ("y", 2334),
            ("z", 2336),
            ("u", 2338),
            ("v", 2340),
            ("w", 2342),
        ]
        i = 2
        for d in data:
            result = client.read_holding_registers(d[1], 2, MODBUS_SLAVE_ID)
            value = two_byte_convert(result)
            adjusted_value = value / 1000.0
            i *= 2
            print(d[0], result.registers, adjusted_value)
    except Exception as e:
        raise e


def collect_coordinates():
    # собираем предварительные данные
    x = two_byte_convert(client.read_holding_registers(2332, 2, MODBUS_SLAVE_ID))
    y = two_byte_convert(client.read_holding_registers(2334, 2, MODBUS_SLAVE_ID))
    z = two_byte_convert(client.read_holding_registers(2336, 2, MODBUS_SLAVE_ID))
    u = two_byte_convert(client.read_holding_registers(2338, 2, MODBUS_SLAVE_ID))
    v = two_byte_convert(client.read_holding_registers(2340, 2, MODBUS_SLAVE_ID))
    w = two_byte_convert(client.read_holding_registers(2342, 2, MODBUS_SLAVE_ID))
    print(
        f"Предварительные данные X {x * 0.001}, Y {y * 0.001}, Z {z*0.001}, U {u*0.001}, V {v*0.001}, W {w*0.001}"
    )
    return [x, y, z, u, v, w]


def get_or_set_speed(number=None):
    result = client.read_holding_registers(20200, 1, MODBUS_SLAVE_ID)
    print("сейчас скорость", result.registers[0] * 0.1)
    if number and number > 0 and number <= 1000:
        client.write_register(20200, number * 10, MODBUS_SLAVE_ID)
        result = client.read_holding_registers(20200, 1, MODBUS_SLAVE_ID)
        print("установлена скорость", result.registers[0] * 0.1)


def set_user_register(n, values):
    client.write_registers(indent + s + k * n, values, MODBUS_SLAVE_ID)


def to_double(integr):
    return [(integr >> 16) & 0xFFFF, integr & 0xFFFF]


def set_user_reg_800(data):
    for item in data:
        d, values = item
        set_user_register(d, values)


def start_in_auto():
    # старт в авторежиме single loop
    client.write_register(20002, 1, MODBUS_SLAVE_ID)


client = ModbusTcpClient(
    host=MODBUS_SERVER_HOST,
    port=MODBUS_SERVER_PORT,
)
client.connect()
# максимальное количество coils = 286

# bulb(0)
# get_coordinates()
# get_or_set_speed(10)
try:
    total = 0
    state = None
    step = 0
    counter = 0
    type = "rel"
    paths = [
        ("line", 100, 100, 0),
        ("line", 100, -100, 0),
        ("line", -100, -100, 0),
        ("line", -100, 100, 0),
        ("line", 0, -50, 0),
    ]

    # ставим пользовательские переменные в ноль
    set_user_reg_800(
        [
            (0, to_double(0)),
            (1, to_double(0)),
            (2, to_double(0)),
            (3, [0, 0]),
            (4, [0, 0]),
            (5, [0, 0]),
        ]
    )

    # старт в авторежиме single loop
    start_in_auto()

    while True:
        time.sleep(1)
        total += 1
        green_light = client.read_coils(0, 1, MODBUS_SLAVE_ID).bits[0]
        print(f"total {total}, state {state}, green light {green_light}")

        if state == green_light:
            continue

        state = green_light

        if state == True:
            continue

        if type == "rel":
            # работаем с REL относительными координатами
            x = 0
            y = 0
            z = 0
        else:
            # работаем с абсолютными координатами
            x, y, z, u, v, w = collect_coordinates()

        if step >= len(paths):
            counter += 1
            step = 0
        
        current_step = paths[step]
        line_type, *coord = current_step

        if line_type == "line":
            x, y, z = coord
            set_user_reg_800(
                [
                    (0, to_double(x)),
                    (1, to_double(y)),
                    (2, to_double(z)),
                ]
            )
            bulb(4)
        
            
        bulb(0)
        step += 1



except Exception as e:
    print("error", e)

client.close()