Communicate with a SICK FlexiCompact ove Modbus TCP

flexicompact.py

#!/usr/bin/env python3
#-*- coding: utf-8 -*-

from pymodbus.client.sync import ModbusTcpClient as ModbusClient
from pymodbus.constants import Endian
from pymodbus.payload import BinaryPayloadBuilder, BinaryPayloadDecoder


class FlexiCompact:
    def __init__(self, ip, port=502, unit=1):
        self.unit = unit
        self.client = ModbusClient(ip, port=port)

    @staticmethod
    def to_string(r):
        s = ""
        for x in r:
            s += chr(x >> 8)
            s += chr(x & 0xFF)
        return s

    def read_device(self):
        # https://cdn.sick.com/media/docs/5/15/415/operating_instructions_flexi_compact_safety_controller_de_im0093415.pdf
        # Seite 144
        print("\n== Geräteidentifikation ==")
        count = 0
        device = {
            "Herstellername": 4,
            "Produktcode": 4,
            "MajorMinorRevision": 6,
            "Hersteller-URL": 6,
            "Produktname": 16,
            "Modelname": 9,
            "Seriennummer": 4,
            "Applikationsname": 16,
            "Modbus TCP-Profil‐version": 6,
        }
        rr = self.client.read_input_registers(0, 71, unit=self.unit)
        for d in device:
            print(f"{d:<30}: {self.to_string(rr.registers[count:count+device[d]])}")
            count += device[d]

    @staticmethod
    def get_bits(b):
        res = [
            (b & 0x80) >> 7,
            (b & 0x40) >> 6,
            (b & 0x20) >> 5,
            (b & 0x10) >> 4,
            (b & 0x08) >> 3,
            (b & 0x04) >> 2,
            (b & 0x02) >> 1,
            (b & 0x01)
        ]
        return res

    def read_result(self):
        # https://cdn.sick.com/media/docs/5/15/415/operating_instructions_flexi_compact_safety_controller_de_im0093415.pdf
        # Seite 145
        print("\n== Eingangsdaten ==")
        rr = self.client.read_input_registers(256, 25, unit=self.unit)
        counter = 0
        for res in rr.registers:
            high, low = res>>8, res & 0xFF
            print(f"{counter:<4} "\
                  f"{(high & 0x80)>>7} "\
                  f"{(high & 0x40)>>6} "\
                  f"{(high & 0x20)>>5} "\
                  f"{(high & 0x10)>>4} "\
                  f"{(high & 0x08)>>3} "\
                  f"{(high & 0x04)>>2} "\
                  f"{(high & 0x02)>>1} "\
                  f"{(high & 0x01)}"
                )
            print(f"{counter+1:<4} "\
                  f"{(low & 0x80)>>7} "\
                  f"{(low & 0x40)>>6} "\
                  f"{(low & 0x20)>>5} "\
                  f"{(low & 0x10)>>4} "\
                  f"{(low & 0x08)>>3} "\
                  f"{(low & 0x04)>>2} "\
                  f"{(low & 0x02)>>1} "\
                  f"{(low & 0x01)}"
                )
            counter += 2


    def read_status(self):
        # https://cdn.sick.com/media/docs/5/15/415/operating_instructions_flexi_compact_safety_controller_de_im0093415.pdf
        # Seite 145
        print("\n== Prüfsummen / Statusinformationen / Metadaten ==")
        rr = self.client.read_input_registers(2000, 12, unit=self.unit)
        status = [
            "Unknown State",
            "Start Up",
            "Service Mode",
            "Normal Operation",
            "Suspended Operation",
            "unknown",
            "Service required",
            "Recoverable Error",
            "Fatal Error",
        ]
        print(f"{'Gerätestatus':<30}: {status[rr.registers[0]]}")
        user_interaction = {
            0x00: "Keine Interaktion erforderlich",
            0x01: "Konfiguration bestätigen",
            0x02: "Konfiguration prüfen",
            0x08: "Applikationsschnittstelle prüfen",
            0x10: "Gerät prüfen",
            0x20: "Setup Vorgang ausführen",
            0x40: "Firmware prüfen",
            0x80: "Warten",
        }
        print(
            f"{'Benutzerinteraktion':<30}: {user_interaction.get(rr.registers[1],'?')}"
        )
        print(f"{'Gesamtprüfsumme':<30}: 0x{rr.registers[2]:0x}{rr.registers[3]:0x}")
        print(
            f"{'Sicherheitsprüfsumme':<30}: 0x{rr.registers[4]:0x}{rr.registers[5]:0x}"
        )
        print(f"{'Standardprüfsumme':<30}: 0x{rr.registers[6]:0x}{rr.registers[7]:0x}")
        print(
            f"{'Verifizierungsprüfsumme':<30}: 0x{rr.registers[10]:0x}{rr.registers[11]:0x}"
        )

        print("\n== Hardware ==")
        devices = {
            0x3001: "Hauptmodul CPUc1",
            0x3101: "Hauptmodul CPUc2",
            0x0005: "Erweiterungsmodul Gateway EtherCAT",
            0x0006: "Erweiterungsmodul Gateway PROFINET",
            0x300B: "Erweiterungsmodul IO - XTDI",
            0x300C: "Erweiterungsmodul IO - XTDO",
            0x30FF: "SmartPlug"
        }
        rr = self.client.read_input_registers(2020, 36, unit=self.unit)
        modules = rr.registers[::2]
        for n, i in enumerate(modules):
            if i > 0:
                print(f"{n:<4} {devices.get(i, '?')}")
        
        print("\n== Modulstatus ==")
        #
        # 2021-02-22: Firmware Bug, 2300 liefert 0xFFFF für alle 80 Register
        #
        rr = self.client.read_input_registers(2300, 80, unit=self.unit)
        for n, r in enumerate(rr.registers):
            print(f"{n:<3} {hex(r)}")

        rr = self.client.read_input_registers(3000, 32, unit=self.unit)
        print(
                f"{'Gerätename':<30}: {self.to_string(rr.registers)}"
            )
        rr = self.client.read_input_registers(3032, 32, unit=self.unit)
        print(
                f"{'Projektname':<30}: {self.to_string(rr.registers)}"
            )
        
        rr = self.client.read_input_registers(3064, 32, unit=self.unit)
        print(
                f"{'Applikationsname':<30}: {self.to_string(rr.registers)}"
            )
        rr = self.client.read_input_registers(3096, 32, unit=self.unit)
        print(
                f"{'Benutzername':<30}: {self.to_string(rr.registers)}"
            )
        rr = self.client.read_input_registers(3128, 50, unit=self.unit)
        print(
                f"{'Beschreibung':<30}: {self.to_string(rr.registers)}"
            )

    def set_all_bits(self):
        print("\n== Setze alle Bits zur CPU ==")
        rr = self.client.write_registers(256, [0xFFFF]*25, unit=self.unit)
        print(rr)

    def clear_all_bits(self):
        print("\n== Lösche alle Bits zur CPU  ==")
        rr = self.client.write_registers(256, [0x0000]*25, unit=self.unit)
        print(rr)

    def connect(self):
        self.client.connect()

    def disconnect(self):
        self.client.close()


if __name__ == "__main__":
    fc = FlexiCompact("192.168.255.3")
    fc.connect()
    #fc.read_device()
    #fc.read_result()
    #fc.read_status()
    fc.set_all_bits()
    #fc.clear_all_bits()
    fc.disconnect()