499 lines
26 KiB
Python
499 lines
26 KiB
Python
import re
|
|
import json
|
|
from dataclasses import dataclass, field
|
|
from typing import List, Dict, Optional, Union, Tuple, Any
|
|
import copy
|
|
|
|
# --- Estructuras de Datos ---
|
|
@dataclass
|
|
class ArrayDimension:
|
|
lower_bound: int
|
|
upper_bound: int
|
|
|
|
@property
|
|
def count(self) -> int:
|
|
return self.upper_bound - self.lower_bound + 1
|
|
|
|
@dataclass
|
|
class VariableInfo:
|
|
name: str
|
|
data_type: str # Tipo base limpio (ej. "INT", "REAL", nombre del UDT sin comillas)
|
|
byte_offset: float
|
|
size_in_bytes: int # Para BOOL arrays, bytes que ocupa. Para BOOL individual, 0.
|
|
bit_size: int = 0 # Para BOOL, 1. Para otros, 0.
|
|
udt_source_name: Optional[str] = None # Nombre original del UDT con comillas, si es UDT
|
|
string_length: Optional[int] = None
|
|
array_dimensions: List[ArrayDimension] = field(default_factory=list)
|
|
initial_value: Optional[str] = None # Valor de la declaración
|
|
current_value: Optional[str] = None # Valor efectivo (del bloque BEGIN o fallback a initial_value)
|
|
comment: Optional[str] = None
|
|
children: List['VariableInfo'] = field(default_factory=list) # Para STRUCTs o UDTs expandidos
|
|
is_udt_expanded_member: bool = False
|
|
# Nuevo campo para almacenar valores de elementos de array del bloque BEGIN
|
|
current_element_values: Optional[Dict[str, str]] = None # Ej: {"1": "val1", "1,0":"val_multi"}
|
|
|
|
@dataclass
|
|
class UdtInfo:
|
|
name: str # Nombre del UDT (sin comillas)
|
|
family: Optional[str] = None
|
|
version: Optional[str] = None
|
|
members: List[VariableInfo] = field(default_factory=list)
|
|
total_size_in_bytes: int = 0
|
|
|
|
@dataclass
|
|
class DbInfo:
|
|
name: str # Nombre del DB (sin comillas)
|
|
title: Optional[str] = None
|
|
family: Optional[str] = None
|
|
version: Optional[str] = None
|
|
members: List[VariableInfo] = field(default_factory=list)
|
|
total_size_in_bytes: int = 0
|
|
# Este diccionario almacena las asignaciones literales del bloque BEGIN
|
|
_initial_values_from_begin_block: Dict[str, str] = field(default_factory=dict)
|
|
|
|
@dataclass
|
|
class ParsedData:
|
|
udts: List[UdtInfo] = field(default_factory=list)
|
|
dbs: List[DbInfo] = field(default_factory=list)
|
|
|
|
@dataclass
|
|
class OffsetContext:
|
|
byte_offset: int = 0
|
|
bit_offset: int = 0 # 0-7
|
|
|
|
def get_combined_offset(self) -> float:
|
|
if self.bit_offset == 0:
|
|
return float(self.byte_offset)
|
|
return float(self.byte_offset * 10 + self.bit_offset) / 10.0
|
|
|
|
def advance_bits(self, num_bits: int):
|
|
self.bit_offset += num_bits
|
|
self.byte_offset += self.bit_offset // 8
|
|
self.bit_offset %= 8
|
|
|
|
def align_to_byte(self):
|
|
if self.bit_offset > 0:
|
|
self.byte_offset += 1
|
|
self.bit_offset = 0
|
|
|
|
def align_to_word(self):
|
|
self.align_to_byte()
|
|
if self.byte_offset % 2 != 0:
|
|
self.byte_offset += 1
|
|
# --- Fin Estructuras de Datos ---
|
|
|
|
S7_PRIMITIVE_SIZES = {
|
|
"BOOL": (0, 1, True), "BYTE": (1, 1, False), "CHAR": (1, 1, False),
|
|
"SINT": (1, 1, False), "USINT": (1, 1, False), "WORD": (2, 2, False),
|
|
"INT": (2, 2, False), "UINT": (2, 2, False), "S5TIME": (2, 2, False),
|
|
"DATE": (2, 2, False), "DWORD": (4, 2, False), "DINT": (4, 2, False),
|
|
"UDINT": (4, 2, False), "REAL": (4, 2, False), "TIME": (4, 2, False),
|
|
"TIME_OF_DAY": (4, 2, False), "TOD": (4, 2, False),
|
|
"LREAL": (8, 2, False), "LINT": (8, 2, False), "ULINT": (8, 2, False),
|
|
"LWORD": (8, 2, False), "DATE_AND_TIME": (8, 2, False), "DT": (8, 2, False),
|
|
}
|
|
|
|
class S7Parser:
|
|
def __init__(self):
|
|
self.parsed_data = ParsedData()
|
|
self.known_udts: Dict[str, UdtInfo] = {} # Clave es el nombre del UDT sin comillas
|
|
self.type_start_regex = re.compile(r'^\s*TYPE\s+"([^"]+)"', re.IGNORECASE)
|
|
self.db_start_regex = re.compile(r'^\s*DATA_BLOCK\s+"([^"]+)"', re.IGNORECASE)
|
|
self.property_regex = re.compile(r'^\s*([A-Z_]+)\s*:\s*(.+?)(?:\s*;)?\s*(?://.*)?$', re.IGNORECASE)
|
|
self.struct_start_regex = re.compile(r'^\s*STRUCT\b', re.IGNORECASE)
|
|
self.end_struct_regex = re.compile(r'^\s*END_STRUCT\b', re.IGNORECASE)
|
|
self.end_type_regex = re.compile(r'^\s*END_TYPE\b', re.IGNORECASE)
|
|
self.end_db_regex = re.compile(r'^\s*END_DATA_BLOCK\b', re.IGNORECASE)
|
|
self.begin_regex = re.compile(r'^\s*BEGIN\b', re.IGNORECASE)
|
|
|
|
self.var_regex_simplified = re.compile(
|
|
r'^\s*(?P<name>[a-zA-Z_][a-zA-Z0-9_]*)\s*:\s*'
|
|
r'(?P<typefull>'
|
|
r'(?:ARRAY\s*\[(?P<arraydims>[^\]]+?)\]\s*OF\s*)?'
|
|
r'(?P<basetype>(?:"[^"]+"|[a-zA-Z_][a-zA-Z0-9_]*))'
|
|
r'(?:\s*\[\s*(?P<stringlength>\d+)\s*\])?'
|
|
r')'
|
|
r'(?:\s*:=\s*(?P<initval>[^;]*?))??\s*'
|
|
r';?\s*$',
|
|
re.IGNORECASE
|
|
)
|
|
self.array_dim_regex = re.compile(r'(\d+)\s*\.\.\s*(\d+)')
|
|
# Regex para capturar elementos de array en el bloque BEGIN, ej: MyArray[1] o MyArray[1,2]
|
|
self.array_element_assign_regex_template = r'^\s*{array_var_name_escaped}\[(?P<indices>[0-9,\s]+)\]\s*$'
|
|
|
|
|
|
def _get_type_details(self, type_name_raw_cleaned: str) -> Tuple[int, int, bool, str]:
|
|
"""
|
|
type_name_raw_cleaned: Nombre del tipo ya sin comillas y en mayúsculas para primitivos,
|
|
o en su caso original para UDTs.
|
|
Returns: (size_in_bytes, alignment_req_bytes, is_bool, effective_type_name_for_lookup)
|
|
"""
|
|
type_name_upper = type_name_raw_cleaned.upper()
|
|
|
|
if type_name_upper in S7_PRIMITIVE_SIZES:
|
|
size, align, is_bool = S7_PRIMITIVE_SIZES[type_name_upper]
|
|
return size, align, is_bool, type_name_upper
|
|
elif type_name_raw_cleaned in self.known_udts: # Búsqueda case-sensitive para UDTs
|
|
udt = self.known_udts[type_name_raw_cleaned]
|
|
return udt.total_size_in_bytes, 2, False, type_name_raw_cleaned
|
|
elif type_name_upper == "STRUCT":
|
|
return 0, 2, False, "STRUCT"
|
|
raise ValueError(f"Tipo de dato desconocido o UDT no definido: '{type_name_raw_cleaned}'")
|
|
|
|
@staticmethod
|
|
def _adjust_children_offsets(children: List[VariableInfo], base_offset_add: float):
|
|
for child in children:
|
|
child.byte_offset += base_offset_add
|
|
if child.byte_offset == float(int(child.byte_offset)): # Limpiar .0 innecesario
|
|
child.byte_offset = float(int(child.byte_offset))
|
|
if child.children:
|
|
S7Parser._adjust_children_offsets(child.children, base_offset_add)
|
|
|
|
|
|
def _parse_struct_members(self, lines: List[str], current_line_idx: int,
|
|
parent_members_list: List[VariableInfo],
|
|
active_context: OffsetContext,
|
|
is_top_level_struct_in_block: bool = False) -> int:
|
|
idx_to_process = current_line_idx
|
|
|
|
while idx_to_process < len(lines):
|
|
original_line_text = lines[idx_to_process].strip()
|
|
|
|
line_to_parse = original_line_text
|
|
line_comment = None
|
|
comment_marker_idx = original_line_text.find("//")
|
|
if comment_marker_idx != -1:
|
|
line_to_parse = original_line_text[:comment_marker_idx].strip()
|
|
line_comment = original_line_text[comment_marker_idx + 2:].strip()
|
|
|
|
line_index_for_return = idx_to_process
|
|
idx_to_process += 1
|
|
|
|
if not line_to_parse: continue
|
|
|
|
if self.end_struct_regex.match(line_to_parse):
|
|
if not is_top_level_struct_in_block:
|
|
active_context.align_to_byte()
|
|
if active_context.byte_offset % 2 != 0: active_context.byte_offset += 1
|
|
return idx_to_process
|
|
|
|
if is_top_level_struct_in_block and \
|
|
(self.end_type_regex.match(line_to_parse) or \
|
|
self.end_db_regex.match(line_to_parse) or \
|
|
self.begin_regex.match(line_to_parse)):
|
|
active_context.align_to_byte()
|
|
if active_context.byte_offset % 2 != 0: active_context.byte_offset += 1
|
|
return line_index_for_return
|
|
|
|
var_match = self.var_regex_simplified.match(line_to_parse)
|
|
if var_match:
|
|
var_data = var_match.groupdict()
|
|
|
|
# data_type se refiere al tipo base limpio, udt_source_name al original con comillas
|
|
raw_base_type_from_regex = var_data['basetype'].strip()
|
|
clean_data_type = raw_base_type_from_regex.strip('"')
|
|
udt_source_name_val = raw_base_type_from_regex if raw_base_type_from_regex.startswith('"') else None
|
|
|
|
var_info = VariableInfo(name=var_data['name'],
|
|
data_type=clean_data_type,
|
|
byte_offset=0, size_in_bytes=0,
|
|
udt_source_name=udt_source_name_val)
|
|
|
|
if var_data.get('initval'):
|
|
var_info.initial_value = var_data['initval'].strip()
|
|
if line_comment:
|
|
var_info.comment = line_comment
|
|
|
|
num_array_elements = 1
|
|
if var_data['arraydims']:
|
|
for dim_match in self.array_dim_regex.finditer(var_data['arraydims']):
|
|
var_info.array_dimensions.append(ArrayDimension(int(dim_match.group(1)), int(dim_match.group(2))))
|
|
if var_info.array_dimensions:
|
|
for dim in var_info.array_dimensions: num_array_elements *= dim.count
|
|
|
|
if var_info.data_type.upper() == "STRUCT": # Un miembro definido como STRUCT explícitamente
|
|
active_context.align_to_word()
|
|
var_info.byte_offset = active_context.get_combined_offset()
|
|
|
|
nested_struct_context = OffsetContext()
|
|
idx_after_nested_struct = self._parse_struct_members(
|
|
lines, idx_to_process, var_info.children,
|
|
nested_struct_context, is_top_level_struct_in_block=False
|
|
)
|
|
var_info.size_in_bytes = nested_struct_context.byte_offset
|
|
|
|
# Ajustar offsets de los hijos del struct anidado para que sean globales
|
|
for child in var_info.children:
|
|
child.byte_offset += var_info.byte_offset
|
|
if child.byte_offset == float(int(child.byte_offset)):
|
|
child.byte_offset = float(int(child.byte_offset))
|
|
if child.children:
|
|
S7Parser._adjust_children_offsets(child.children, var_info.byte_offset)
|
|
|
|
|
|
active_context.byte_offset += var_info.size_in_bytes
|
|
idx_to_process = idx_after_nested_struct
|
|
|
|
elif var_info.data_type.upper() == "STRING" and var_data['stringlength']:
|
|
var_info.string_length = int(var_data['stringlength'])
|
|
unit_size = var_info.string_length + 2 # N chars + 2 header bytes
|
|
active_context.align_to_word() # STRING se alinea a palabra
|
|
var_info.byte_offset = active_context.get_combined_offset()
|
|
var_info.size_in_bytes = unit_size * num_array_elements
|
|
active_context.byte_offset += var_info.size_in_bytes
|
|
|
|
else: # Primitivo o instancia de UDT
|
|
# Usar var_info.data_type (limpio) para _get_type_details
|
|
unit_size_bytes, unit_alignment_req, is_bool, type_name_for_udt_lookup = self._get_type_details(var_info.data_type)
|
|
|
|
if is_bool:
|
|
var_info.bit_size = 1
|
|
var_info.byte_offset = active_context.get_combined_offset()
|
|
active_context.advance_bits(num_array_elements)
|
|
|
|
start_byte_abs = int(var_info.byte_offset)
|
|
start_bit_in_byte = int(round((var_info.byte_offset - start_byte_abs) * 10))
|
|
if num_array_elements == 1:
|
|
var_info.size_in_bytes = 0 # Convención para un solo bit
|
|
else: # Array de BOOLs
|
|
bits_rem = num_array_elements; bytes_spanned = 0
|
|
if start_bit_in_byte > 0:
|
|
bits_in_first = 8 - start_bit_in_byte
|
|
if bits_rem <= bits_in_first: bytes_spanned = 1
|
|
else: bytes_spanned = 1; bits_rem -= bits_in_first; bytes_spanned += (bits_rem + 7) // 8
|
|
else: bytes_spanned = (bits_rem + 7) // 8
|
|
var_info.size_in_bytes = bytes_spanned
|
|
|
|
else: # No es BOOL (Primitivo > byte, o UDT)
|
|
active_context.align_to_byte()
|
|
if unit_alignment_req == 2: active_context.align_to_word()
|
|
|
|
var_info.byte_offset = active_context.get_combined_offset()
|
|
var_info.size_in_bytes = unit_size_bytes * num_array_elements
|
|
active_context.byte_offset += var_info.size_in_bytes
|
|
|
|
if type_name_for_udt_lookup in self.known_udts and not is_bool: # Es una instancia de UDT
|
|
udt_def = self.known_udts[type_name_for_udt_lookup]
|
|
udt_instance_abs_start_offset = var_info.byte_offset
|
|
|
|
for udt_member_template in udt_def.members:
|
|
expanded_member = copy.deepcopy(udt_member_template)
|
|
expanded_member.is_udt_expanded_member = True
|
|
expanded_member.byte_offset += udt_instance_abs_start_offset # Hacer absoluto
|
|
if expanded_member.byte_offset == float(int(expanded_member.byte_offset)):
|
|
expanded_member.byte_offset = float(int(expanded_member.byte_offset))
|
|
|
|
if expanded_member.children: # Si el miembro del UDT era un STRUCT
|
|
S7Parser._adjust_children_offsets(expanded_member.children, udt_instance_abs_start_offset)
|
|
|
|
var_info.children.append(expanded_member)
|
|
parent_members_list.append(var_info)
|
|
else: # No es variable, ni fin de struct, ni fin de bloque principal
|
|
if line_to_parse and \
|
|
not self.struct_start_regex.match(line_to_parse): # Si no es tampoco un "STRUCT" de inicio de miembro
|
|
print(f"DEBUG: Line not parsed as variable or known keyword: Original='{original_line_text}' | Processed='{line_to_parse}'")
|
|
|
|
# El padding final del struct/bloque se maneja al detectar END_STRUCT o END_TYPE/DB/BEGIN
|
|
return idx_to_process
|
|
|
|
def _parse_begin_block(self, lines: List[str], start_idx: int, db_info: DbInfo) -> int:
|
|
idx = start_idx
|
|
assignment_regex = re.compile(r'^\s*(?P<path>.+?)\s*:=\s*(?P<value>.+?)\s*;?\s*$', re.IGNORECASE)
|
|
|
|
while idx < len(lines):
|
|
original_line = lines[idx].strip()
|
|
line_to_parse = original_line
|
|
comment_marker = original_line.find("//")
|
|
if comment_marker != -1:
|
|
line_to_parse = original_line[:comment_marker].strip()
|
|
# Comentarios en líneas de asignación no se guardan estructuradamente por ahora
|
|
|
|
if self.end_db_regex.match(line_to_parse):
|
|
return idx
|
|
|
|
idx += 1
|
|
if not line_to_parse: continue
|
|
|
|
match = assignment_regex.match(line_to_parse)
|
|
if match:
|
|
path = match.group("path").strip()
|
|
value = match.group("value").strip().rstrip(';').strip()
|
|
db_info._initial_values_from_begin_block[path] = value
|
|
# else:
|
|
# print(f"DEBUG: Line in BEGIN not matched: '{original_line}'")
|
|
raise SyntaxError("Se esperaba END_DATA_BLOCK después de la sección BEGIN.")
|
|
|
|
def _apply_current_values(self, members: List[VariableInfo], begin_values: Dict[str, str], current_path_prefix: str = ""):
|
|
for var_info in members:
|
|
full_member_path = f"{current_path_prefix}{var_info.name}"
|
|
|
|
if var_info.array_dimensions:
|
|
var_info.current_element_values = {}
|
|
# Lógica para buscar elementos de array en begin_values
|
|
# Asumimos que las claves en begin_values son como "ArrayName[idx]" o "Struct.ArrayName[idx,idx2]"
|
|
|
|
# Construir un patrón para buscar elementos de este array específico
|
|
# Escapar el nombre del array por si contiene caracteres especiales de regex
|
|
# Y luego añadir el patrón para los corchetes e índices.
|
|
# Esto es una simplificación. Un parseo robusto de índices multidimensionales es más complejo.
|
|
# Ejemplo de clave en begin_values: "Spare1[1]", "Path.To.Array[5]"
|
|
prefix_for_search = full_member_path + "[" # ej. "Spare1["
|
|
|
|
for key_in_begin, val_in_begin in begin_values.items():
|
|
if key_in_begin.startswith(prefix_for_search) and key_in_begin.endswith("]"):
|
|
# Extraer el contenido de los corchetes
|
|
try:
|
|
indices_str = key_in_begin[len(prefix_for_search):-1]
|
|
var_info.current_element_values[indices_str] = val_in_begin
|
|
except:
|
|
print(f"Advertencia: No se pudo parsear el índice para la clave del array: {key_in_begin}")
|
|
|
|
if not var_info.current_element_values:
|
|
var_info.current_element_values = None # No guardar dict vacío
|
|
|
|
# El current_value del array en sí mismo no se establece a menos que haya una asignación global
|
|
# como "ArrayName := ..." lo cual es raro para arrays complejos en BEGIN.
|
|
if full_member_path in begin_values: # Asignación global al array?
|
|
var_info.current_value = begin_values[full_member_path]
|
|
|
|
|
|
elif full_member_path in begin_values: # Variable simple
|
|
var_info.current_value = begin_values[full_member_path]
|
|
elif var_info.initial_value is not None: # Fallback a valor de declaración
|
|
var_info.current_value = var_info.initial_value
|
|
|
|
# Aplicar recursivamente para hijos de STRUCTs (no UDTs expandidos directamente aquí)
|
|
if var_info.children and not var_info.is_udt_expanded_member: # Es un STRUCT definido aquí
|
|
self._apply_current_values(var_info.children, begin_values, f"{full_member_path}.")
|
|
# Para miembros expandidos de UDT (is_udt_expanded_member = True),
|
|
# sus current_values se establecen si su ruta completa (ej. "MyUdtInstance._Member") está en begin_values.
|
|
# Esto se maneja en la iteración principal si los hijos UDT son procesados.
|
|
# La llamada recursiva anterior para STRUCTs anidados maneja "MyStruct.MyUdtInstance._Member".
|
|
|
|
# Si var_info es una instancia de UDT (tiene udt_source_name y children son sus miembros expandidos)
|
|
elif var_info.udt_source_name and var_info.children:
|
|
self._apply_current_values(var_info.children, begin_values, f"{full_member_path}.")
|
|
|
|
|
|
def parse_file(self, filepath: str) -> ParsedData:
|
|
try:
|
|
with open(filepath, 'r', encoding='utf-8-sig') as f: lines = f.readlines()
|
|
except Exception as e:
|
|
print(f"Error al leer el archivo {filepath}: {e}"); return self.parsed_data
|
|
|
|
current_block_handler: Optional[Union[UdtInfo, DbInfo]] = None
|
|
active_block_context = OffsetContext()
|
|
|
|
idx = 0
|
|
while idx < len(lines):
|
|
original_line_with_space = lines[idx]
|
|
original_line = original_line_with_space.strip()
|
|
|
|
line_to_parse = original_line
|
|
comment_marker = original_line.find("//")
|
|
if comment_marker != -1:
|
|
line_to_parse = original_line[:comment_marker].strip()
|
|
|
|
type_match = self.type_start_regex.match(line_to_parse)
|
|
db_match = self.db_start_regex.match(line_to_parse)
|
|
|
|
if type_match:
|
|
udt_name = type_match.group(1)
|
|
current_block_handler = UdtInfo(name=udt_name)
|
|
self.parsed_data.udts.append(current_block_handler)
|
|
active_block_context = OffsetContext()
|
|
idx +=1; continue
|
|
elif db_match:
|
|
db_name = db_match.group(1)
|
|
current_block_handler = DbInfo(name=db_name)
|
|
self.parsed_data.dbs.append(current_block_handler)
|
|
active_block_context = OffsetContext()
|
|
idx +=1; continue
|
|
|
|
if not current_block_handler:
|
|
idx +=1; continue
|
|
|
|
prop_match = self.property_regex.match(original_line) # Las propiedades pueden tener comentarios
|
|
struct_keyword_match = self.struct_start_regex.match(line_to_parse)
|
|
|
|
if prop_match:
|
|
key, value = prop_match.group(1).upper(), prop_match.group(2).strip()
|
|
attr = key.lower()
|
|
if hasattr(current_block_handler, attr): setattr(current_block_handler, attr, value)
|
|
|
|
elif struct_keyword_match and not current_block_handler.members: # Inicio del STRUCT principal del bloque
|
|
idx = self._parse_struct_members(lines, idx + 1, current_block_handler.members,
|
|
active_block_context, is_top_level_struct_in_block=True)
|
|
continue
|
|
|
|
elif self.begin_regex.match(line_to_parse) and isinstance(current_block_handler, DbInfo):
|
|
# El tamaño total de la sección de declaración se finaliza aquí
|
|
current_block_handler.total_size_in_bytes = active_block_context.byte_offset
|
|
idx = self._parse_begin_block(lines, idx + 1, current_block_handler)
|
|
continue
|
|
|
|
elif self.end_type_regex.match(line_to_parse) and isinstance(current_block_handler, UdtInfo):
|
|
# Si total_size_in_bytes no fue establecido (ej. UDT vacío o error), calcularlo aquí
|
|
if current_block_handler.total_size_in_bytes == 0:
|
|
current_block_handler.total_size_in_bytes = active_block_context.byte_offset
|
|
self.known_udts[current_block_handler.name] = current_block_handler
|
|
print(f"Parsed UDT: {current_block_handler.name}, Size: {current_block_handler.total_size_in_bytes} bytes. Members: {len(current_block_handler.members)}")
|
|
current_block_handler = None
|
|
|
|
elif self.end_db_regex.match(line_to_parse) and isinstance(current_block_handler, DbInfo):
|
|
if current_block_handler.total_size_in_bytes == 0 : # Si no hubo sección BEGIN
|
|
current_block_handler.total_size_in_bytes = active_block_context.byte_offset
|
|
# Aplicar valores del bloque BEGIN a los current_value de los miembros
|
|
self._apply_current_values(current_block_handler.members, current_block_handler._initial_values_from_begin_block)
|
|
print(f"Parsed DB: {current_block_handler.name}, Decl. Size: {current_block_handler.total_size_in_bytes} bytes. Members: {len(current_block_handler.members)}")
|
|
current_block_handler = None
|
|
|
|
idx += 1
|
|
|
|
return self.parsed_data
|
|
|
|
# --- Serializador JSON Personalizado ---
|
|
def custom_json_serializer(obj: Any) -> Any:
|
|
if isinstance(obj, OffsetContext): return None # No serializar OffsetContext
|
|
if hasattr(obj, '__dict__'):
|
|
d = {k: v for k, v in obj.__dict__.items()
|
|
if v is not None and \
|
|
not (isinstance(v, list) and not v)} # Ya no filtra _initial_values_from_begin_block
|
|
|
|
if isinstance(obj, VariableInfo):
|
|
# Asegurar que 'is_udt_expanded_member' esté presente si es False
|
|
if 'is_udt_expanded_member' not in d and obj.is_udt_expanded_member is False:
|
|
d['is_udt_expanded_member'] = False
|
|
elif obj.is_udt_expanded_member is True: # Asegurar que se incluya si es True
|
|
d['is_udt_expanded_member'] = True
|
|
|
|
# Incluir current_element_values si existe y no está vacío
|
|
if hasattr(obj, 'current_element_values') and obj.current_element_values:
|
|
d['current_element_values'] = obj.current_element_values
|
|
# Eliminarlo si está presente pero vacío (ya que el default es None)
|
|
elif hasattr(obj, 'current_element_values') and 'current_element_values' in d and not d['current_element_values']:
|
|
del d['current_element_values']
|
|
return d
|
|
raise TypeError(f"Object of type {obj.__class__.__name__} is not JSON serializable")
|
|
|
|
# --- Bloque Principal ---
|
|
if __name__ == "__main__":
|
|
parser = S7Parser()
|
|
# Asegúrate que este path es el correcto para tu archivo .db o .db.txt
|
|
filepath = "db1001_format.db.txt"
|
|
|
|
print(f"Intentando parsear el archivo: {filepath}")
|
|
parsed_result = parser.parse_file(filepath)
|
|
|
|
# Nuevo nombre para el archivo JSON de salida para esta versión
|
|
json_output_filename = "parsed_s7_data.json"
|
|
print(f"\nParseo completo. Intentando serializar a JSON.")
|
|
|
|
try:
|
|
json_output = json.dumps(parsed_result, default=custom_json_serializer, indent=2)
|
|
# print(json_output) # Descomentar para ver el JSON en la consola
|
|
with open(json_output_filename, "w", encoding='utf-8') as f:
|
|
f.write(json_output)
|
|
print(f"Resultado guardado en: {json_output_filename}")
|
|
except Exception as e:
|
|
print(f"Error durante la serialización JSON o escritura de archivo: {e}") |