compile_s7_format2
  1. import re
  2. import struct
  3.  
  4. class S7String:
  5. def __init__(self, s: bytes | str, size: int = 254, is_wstring: bool = False):
  6. """
  7. Kapselt Siemens S7 STRINGS und WSTRINGS.
  8. - STRING (is_wstring=False): Max 254 Zeichen. Header: 2 Bytes (Max, Aktuell).
  9. - WSTRING (is_wstring=True): Max 16382 Zeichen. Header: 4 Bytes (Max, Aktuell als UInt16).
  10. """
  11. self.is_wstring = is_wstring
  12. self.char_size = 2 if is_wstring else 1
  13. self._encoding = "utf-16-be" if is_wstring else "ascii"
  14.  
  15. max_limit = 16382 if is_wstring else 254
  16. if not 0 <= size <= max_limit:
  17. raise ValueError(f"Größe für {'W' if is_wstring else ''}STRING muss zwischen 0 und {max_limit} liegen.")
  18.  
  19. match s:
  20. case str():
  21. encoded = s.encode(self._encoding)
  22. if len(encoded) // self.char_size > size:
  23. raise ValueError(f"String zu lang für die deklarierte Kapazität von {size} Zeichen.")
  24.  
  25. header_size = 4 if is_wstring else 2
  26. self.buffer = bytearray(header_size + (size * self.char_size))
  27. self.size = size
  28.  
  29. # Header schreiben
  30. if is_wstring:
  31. self.buffer[0:4] = struct.pack(">HH", size, len(encoded) // 2)
  32. else:
  33. self.buffer[0:2] = size, len(encoded)
  34.  
  35. # Inhalt schreiben
  36. self.buffer[header_size : header_size + len(encoded)] = encoded
  37.  
  38. case bytes():
  39. header_size = 4 if is_wstring else 2
  40. if len(s) < header_size:
  41. raise ValueError("Bytes müssen mindestens die Header-Bytes enthalten.")
  42. self.buffer = bytearray(s)
  43. if is_wstring:
  44. self.size = struct.unpack(">H", self.buffer[0:2])[0]
  45. else:
  46. self.size = self.buffer[0]
  47.  
  48. def __len__(self):
  49. """Gibt die maximal deklarierte Kapazität zurück."""
  50. return self.size
  51.  
  52. @property
  53. def slen(self):
  54. """Gibt die aktuelle Textlänge in Zeichen zurück."""
  55. if self.is_wstring:
  56. return struct.unpack(">H", self.buffer[2:4])[0]
  57. return self.buffer[1]
  58.  
  59. def set(self, s: str):
  60. """Überschreibt den String-Inhalt sicher."""
  61. encoded = s.encode(self._encoding)
  62. if len(encoded) // self.char_size > self.size:
  63. raise ValueError(f"String ist zu lang für die Puffergröße von {self.size}")
  64.  
  65. self.clear()
  66. header_size = 4 if self.is_wstring else 2
  67.  
  68. if self.is_wstring:
  69. self.buffer[2:4] = struct.pack(">H", len(encoded) // 2)
  70. else:
  71. self.buffer[1] = len(encoded)
  72.  
  73. self.buffer[header_size : header_size + len(encoded)] = encoded
  74.  
  75. def clear(self):
  76. """Setzt die aktuelle Länge auf 0 und nullt den Datenbereich."""
  77. header_size = 4 if self.is_wstring else 2
  78. if self.is_wstring:
  79. self.buffer[2:4] = b"\x00\x00"
  80. else:
  81. self.buffer[1] = 0
  82. self.buffer[header_size:] = b"\x00" * (self.size * self.char_size)
  83.  
  84. def __str__(self):
  85. """Konvertiert in einen nativen Python-String."""
  86. header_size = 4 if self.is_wstring else 2
  87. actual_bytes = self.slen * self.char_size
  88. return self.buffer[header_size : header_size + actual_bytes].decode(self._encoding)
  89.  
  90. def __bytes__(self):
  91. """Gibt das vollständige S7-Byte-Array zurück."""
  92. return bytes(self.buffer)
  93.  
  94. @classmethod
  95. def from_bytes(cls, raw_bytes: bytes, is_wstring: bool = False):
  96. """Factory-Methode: Erzeugt ein S7String-Objekt aus den von struct.unpack kommenden Rohdaten."""
  97. return cls(raw_bytes, is_wstring=is_wstring)
  98.  
  99.  
  100. def compile_s7_format(fmt: str) -> str:
  101. """
  102. Kompiliert S7-Formate inklusive Siemens-Strings (S=STRING, W=WSTRING)
  103. in standardkonforme Python struct-Formatstrings unter Beachtung des S7 Natural Alignments.
  104. """
  105. s7_fmt = ">"
  106. current_size = 0
  107.  
  108. for match in re.finditer(r"(\d*)([a-zA-Z])", fmt):
  109. count_str, dtype = match.groups()
  110. count = int(count_str) if count_str else 1
  111.  
  112. # 1. Füllbytes (x) direkt anhängen
  113. if dtype == "x":
  114. s7_fmt += f"{count}x"
  115. current_size += count
  116. continue
  117.  
  118. # 2. Siemens STRINGS (S) und WSTRINGS (W)
  119. if dtype in ("S", "W"):
  120. is_wstring = (dtype == "W")
  121. char_size = 2 if is_wstring else 1
  122. header_size = 4 if is_wstring else 2
  123.  
  124. total_bytes = header_size + (count * char_size)
  125. ssize = 2 if is_wstring else 1 # WSTRING richtet sich Wortweise (2 Bytes) aus
  126.  
  127. if (remainder := current_size % ssize) != 0:
  128. padding = ssize - remainder
  129. s7_fmt += f"{padding}x"
  130. current_size += padding
  131.  
  132. s7_fmt += f"{total_bytes}s"
  133. current_size += total_bytes
  134. continue
  135.  
  136. # 3. Standard-Datentypen
  137. ssize = struct.calcsize(f">{dtype}")
  138. if (remainder := current_size % ssize) != 0:
  139. padding_bytes = ssize - remainder
  140. s7_fmt += f"{padding_bytes}x"
  141. current_size += padding_bytes
  142.  
  143. s7_fmt += f"{count_str}{dtype}"
  144. current_size += ssize * count
  145.  
  146. return s7_fmt
  147.  
  148.  
  149.  
  150. # S7-Strukturdefinition (z.B. aus einem globalen DB):
  151. # 1 Byte (B), 1 normaler STRING mit max 10 Zeichen (10S), 1 Real/Float (f), 1 WSTRING mit max 5 Zeichen (5W)
  152. s7_schema = "b 10S f 5W"
  153.  
  154. # 1. Format-String einmalig für das Projekt kompilieren
  155. python_fmt = compile_s7_format(s7_schema)
  156. print(f"Kompiliertes Python-Format: {python_fmt}\n")
  157. # Erwartete Ausgabe enthält automatisch berechnete Füllbytes (x), um das Alignment zu halten
  158.  
  159. # ==============================================================================
  160. # LEGENDE 1: DATEN PACKEN (Senden zur SPS)
  161. # ==============================================================================
  162. my_string = S7String("Siemens", size=10, is_wstring=False)
  163. my_wstring = S7String("S7-15", size=5, is_wstring=True)
  164.  
  165. # Datentupel vorbereiten. Die S7String-Objekte müssen per bytes() übergeben werden.
  166. werte_zum_senden = (42, bytes(my_string), 123.45, bytes(my_wstring))
  167.  
  168. packed_bytes = struct.pack(python_fmt, *werte_zum_senden)
  169. print("SPS-Sende-Payload (Hex):", packed_bytes.hex())
  170.  
  171. # ==============================================================================
  172. # LEGENDE 2: DATEN ENTPACKEN (Empfangen von SPS via struct.unpack)
  173. # ==============================================================================
  174. # Wir simulieren den Empfang der 'packed_bytes' und jagen sie durch unpack
  175. unpacked_raw = struct.unpack(python_fmt, packed_bytes)
  176.  
  177. # Die entpackten Felder den Python-Variablen zuordnen
  178. s7_byte_val = unpacked_raw[0]
  179. s7_string_obj = S7String.from_bytes(unpacked_raw[1], is_wstring=False)
  180. s7_float_val = unpacked_raw[2]
  181. s7_wstring_obj = S7String.from_bytes(unpacked_raw[3], is_wstring=True)
  182.  
  183. print("\n--- Ausgelesene SPS-Daten ---")
  184. print(f"Byte-Wert: {s7_byte_val}")
  185. print(f"STRING (Max={len(s7_string_obj)}, Aktuell={s7_string_obj.slen}): '{s7_string_obj}'")
  186. print(f"Float-Wert: {s7_float_val:.2f}")
  187. print(f"WSTRING (Max={len(s7_wstring_obj)}, Aktuell={s7_wstring_obj.slen}): '{s7_wstring_obj}'")
  188.  
  189.  
  190. raise SystemExit("2. Teil nur mit sps")
  191.  
  192. import struct
  193. import time
  194. import snap7
  195. from snap7.util import *
  196.  
  197. # --- [Hier die S7String-Klasse und compile_s7_format einfügen] ---
  198. # (Aus der vorherigen Nachricht übernehmen)
  199.  
  200. # 1. Konfiguration der SPS-Verbindung und Datenstruktur
  201. PLC_IP = "192.168.0.1"
  202. PLC_RACK = 0
  203. PLC_SLOT = 1 # Bei S7-1200 meist 0, bei S7-1500 meist 1
  204. DB_NUMBER = 10 # Die Nummer des Datenbausteins (DB10)
  205.  
  206. # Unsere bekannte SPS-Struktur: 1 Byte, 1 String (max 12), 1 Füllbyte, 1 Real (Float)
  207. S7_STRUCTURE = "B 12S x f"
  208. python_format = compile_s7_format(S7_STRUCTURE)
  209. BUFFER_SIZE = struct.calcsize(python_format)
  210.  
  211. print(f"Kompiliertes Format: {python_format}")
  212. print(f"Erwartete Bytegröße aus DB: {BUFFER_SIZE} Bytes\n")
  213.  
  214. # 2. Verbindung zur SPS aufbauen
  215. client = snap7.client.Client()
  216. try:
  217. client.connect(PLC_IP, PLC_RACK, PLC_SLOT)
  218. print(f"Erfolgreich mit S7-SPS ({PLC_IP}) verbunden.")
  219.  
  220. # ==========================================
  221. # TEIL 1: DATEN AUS DEM DB LESEN
  222. # ==========================================
  223. print("\n--- Lese Daten aus DB ---")
  224. # Liest den gesamten DB-Bereich ab Offset 0 in ein Bytearray
  225. raw_bytes = client.db_read(DB_NUMBER, 0, BUFFER_SIZE)
  226.  
  227. # Rohe Bytes entpacken
  228. unpacked_list = struct.unpack(python_format, raw_bytes)
  229. data_iter = iter(unpacked_list)
  230.  
  231. # Den Werten Variablen zuweisen (Reihenfolge muss exakt stimmen!)
  232. res_byte = next(data_iter)
  233. res_string = S7String.from_struct_unpack(
  234. max_len=next(data_iter),
  235. act_len=next(data_iter),
  236. payload_bytes=next(data_iter)
  237. )
  238. res_float = next(data_iter)
  239.  
  240. print(f"Aktueller Wert Byte: {res_byte}")
  241. print(f"Aktueller Wert String: '{res_string}'")
  242. print(f"Aktueller Wert Float: {res_float}")
  243.  
  244. # ==========================================
  245. # TEIL 2: DATEN MODIFIZIEREN UND SCHREIBEN
  246. # ==========================================
  247. print("\n--- Schreibe neue Daten in DB ---")
  248.  
  249. # Neue Werte definieren
  250. neues_byte = 105
  251. res_string.set("SPS laeuft!") # Nutzt die sichere .set() Methode der Klasse
  252. neuer_float = 45.67
  253.  
  254. # Payload für struct.pack vorbereiten
  255. payload = (neues_byte, *res_string.to_struct_args(), neuer_float)
  256.  
  257. # Daten packen (erzeugt das perfekt ausgerichtete Big-Endian Byte-Paket)
  258. send_bytes = struct.pack(python_format, *payload)
  259.  
  260. # Bytes zurück in die SPS schreiben (DB_NUMBER, Start-Offset, Daten)
  261. client.db_write(DB_NUMBER, 0, send_bytes)
  262. print("Neue Daten erfolgreich in die SPS übertragen.")
  263.  
  264. except Exception as e:
  265. print(f"Fehler bei der SPS-Kommunikation: {e}")
  266.  
  267. finally:
  268. # Verbindung sauber trennen
  269. if client.get_connected():
  270. client.disconnect()
  271. print("\nVerbindung zur SPS getrennt.")
  272.  
Pasted 2026-08-22 14:27:48

Short link:

VicePaste is a public paste platform.

Login or Register to edit, delete and keep track of your pastes and more.

Recent Public Pastes
FAQS Contact
Copyright © 1986-2026 VicePaste. All rights reserved.