import re
import struct
class S7String:
def __init__(self, s: bytes | str, size: int = 254, is_wstring: bool = False):
"""
Kapselt Siemens S7 STRINGS und WSTRINGS.
- STRING (is_wstring=False): Max 254 Zeichen. Header: 2 Bytes (Max, Aktuell).
- WSTRING (is_wstring=True): Max 16382 Zeichen. Header: 4 Bytes (Max, Aktuell als UInt16).
"""
self.is_wstring = is_wstring
self.char_size = 2 if is_wstring else 1
self._encoding = "utf-16-be" if is_wstring else "ascii"
max_limit = 16382 if is_wstring else 254
if not 0 <= size <= max_limit:
raise ValueError(f"Größe für {'W' if is_wstring else ''}STRING muss zwischen 0 und {max_limit} liegen.")
match s:
case str():
encoded = s.encode(self._encoding)
if len(encoded) // self.char_size > size:
raise ValueError(f"String zu lang für die deklarierte Kapazität von {size} Zeichen.")
header_size = 4 if is_wstring else 2
self.buffer = bytearray(header_size + (size * self.char_size))
self.size = size
# Header schreiben
if is_wstring:
self.buffer[0:4] = struct.pack(">HH", size, len(encoded) // 2)
else:
self.buffer[0:2] = size, len(encoded)
# Inhalt schreiben
self.buffer[header_size : header_size + len(encoded)] = encoded
case bytes():
header_size = 4 if is_wstring else 2
if len(s) < header_size:
raise ValueError("Bytes müssen mindestens die Header-Bytes enthalten.")
self.buffer = bytearray(s)
if is_wstring:
self.size = struct.unpack(">H", self.buffer[0:2])[0]
else:
self.size = self.buffer[0]
def __len__(self):
"""Gibt die maximal deklarierte Kapazität zurück."""
return self.size
@property
def slen(self):
"""Gibt die aktuelle Textlänge in Zeichen zurück."""
if self.is_wstring:
return struct.unpack(">H", self.buffer[2:4])[0]
return self.buffer[1]
def set(self, s: str):
"""Überschreibt den String-Inhalt sicher."""
encoded = s.encode(self._encoding)
if len(encoded) // self.char_size > self.size:
raise ValueError(f"String ist zu lang für die Puffergröße von {self.size}")
self.clear()
header_size = 4 if self.is_wstring else 2
if self.is_wstring:
self.buffer[2:4] = struct.pack(">H", len(encoded) // 2)
else:
self.buffer[1] = len(encoded)
self.buffer[header_size : header_size + len(encoded)] = encoded
def clear(self):
"""Setzt die aktuelle Länge auf 0 und nullt den Datenbereich."""
header_size = 4 if self.is_wstring else 2
if self.is_wstring:
self.buffer[2:4] = b"\x00\x00"
else:
self.buffer[1] = 0
self.buffer[header_size:] = b"\x00" * (self.size * self.char_size)
def __str__(self):
"""Konvertiert in einen nativen Python-String."""
header_size = 4 if self.is_wstring else 2
actual_bytes = self.slen * self.char_size
return self.buffer[header_size : header_size + actual_bytes].decode(self._encoding)
def __bytes__(self):
"""Gibt das vollständige S7-Byte-Array zurück."""
return bytes(self.buffer)
@classmethod
def from_bytes(cls, raw_bytes: bytes, is_wstring: bool = False):
"""Factory-Methode: Erzeugt ein S7String-Objekt aus den von struct.unpack kommenden Rohdaten."""
return cls(raw_bytes, is_wstring=is_wstring)
def compile_s7_format(fmt: str) -> str:
"""
Kompiliert S7-Formate inklusive Siemens-Strings (S=STRING, W=WSTRING)
in standardkonforme Python struct-Formatstrings unter Beachtung des S7 Natural Alignments.
"""
s7_fmt = ">"
current_size = 0
for match in re.finditer(r"(\d*)([a-zA-Z])", fmt):
count_str, dtype = match.groups()
count = int(count_str) if count_str else 1
# 1. Füllbytes (x) direkt anhängen
if dtype == "x":
s7_fmt += f"{count}x"
current_size += count
continue
# 2. Siemens STRINGS (S) und WSTRINGS (W)
if dtype in ("S", "W"):
is_wstring = (dtype == "W")
char_size = 2 if is_wstring else 1
header_size = 4 if is_wstring else 2
total_bytes = header_size + (count * char_size)
ssize = 2 if is_wstring else 1 # WSTRING richtet sich Wortweise (2 Bytes) aus
if (remainder := current_size % ssize) != 0:
padding = ssize - remainder
s7_fmt += f"{padding}x"
current_size += padding
s7_fmt += f"{total_bytes}s"
current_size += total_bytes
continue
# 3. Standard-Datentypen
ssize = struct.calcsize(f">{dtype}")
if (remainder := current_size % ssize) != 0:
padding_bytes = ssize - remainder
s7_fmt += f"{padding_bytes}x"
current_size += padding_bytes
s7_fmt += f"{count_str}{dtype}"
current_size += ssize * count
return s7_fmt
# S7-Strukturdefinition (z.B. aus einem globalen DB):
# 1 Byte (B), 1 normaler STRING mit max 10 Zeichen (10S), 1 Real/Float (f), 1 WSTRING mit max 5 Zeichen (5W)
s7_schema = "b 10S f 5W"
# 1. Format-String einmalig für das Projekt kompilieren
python_fmt = compile_s7_format(s7_schema)
print(f"Kompiliertes Python-Format: {python_fmt}\n")
# Erwartete Ausgabe enthält automatisch berechnete Füllbytes (x), um das Alignment zu halten
# ==============================================================================
# LEGENDE 1: DATEN PACKEN (Senden zur SPS)
# ==============================================================================
my_string = S7String("Siemens", size=10, is_wstring=False)
my_wstring = S7String("S7-15", size=5, is_wstring=True)
# Datentupel vorbereiten. Die S7String-Objekte müssen per bytes() übergeben werden.
werte_zum_senden = (42, bytes(my_string), 123.45, bytes(my_wstring))
packed_bytes = struct.pack(python_fmt, *werte_zum_senden)
print("SPS-Sende-Payload (Hex):", packed_bytes.hex())
# ==============================================================================
# LEGENDE 2: DATEN ENTPACKEN (Empfangen von SPS via struct.unpack)
# ==============================================================================
# Wir simulieren den Empfang der 'packed_bytes' und jagen sie durch unpack
unpacked_raw = struct.unpack(python_fmt, packed_bytes)
# Die entpackten Felder den Python-Variablen zuordnen
s7_byte_val = unpacked_raw[0]
s7_string_obj = S7String.from_bytes(unpacked_raw[1], is_wstring=False)
s7_float_val = unpacked_raw[2]
s7_wstring_obj = S7String.from_bytes(unpacked_raw[3], is_wstring=True)
print("\n--- Ausgelesene SPS-Daten ---")
print(f"Byte-Wert: {s7_byte_val}")
print(f"STRING (Max={len(s7_string_obj)}, Aktuell={s7_string_obj.slen}): '{s7_string_obj}'")
print(f"Float-Wert: {s7_float_val:.2f}")
print(f"WSTRING (Max={len(s7_wstring_obj)}, Aktuell={s7_wstring_obj.slen}): '{s7_wstring_obj}'")
raise SystemExit("2. Teil nur mit sps")
import struct
import time
import snap7
from snap7.util import *
# --- [Hier die S7String-Klasse und compile_s7_format einfügen] ---
# (Aus der vorherigen Nachricht übernehmen)
# 1. Konfiguration der SPS-Verbindung und Datenstruktur
PLC_IP = "192.168.0.1"
PLC_RACK = 0
PLC_SLOT = 1 # Bei S7-1200 meist 0, bei S7-1500 meist 1
DB_NUMBER = 10 # Die Nummer des Datenbausteins (DB10)
# Unsere bekannte SPS-Struktur: 1 Byte, 1 String (max 12), 1 Füllbyte, 1 Real (Float)
S7_STRUCTURE = "B 12S x f"
python_format = compile_s7_format(S7_STRUCTURE)
BUFFER_SIZE = struct.calcsize(python_format)
print(f"Kompiliertes Format: {python_format}")
print(f"Erwartete Bytegröße aus DB: {BUFFER_SIZE} Bytes\n")
# 2. Verbindung zur SPS aufbauen
client = snap7.client.Client()
try:
client.connect(PLC_IP, PLC_RACK, PLC_SLOT)
print(f"Erfolgreich mit S7-SPS ({PLC_IP}) verbunden.")
# ==========================================
# TEIL 1: DATEN AUS DEM DB LESEN
# ==========================================
print("\n--- Lese Daten aus DB ---")
# Liest den gesamten DB-Bereich ab Offset 0 in ein Bytearray
raw_bytes = client.db_read(DB_NUMBER, 0, BUFFER_SIZE)
# Rohe Bytes entpacken
unpacked_list = struct.unpack(python_format, raw_bytes)
data_iter = iter(unpacked_list)
# Den Werten Variablen zuweisen (Reihenfolge muss exakt stimmen!)
res_byte = next(data_iter)
res_string = S7String.from_struct_unpack(
max_len=next(data_iter),
act_len=next(data_iter),
payload_bytes=next(data_iter)
)
res_float = next(data_iter)
print(f"Aktueller Wert Byte: {res_byte}")
print(f"Aktueller Wert String: '{res_string}'")
print(f"Aktueller Wert Float: {res_float}")
# ==========================================
# TEIL 2: DATEN MODIFIZIEREN UND SCHREIBEN
# ==========================================
print("\n--- Schreibe neue Daten in DB ---")
# Neue Werte definieren
neues_byte = 105
res_string.set("SPS laeuft!") # Nutzt die sichere .set() Methode der Klasse
neuer_float = 45.67
# Payload für struct.pack vorbereiten
payload = (neues_byte, *res_string.to_struct_args(), neuer_float)
# Daten packen (erzeugt das perfekt ausgerichtete Big-Endian Byte-Paket)
send_bytes = struct.pack(python_format, *payload)
# Bytes zurück in die SPS schreiben (DB_NUMBER, Start-Offset, Daten)
client.db_write(DB_NUMBER, 0, send_bytes)
print("Neue Daten erfolgreich in die SPS übertragen.")
except Exception as e:
print(f"Fehler bei der SPS-Kommunikation: {e}")
finally:
# Verbindung sauber trennen
if client.get_connected():
client.disconnect()
print("\nVerbindung zur SPS getrennt.")
Comments