commit 75cc7cde252b5c763c1d73d13eed470da17eaf3d Author: Jonas Bengtsson Date: Fri Oct 2 06:10:55 2026 +0200 🎉 Initial commit diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..2ffcf2f --- /dev/null +++ b/.gitignore @@ -0,0 +1,9 @@ +.venv/ +__pycache__/ +*.py[cod] +*.egg-info/ +build/ +dist/ +.pytest_cache/ +output/ +*.xlsx \ No newline at end of file diff --git a/README.md b/README.md new file mode 100644 index 0000000..ba66a6c --- /dev/null +++ b/README.md @@ -0,0 +1,176 @@ +# mitsubishi-label-converter-py + +A Python CLI tool that converts Mitsubishi GX Works3 global label XML exports +(`Global.xml`, PLCopen TC6 format, typically UTF-16 LE encoded) **to Excel** +and **back to XML** with a lossless round-trip. + +Each `` section group in the source file becomes its own worksheet: +sections sharing the same name (e.g. a plain and a retain ``Global`` list) +are merged into one sheet, with the per-variable retain flag in the +``Retain`` column. The reverse conversion splits the rows back into the +original sections. Every `` becomes one row with: + +| Column | Content | +|-----------|-------------------------------------------------------------------------| +| Name | The variable's `name` attribute | +| Address | The variable's `address` attribute (empty when unassigned) | +| Type | The declared type (see below) | +| Comment | The variable comment from `variableComments` addData (if present) | +| Line | Original source line number from `variableLineNumber` addData | +| Retain | `TRUE` when the variable belongs to a `retain="true"` globalVars section | + +**Structured variables** (derived types like `ST_Din_STA`) are expanded: each +member with a device address gets its own row directly under the variable, +using a dotted name (`PLC_FEL_STA.Value -> D0.0`). The addresses come from +`variableStructDeviceAssignment` addData. + +**Member types** — the Global XML export carries only member names and +addresses; member *types* live in a separate SDT export. Pass it via `--sdt` +to populate the Type column of member rows (`BOOL`, `WORD`, `STRING[16]`, +`ARRAY[1..4] OF INT`, nested struct names, ...). Without the flag, member +Type cells stay empty. + +A **`Meta`** worksheet records document metadata (fileHeader, contentHeader, +PLC type) plus the sheet-to-section mapping (names and `varListProperties` +timestamps) so the reverse conversion is lossless. Retain information lives +in the `Retain` column, not in Meta. + +## Type rendering + +- Primitive types (``, ``, ``, ``, ...) are written as-is. +- `` is written as `STRING[12]`. +- Structured/derived types (``) are written using the + derived name, e.g. `FB_Din`. + +The reverse conversion recognizes primitives from the standard IEC set; +everything else is emitted as a `` reference again. + +## Validation report + +The `report` subcommand runs consistency checks on an XML export or an Excel +workbook and prints the findings grouped by severity: + +| Severity | Check | Code | +|----------|-------|------| +| error | duplicate variable names | `NAME_DUPLICATE` | +| error | identical device address used by several variables | `ADDR_DUPLICATE` | +| warning | bit address inside another variable's word range | `ADDR_BIT_IN_WORD` | +| warning | word ranges of two variables overlap (types sized in 16-bit registers; `STRING[n]` occupies `(n+2)/2` registers) | `ADDR_OVERLAP` | +| warning | address in an unrecognized format | `ADDR_FORMAT` | +| warning | struct member not defined in the SDT struct | `MEMBER_UNKNOWN` | +| info | struct members without a device address | `MEMBER_UNASSIGNED` | +| info | structured variable without any device assignment | `STRUCT_UNASSIGNED` | +| info | derived type missing from the SDT file (aggregated per type) | `TYPE_UNKNOWN` | + +Members of the same structured variable may legitimately share word +registers, so only cross-variable collisions are reported. The command +exits with code 1 when errors are found (0 for warnings/info only), so it +can gate CI pipelines. Address parsing handles word-level (`D450`), +bit-level (`D6005.0`), and hexadecimal bit suffixes (`D201.A` = bit 10). + +## Installation + +```bash +# from the project root +pip install -e .[dev] +``` + +## Usage + +```bash +# XML -> Excel: output is written next to the source file (Global.xlsx) +mitsubishi-label-converter to-xlsx assets/Global.xml + +# with member types from an SDT (structured data type) export +mitsubishi-label-converter to-xlsx assets/Global.xml --sdt assets/SDT.xml + +# explicit output path +mitsubishi-label-converter to-xlsx assets/Global.xml -o labels.xlsx + +# Excel -> XML: regenerate a GX Works3-compatible XML file +mitsubishi-label-converter to-xml assets/Global.xlsx -o output/Global.xml + +# validate labels (XML or Excel input); exit code 1 on errors (CI-friendly) +mitsubishi-label-converter report assets/Global.xml --sdt assets/SDT.xml + +# save the report to a file instead of stdout +mitsubishi-label-converter report assets/Global.xlsx -o output/validation.txt + +# overwrite existing outputs +mitsubishi-label-converter to-xlsx assets/Global.xml --force +``` + +The round-trip is lossless: parsing the regenerated XML yields exactly the +same sections, variables, members, comments, line numbers, retain flags, +and metadata as the original file. The `to-xml` conversion regroups the +rows of a merged sheet by their `Retain` flag, restoring the original +section split (plain first, retain second). Member types from the SDT file +are Excel-only decoration and are intentionally not written to the XML (the +XML format stores member types in its `dataTypes` section, not in device +assignments). + +## Development + +```bash +pip install -e .[dev] +pytest +``` + +## TODO / Roadmap + +Done: + +- [x] **Expose struct members with addresses in Excel** — dotted member rows + inline under each structured variable (189 structured variables -> 1,220 + member rows in the sample file). +- [x] **Member types from SDT export** — `--sdt` flag populates the Type + column of member rows from a structured data type XML export; without it + the workbook is unchanged. +- [x] **Merged sheets + Retain column** — sections sharing a name merge into + one worksheet; the per-variable `Retain` column marks retain variables, + and `to-xml` regroups rows by flag to restore the original sections. +- [x] **Validation report** — `report` subcommand checks duplicate + names/addresses, address overlaps (bit-in-word, word ranges), struct + member consistency, and unassigned structs; works on XML and Excel + input; CI-friendly exit codes. +- [x] **Excel -> XML round-trip conversion** — `to-xml` subcommand; + `Meta` sheet carries document metadata and sheet mapping; verified + lossless against the real 290-variable file. +- [x] **CLI subcommands** — `to-xlsx` / `to-xml` with `-o/--output` and + `--force` flags. + +Planned: + +1. **`--all-members`: expose SDT members without device addresses (requested)** + GX Works3 only exports struct members that carry a device address; a + structured variable without any assignment (e.g. `GP01_RUNTIME_PAR` of + type `ST_Runtime_PAR`) has no member rows at all — its internal members + (`StartsTot`, `RunTimeTot`, ...) exist only in the SDT file. Plan: + - `expand_missing_members()` in `sdt_parser.py`: append every SDT-defined + member missing from the device assignment as an address-less row + (Type populated, Retain inherited from the parent variable). + - `--all-members` flag on `to-xlsx` (requires `--sdt`). + - `xml_writer` must skip address-less `` entries so `to-xml` + stays lossless (the original XML has no such members). + - Optional follow-up: recursive expansion for nested struct members + (`Parent.Sub.uHour`, e.g. `stError_Time -> M+FX5UCPU_Error_Time`). + - Verification: fabricated fixture (`tests/test_all_members.py`) with + (a) a partially assigned struct, (b) a struct variable with no + assignment at all, (c) both in the retain section. + +2. **CLI conveniences** + - `--list` flag: print the parsed label tables to the terminal. + - `--sheet-per-type` flag: split variables into one worksheet per type. + - `--filter ` flag: convert only variables matching a name pattern. + - `--round-trip` flag: convert back and forth and verify losslessness + automatically. + +3. **Multiple input files** — accept several XML files in one invocation, + each producing its own workbook (e.g. when a project contains several + resources/POUs). + +4. **CSV output** — `--format csv` to emit one CSV file per globalVars + section, for tools that cannot read `.xlsx`. + +5. **GUI drag & drop (long-term)** — a small GUI (e.g. tkinter or a web UI) + for drag-and-drop conversion for non-CLI users. \ No newline at end of file diff --git a/assets/21014JON-B14_20260930.gx3 b/assets/21014JON-B14_20260930.gx3 new file mode 100644 index 0000000..76f4dbb Binary files /dev/null and b/assets/21014JON-B14_20260930.gx3 differ diff --git a/assets/Global.xml b/assets/Global.xml new file mode 100644 index 0000000..b71380e Binary files /dev/null and b/assets/Global.xml differ diff --git a/assets/SDT.xml b/assets/SDT.xml new file mode 100644 index 0000000..7b3b0e3 Binary files /dev/null and b/assets/SDT.xml differ diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..9f1c816 --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,24 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "mitsubishi-label-converter" +version = "0.1.0" +description = "Convert Mitsubishi GX Works3 global label XML files to Excel workbooks" +readme = "README.md" +requires-python = ">=3.10" +dependencies = [ + "openpyxl>=3.1", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0", +] + +[project.scripts] +mitsubishi-label-converter = "mitsubishi_label_converter.cli:main" + +[tool.setuptools.packages.find] +where = ["src"] \ No newline at end of file diff --git a/src/mitsubishi_label_converter/__init__.py b/src/mitsubishi_label_converter/__init__.py new file mode 100644 index 0000000..dded564 --- /dev/null +++ b/src/mitsubishi_label_converter/__init__.py @@ -0,0 +1,3 @@ +"""Convert Mitsubishi GX Works3 global label XML files to Excel.""" + +__version__ = "0.1.0" \ No newline at end of file diff --git a/src/mitsubishi_label_converter/cli.py b/src/mitsubishi_label_converter/cli.py new file mode 100644 index 0000000..e0334f6 --- /dev/null +++ b/src/mitsubishi_label_converter/cli.py @@ -0,0 +1,265 @@ +"""CLI entry point: convert GX Works3 XML <-> Excel workbooks.""" + +from __future__ import annotations + +import argparse +import sys +import xml.etree.ElementTree as ET +from pathlib import Path + +from .excel import write_workbook +from .parser import parse_global_vars, parse_project_metadata +from .sdt_parser import apply_struct_types, parse_struct_definitions +from .validator import validate + + +def _is_excel(path: Path) -> bool: + return path.suffix.lower() == ".xlsx" + + +def build_arg_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + prog="mitsubishi-label-converter", + description=( + "Convert a Mitsubishi GX Works3 global label XML file to an Excel " + "workbook and back. Each section becomes its own " + "worksheet; a Meta sheet holds metadata for lossless round-trips." + ), + ) + subparsers = parser.add_subparsers(dest="command", required=True) + + to_xlsx = subparsers.add_parser( + "to-xlsx", + help="Convert a GX Works3 XML export to an Excel workbook", + ) + to_xlsx.add_argument( + "input", + type=Path, + help="Path to the GX Works3 XML export (e.g. assets/Global.xml)", + ) + to_xlsx.add_argument( + "-o", + "--output", + type=Path, + default=None, + help=( + "Output .xlsx path. Default: same name and directory as the " + "input file (e.g. Global.xml -> Global.xlsx)." + ), + ) + to_xlsx.add_argument( + "--sdt", + type=Path, + default=None, + metavar="SDT_XML", + help=( + "Optional path to an SDT (structured data type) XML export. " + "When given, the Type column of struct member rows is " + "populated from the data type definitions." + ), + ) + to_xlsx.add_argument( + "--force", + action="store_true", + help="Overwrite the output file if it already exists.", + ) + + report_parser = subparsers.add_parser( + "report", + help="Validate labels and print a findings report", + description=( + "Run consistency checks (duplicate names/addresses, address " + "overlaps, struct member mismatches) on a GX Works3 XML export " + "or an Excel workbook and print the findings." + ), + ) + report_parser.add_argument( + "input", + type=Path, + help=( + "Path to a GX Works3 XML export (.xml) or an Excel workbook " + "(.xlsx)" + ), + ) + report_parser.add_argument( + "--sdt", + type=Path, + default=None, + metavar="SDT_XML", + help=( + "Optional path to an SDT (structured data type) XML export; " + "enables struct member consistency checks." + ), + ) + report_parser.add_argument( + "-o", + "--output", + type=Path, + default=None, + help="Write the report to this file instead of stdout.", + ) + + to_xml = subparsers.add_parser( + "to-xml", + help="Convert an Excel workbook back to a GX Works3 XML file", + ) + to_xml.add_argument( + "input", + type=Path, + help="Path to the Excel workbook (e.g. assets/Global.xlsx)", + ) + to_xml.add_argument( + "-o", + "--output", + type=Path, + default=None, + help=( + "Output .xml path. Default: same name and directory as the " + "input file (e.g. Global.xlsx -> Global.xml)." + ), + ) + to_xml.add_argument( + "--force", + action="store_true", + help="Overwrite the output file if it already exists.", + ) + return parser + + +def default_output_path(input_path: Path, suffix: str) -> Path: + return input_path.with_suffix(suffix) + + +def cmd_to_xlsx(args: argparse.Namespace) -> int: + if not args.input.is_file(): + print(f"error: input file not found: {args.input}", file=sys.stderr) + return 2 + + output = args.output or default_output_path(args.input, ".xlsx") + if output.exists() and not args.force: + print( + f"error: output file already exists: {output} " + "(use --force to overwrite)", + file=sys.stderr, + ) + return 2 + + try: + sections = parse_global_vars(args.input) + metadata = parse_project_metadata(args.input) + except ET.ParseError as exc: + print(f"error: failed to parse XML: {exc}", file=sys.stderr) + return 2 + + if args.sdt is not None: + if not args.sdt.is_file(): + print( + f"error: SDT file not found: {args.sdt}", file=sys.stderr + ) + return 2 + try: + definitions = parse_struct_definitions(args.sdt) + except ET.ParseError as exc: + print( + f"error: failed to parse SDT XML: {exc}", file=sys.stderr + ) + return 2 + apply_struct_types(sections, definitions) + metadata.project_add_data["sdtFile"] = args.sdt.name + + if not sections: + print(f"warning: no sections found in {args.input}") + + write_workbook(sections, output, metadata=metadata) + + total = sum(len(section.variables) for section in sections) + sheet_count = len({section.name or "Global" for section in sections}) + print( + f"Converted {total} variables in {sheet_count} sheet(s) " + f"-> {output}" + ) + return 0 + + +def cmd_to_xml(args: argparse.Namespace) -> int: + from .excel_reader import read_workbook + from .xml_writer import write_xml + + if not args.input.is_file(): + print(f"error: input file not found: {args.input}", file=sys.stderr) + return 2 + + output = args.output or default_output_path(args.input, ".xml") + if output.exists() and not args.force: + print( + f"error: output file already exists: {output} " + "(use --force to overwrite)", + file=sys.stderr, + ) + return 2 + + try: + sections, metadata = read_workbook(args.input) + except Exception as exc: # noqa: BLE001 - user-facing error + print(f"error: failed to read workbook: {exc}", file=sys.stderr) + return 2 + + write_xml(sections, metadata, output) + total = sum(len(section.variables) for section in sections) + print(f"Converted {total} variables in {len(sections)} section(s) -> {output}") + return 0 + + +def cmd_report(args: argparse.Namespace) -> int: + if not args.input.is_file(): + print(f"error: input file not found: {args.input}", file=sys.stderr) + return 2 + + definitions: dict[str, dict[str, str]] = {} + if args.sdt is not None: + if not args.sdt.is_file(): + print(f"error: SDT file not found: {args.sdt}", file=sys.stderr) + return 2 + try: + definitions = parse_struct_definitions(args.sdt) + except ET.ParseError as exc: + print(f"error: failed to parse SDT XML: {exc}", file=sys.stderr) + return 2 + + try: + if _is_excel(args.input): + from .excel_reader import read_workbook + + sections, _ = read_workbook(args.input) + else: + sections = parse_global_vars(args.input) + except Exception as exc: # noqa: BLE001 - user-facing error + print(f"error: failed to read input: {exc}", file=sys.stderr) + return 2 + + validation = validate(sections, definitions or None) + text = validation.format_text() + if args.output is not None: + args.output.parent.mkdir(parents=True, exist_ok=True) + args.output.write_text(text, encoding="utf-8") + print(f"Wrote report -> {args.output}") + else: + print(text) + + # Non-zero exit when errors were found (useful in CI pipelines). + return 1 if validation.has_errors else 0 + + +def main(argv: list[str] | None = None) -> int: + args = build_arg_parser().parse_args(argv) + if args.command == "to-xlsx": + return cmd_to_xlsx(args) + if args.command == "to-xml": + return cmd_to_xml(args) + if args.command == "report": + return cmd_report(args) + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) \ No newline at end of file diff --git a/src/mitsubishi_label_converter/excel.py b/src/mitsubishi_label_converter/excel.py new file mode 100644 index 0000000..a5a3a39 --- /dev/null +++ b/src/mitsubishi_label_converter/excel.py @@ -0,0 +1,201 @@ +"""Write parsed label sections to an Excel workbook.""" + +from __future__ import annotations + +from pathlib import Path + +from openpyxl import Workbook +from openpyxl.styles import Alignment, Font, PatternFill +from openpyxl.utils import get_column_letter + +from .models import GlobalVarSection, ProjectMetadata + +HEADER_ROW = ["Name", "Address", "Type", "Comment", "Line", "Retain"] + +RETAIN_TRUE = "TRUE" +RETAIN_FALSE = "FALSE" + +META_SHEET_NAME = "Meta" + +_HEADER_FONT = Font(bold=True) +_HEADER_FILL = PatternFill("solid", fgColor="DDEBF7") + + +def _sanitize_sheet_name(name: str) -> str: + """Remove characters Excel forbids in worksheet titles.""" + forbidden = set("[]:*?/\\") + cleaned = "".join(ch if ch not in forbidden else "_" for ch in name) + return cleaned.strip() or "Global" + + +def merge_sections(sections: list[GlobalVarSection]) -> list[GlobalVarSection]: + """Merge globalVars sections that share the same name into one. + + All sections named ``Global`` (e.g. one plain, one retain) become a + single worksheet; each variable row carries its retain flag in the + ``Retain`` column so the reverse conversion can split them back apart. + """ + merged: list[GlobalVarSection] = [] + index_by_name: dict[str, int] = {} + for section in sections: + key = section.name or "Global" + if key in index_by_name: + target = merged[index_by_name[key]] + # The merged section keeps the first section's name; variables + # from both are appended, preserving document order. + merged[index_by_name[key]] = GlobalVarSection( + name=target.name, + retain=target.retain or section.retain, + variables=target.variables + section.variables, + modification_date_time=target.modification_date_time + or section.modification_date_time, + ) + else: + index_by_name[key] = len(merged) + merged.append(section) + return merged + + +def _unique_sheet_names(sections: list[GlobalVarSection]) -> list[str]: + """Build unique worksheet titles. + + Excel sheet names must be unique, limited to 31 characters, and may not + contain ``[]:*?/\\``. Collisions fall back to a numeric suffix. + """ + titles: list[str] = [] + used: set[str] = set() + for position, section in enumerate(sections, start=1): + base = _sanitize_sheet_name(section.name or "Global")[:31] + if base in used: + suffix = f"_{position}" + candidate = f"{base[:31 - len(suffix)]}{suffix}" + else: + candidate = base + titles.append(candidate) + used.add(candidate) + return titles + + +def _format_line(line_number: int | None) -> int | str: + """Line number cell value (int when known, blank otherwise).""" + return line_number if line_number is not None else "" + + +def _write_section_rows(ws, section: GlobalVarSection) -> None: + """Write header + variable rows, expanding struct members inline. + + Structured variables produce one row per member using dotted names + (``PLC_FEL_STA.Value -> D0.0``); member rows have an empty Type cell. + """ + ws.append(HEADER_ROW) + for cell in ws[1]: + cell.font = _HEADER_FONT + cell.fill = _HEADER_FILL + cell.alignment = Alignment(horizontal="center") + + for var in section.variables: + ws.append( + [ + var.name, + var.address, + var.type_name, + var.comment, + _format_line(var.line_number), + RETAIN_TRUE if var.retain else RETAIN_FALSE, + ] + ) + for member in var.members: + ws.append( + [ + f"{var.name}.{member.name}", + member.address, + member.type_name, + "", + "", + # Members inherit the parent variable's retain flag so + # the Retain column can be used for filtering. + RETAIN_TRUE if var.retain else RETAIN_FALSE, + ] + ) + + +def _write_meta_sheet( + wb: Workbook, + sections: list[GlobalVarSection], + titles: list[str], + metadata: ProjectMetadata | None, +) -> None: + """Write the ``Meta`` sheet: document metadata + sheet mapping.""" + ws = wb.create_sheet(title=META_SHEET_NAME) + ws.append(["Key", "Value"]) + for cell in ws[1]: + cell.font = _HEADER_FONT + cell.fill = _HEADER_FILL + + def put(key: str, value: str) -> None: + ws.append([key, value]) + + if metadata is not None: + for key, value in metadata.file_header.items(): + put(f"fileHeader.{key}", value) + for key, value in metadata.content_header.items(): + put(f"contentHeader.{key}", value) + for key, value in metadata.project_add_data.items(): + put(f"project.{key}", value) + + # Sheet mapping so the reverse direction can restore section names + # exactly, even when sheet titles had to be adjusted. Retain flags are + # carried per row in the Retain column (merged sheets mix them). + for section, title in zip(sections, titles, strict=True): + put(f"sheet.{title}.name", section.name) + if section.modification_date_time: + put( + f"sheet.{title}.modificationDateTime", + section.modification_date_time, + ) + + ws.column_dimensions["A"].width = 42 + ws.column_dimensions["B"].width = 46 + + +def write_workbook( + sections: list[GlobalVarSection], + output_path: Path, + metadata: ProjectMetadata | None = None, +) -> None: + """Create an ``.xlsx`` file, one worksheet per (merged) section name. + + Sections sharing the same name (e.g. a plain and a retain ``Global``) + are merged into a single worksheet; the per-variable retain flag goes + into the ``Retain`` column so the reverse conversion can split them + back apart. Also writes a ``Meta`` sheet with document metadata and the + sheet-to-section mapping (used by the Excel -> XML conversion). + """ + output_path.parent.mkdir(parents=True, exist_ok=True) + + merged = merge_sections(sections) + + wb = Workbook() + # Remove the default sheet created by openpyxl. + wb.remove(wb.active) + + titles = _unique_sheet_names(merged) + for section, title in zip(merged, titles, strict=True): + ws = wb.create_sheet(title=title) + _write_section_rows(ws, section) + + # Column widths: sensible defaults, capped so comments stay readable. + width_by_column = { + 1: 36, # Name + 2: 12, # Address + 3: 18, # Type + 4: 50, # Comment + 5: 8, # Line + 6: 8, # Retain + } + for column, width in width_by_column.items(): + ws.column_dimensions[get_column_letter(column)].width = width + ws.freeze_panes = "A2" + + _write_meta_sheet(wb, merged, titles, metadata) + wb.save(output_path) \ No newline at end of file diff --git a/src/mitsubishi_label_converter/excel_reader.py b/src/mitsubishi_label_converter/excel_reader.py new file mode 100644 index 0000000..97be8a0 --- /dev/null +++ b/src/mitsubishi_label_converter/excel_reader.py @@ -0,0 +1,244 @@ +"""Read an Excel workbook back into parsed label data (reverse conversion).""" + +from __future__ import annotations + +from pathlib import Path + +from openpyxl import load_workbook + +from .excel import HEADER_ROW, META_SHEET_NAME +from .models import GlobalVarSection, ProjectMetadata, StructMember, Variable + + +class WorkbookValidationError(Exception): + """Raised when the workbook does not match the expected layout.""" + + +def _cell_str(value) -> str: + """Normalize a cell value to a string ('' for None/blank).""" + if value is None: + return "" + return str(value).strip() + + +def _cell_line(value) -> int | None: + """Parse the Line column into an int, if present.""" + if value is None or value == "": + return None + if isinstance(value, int): + return value + text = str(value).strip() + if text.isdigit(): + return int(text) + raise WorkbookValidationError(f"invalid line number: {value!r}") + + +def _read_meta_sheet(ws) -> dict[str, str]: + """Read the Meta sheet into a flat key -> value dict.""" + meta: dict[str, str] = {} + for key, value in ws.iter_rows(min_row=2, values_only=True): + key_text = _cell_str(key) + if key_text: + meta[key_text] = _cell_str(value) + return meta + + +def _parse_metadata(meta: dict[str, str]) -> ProjectMetadata: + """Split the flat Meta dict into the ProjectMetadata groups.""" + file_header: dict[str, str] = {} + content_header: dict[str, str] = {} + project_add_data: dict[str, str] = {} + for key, value in meta.items(): + if key.startswith("fileHeader."): + file_header[key[len("fileHeader."):]] = value + elif key.startswith("contentHeader."): + content_header[key[len("contentHeader."):]] = value + elif key.startswith("project."): + project_add_data[key[len("project."):]] = value + return ProjectMetadata( + file_header=file_header, + content_header=content_header, + project_add_data=project_add_data, + ) + + +def _cell_retain(value) -> bool: + """Parse the Retain column into a bool.""" + if value is None or value == "": + return False + if isinstance(value, bool): + return value + return str(value).strip().upper() in {"TRUE", "YES", "1", "Y", "T"} + + +def _section_from_sheet( + ws, + sheet_name: str, + meta: dict[str, str], +) -> list[GlobalVarSection]: + """Build GlobalVarSection(s) from a worksheet. + + A merged sheet (plain + retain rows mixed) is split back into one + section per retain value, in the order the values first appear. The + variables within each section keep the relative order from the sheet. + """ + header = [_cell_str(c) for c in next(ws.iter_rows(max_row=1, values_only=True))] + if header[:2] != HEADER_ROW[:2]: + raise WorkbookValidationError( + f"sheet '{sheet_name}': unexpected header {header!r}" + ) + + name = meta.get(f"sheet.{sheet_name}.name", sheet_name) + modification_date_time = meta.get( + f"sheet.{sheet_name}.modificationDateTime", "" + ) + # Legacy workbooks (5 columns, no Retain column) carry the retain flag + # in the Meta sheet instead. + legacy_retain = ( + meta.get(f"sheet.{sheet_name}.retain", "false").lower() == "true" + ) + + # Gather variables with their retain flag, keeping sheet order. + collected: list[tuple[bool, Variable]] = [] + pending_members: list[StructMember] = [] + pending_retain: bool = legacy_retain + + for row in ws.iter_rows(min_row=2, values_only=True): + var_name = _cell_str(row[0]) if len(row) > 0 else "" + if not var_name: + continue + address = _cell_str(row[1]) if len(row) > 1 else "" + type_name = _cell_str(row[2]) if len(row) > 2 else "" + comment = _cell_str(row[3]) if len(row) > 3 else "" + line_number = _cell_line(row[4]) if len(row) > 4 else None + # New-format sheets have 6 columns; legacy sheets have 5. + retain = ( + _cell_retain(row[5]) + if len(row) > 5 and row[5] is not None + else legacy_retain + ) + + # Struct member rows are recognized by a dotted name and an empty + # Line cell (member types may be populated from an SDT file); they + # extend the previous variable. + if line_number is None and "." in var_name: + # Member row: regroup under the previous structured variable. + if not collected: + raise WorkbookValidationError( + f"sheet '{sheet_name}': member row {var_name!r} " + "before any variable" + ) + pending_members.append( + StructMember( + name=var_name.split(".", 1)[1], + address=address, + type_name=type_name, + ) + ) + continue + + # A new variable row: flush members collected for the previous one. + if pending_members and collected: + prev_retain, prev = collected[-1] + collected[-1] = ( + prev_retain, + Variable( + name=prev.name, + address=prev.address, + type_name=prev.type_name, + comment=prev.comment, + line_number=prev.line_number, + members=tuple(prev.members) + tuple(pending_members), + retain=prev.retain, + ), + ) + pending_members.clear() + + collected.append( + ( + retain, + Variable( + name=var_name, + address=address, + type_name=type_name, + comment=comment, + line_number=line_number, + retain=retain, + ), + ) + ) + + if pending_members and collected: + prev_retain, prev = collected[-1] + collected[-1] = ( + prev_retain, + Variable( + name=prev.name, + address=prev.address, + type_name=prev.type_name, + comment=prev.comment, + line_number=prev.line_number, + members=tuple(prev.members) + tuple(pending_members), + retain=prev.retain, + ), + ) + + if not collected: + return [ + GlobalVarSection( + name=name, + retain=legacy_retain, + variables=[], + modification_date_time=modification_date_time, + ) + ] + + # Split back into one section per retain value, in first-appearance + # order (matching how GX Works3 orders the plain section before the + # retain section). + retain_order: list[bool] = [] + buckets: dict[bool, list[Variable]] = {} + for retain_flag, variable in collected: + if retain_flag not in buckets: + retain_order.append(retain_flag) + buckets[retain_flag] = [] + buckets[retain_flag].append(variable) + + sections: list[GlobalVarSection] = [] + for retain_flag in retain_order: + sections.append( + GlobalVarSection( + name=name, + retain=retain_flag, + variables=buckets[retain_flag], + modification_date_time=modification_date_time, + ) + ) + return sections + + +def read_workbook( + workbook_path: Path, +) -> tuple[list[GlobalVarSection], ProjectMetadata]: + """Read an Excel workbook produced by write_workbook. + + Returns (sections, metadata). Sheets other than the variable sheets and + the Meta sheet are ignored. + """ + wb = load_workbook(workbook_path, read_only=True, data_only=True) + try: + if META_SHEET_NAME in wb.sheetnames: + meta = _read_meta_sheet(wb[META_SHEET_NAME]) + else: + meta = {} + + sections: list[GlobalVarSection] = [] + for sheet_name in wb.sheetnames: + if sheet_name == META_SHEET_NAME: + continue + sections.extend( + _section_from_sheet(wb[sheet_name], sheet_name, meta) + ) + return sections, _parse_metadata(meta) + finally: + wb.close() \ No newline at end of file diff --git a/src/mitsubishi_label_converter/models.py b/src/mitsubishi_label_converter/models.py new file mode 100644 index 0000000..aaf7ca8 --- /dev/null +++ b/src/mitsubishi_label_converter/models.py @@ -0,0 +1,57 @@ +"""Data models for parsed PLCopen label data.""" + +from __future__ import annotations + +from dataclasses import dataclass, field + + +@dataclass(frozen=True) +class StructMember: + """A ```` entry of a ``variableStructDeviceAssignment`` block.""" + + name: str + address: str + type_name: str = "" + """Member type from an SDT file (empty when no SDT data was applied).""" + + +@dataclass(frozen=True) +class Variable: + """A single PLC label/variable extracted from a ```` element.""" + + name: str + address: str + type_name: str + comment: str + line_number: int | None = None + members: tuple[StructMember, ...] = field(default=()) + retain: bool = False + """Retain flag of the globalVars section the variable belongs to.""" + + @property + def is_structured(self) -> bool: + return bool(self.members) + + +@dataclass(frozen=True) +class GlobalVarSection: + """A ```` section: one worksheet in the output workbook.""" + + name: str + retain: bool + variables: list[Variable] + modification_date_time: str = "" + """``modificationDateTime`` from the section's varListProperties addData.""" + + +@dataclass(frozen=True) +class ProjectMetadata: + """Document-level metadata from fileHeader/contentHeader/addData. + + Stored in the ``Meta`` worksheet so an Excel file can be converted back + to XML losslessly. + """ + + file_header: dict[str, str] = field(default_factory=dict) + content_header: dict[str, str] = field(default_factory=dict) + project_add_data: dict[str, str] = field(default_factory=dict) \ No newline at end of file diff --git a/src/mitsubishi_label_converter/parser.py b/src/mitsubishi_label_converter/parser.py new file mode 100644 index 0000000..2c1964b --- /dev/null +++ b/src/mitsubishi_label_converter/parser.py @@ -0,0 +1,206 @@ +"""Parse Mitsubishi GX Works3 PLCopen TC6 XML global label files. + +The source files are typically UTF-16 LE encoded; ``xml.etree`` handles the +byte-order mark transparently, so the file can simply be opened in binary mode. +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from pathlib import Path + +from .models import GlobalVarSection, ProjectMetadata, StructMember, Variable + +PLCOPEN_NS = "http://www.plcopen.org/xml/tc6_0201" +XHTML_NS = "http://www.w3.org/1999/xhtml" + +_COMMENT_DATA_NAME = "http://www.mitsubishielectric.com/xml/variableComments" +_LINE_NUMBER_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/variableLineNumber" +) +_STRUCT_ASSIGNMENT_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/variableStructDeviceAssignment" +) +_PLC_TYPE_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/projectPlcType" +) +_VAR_LIST_PROPERTIES_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/varListProperties" +) + + +def _localname(tag: str) -> str: + """Strip the XML namespace from a tag name.""" + return tag.rsplit("}", 1)[-1] if "}" in tag else tag + + +def _parse_type(type_elem: ET.Element | None) -> str: + """Render the ```` element into a compact type string. + + - ```` -> ``INT`` + - ```` -> ``STRING[12]`` + - ```` -> ``FB_Din`` + """ + if type_elem is None or len(type_elem) == 0: + return "" + child = type_elem[0] + kind = _localname(child.tag) + if kind == "string": + length = child.get("length", "") + return f"STRING[{length}]" if length else "STRING" + if kind == "derived": + return child.get("name", "") + return kind + + +def _find_data(variable_elem: ET.Element, data_name: str) -> ET.Element | None: + """Find an ``addData/data`` element by its ``name`` attribute.""" + for data in variable_elem.findall( + f"./{{{PLCOPEN_NS}}}addData/{{{PLCOPEN_NS}}}data" + ): + if data.get("name") == data_name: + return data + return None + + +def _parse_comment(variable_elem: ET.Element) -> str: + """Extract the variable comment, if any, from the addData block. + + Comments live in ``addData/data[@name='...variableComments']`` and are + wrapped in xhtml ```` elements. + """ + data = _find_data(variable_elem, _COMMENT_DATA_NAME) + if data is None: + return "" + texts: list[str] = [] + for html in data.findall(f".//{{{PLCOPEN_NS}}}comment/{{{XHTML_NS}}}html"): + text = "".join(html.itertext()).strip() + if text: + texts.append(text) + return "\n".join(texts) + + +def _parse_line_number(variable_elem: ET.Element) -> int | None: + """Extract the original source line number, if present.""" + data = _find_data(variable_elem, _LINE_NUMBER_DATA_NAME) + if data is None: + return None + line_elem = data.find(f".//{{{PLCOPEN_NS}}}variableLineNumber") + if line_elem is None: + return None + number = line_elem.get("number", "") + return int(number) if number.isdigit() else None + + +def _parse_struct_members( + variable_elem: ET.Element, +) -> tuple[StructMember, ...]: + """Extract ```` entries, if any.""" + data = _find_data(variable_elem, _STRUCT_ASSIGNMENT_DATA_NAME) + if data is None: + return () + members: list[StructMember] = [] + for member in data.findall(f".//{{{PLCOPEN_NS}}}member"): + members.append( + StructMember( + name=member.get("name", ""), + address=member.get("address", ""), + ) + ) + return tuple(members) + + +def _parse_variable(elem: ET.Element) -> Variable: + return Variable( + name=elem.get("name", ""), + address=elem.get("address", ""), + type_name=_parse_type(elem.find(f"./{{{PLCOPEN_NS}}}type")), + comment=_parse_comment(elem), + line_number=_parse_line_number(elem), + members=_parse_struct_members(elem), + ) + + +def _parse_var_list_modification_time( + global_vars_elem: ET.Element, +) -> str: + """Extract ``modificationDateTime`` from the varListProperties addData.""" + for data in global_vars_elem.findall( + f"./{{{PLCOPEN_NS}}}addData/{{{PLCOPEN_NS}}}data" + ): + if data.get("name") != _VAR_LIST_PROPERTIES_DATA_NAME: + continue + props = data.find(f".//{{{PLCOPEN_NS}}}varListProperties") + if props is not None: + return props.get("modificationDateTime", "") + return "" + + +def parse_global_vars(xml_path: Path) -> list[GlobalVarSection]: + """Parse all ```` sections from a GX Works3 XML export.""" + root = ET.parse(xml_path).getroot() + sections: list[GlobalVarSection] = [] + for global_vars in root.iter(f"{{{PLCOPEN_NS}}}globalVars"): + name = global_vars.get("name", "Global") + retain = global_vars.get("retain", "false").lower() == "true" + variables = [ + Variable( + name=v.name, + address=v.address, + type_name=v.type_name, + comment=v.comment, + line_number=v.line_number, + members=v.members, + retain=retain, + ) + for v in ( + _parse_variable(child) + for child in global_vars.findall( + f"./{{{PLCOPEN_NS}}}variable" + ) + ) + ] + sections.append( + GlobalVarSection( + name=name, + retain=retain, + variables=variables, + modification_date_time=_parse_var_list_modification_time( + global_vars + ), + ) + ) + return sections + + +def parse_project_metadata(xml_path: Path) -> ProjectMetadata: + """Extract document-level metadata needed for a lossless round-trip. + + Captures: + - all attributes of ```` + - all attributes of ```` + - ``projectPlcType`` (PLC type string, from the project addData) + """ + root = ET.parse(xml_path).getroot() + + file_elem = root.find(f"./{{{PLCOPEN_NS}}}fileHeader") + file_header = dict(file_elem.attrib) if file_elem is not None else {} + + content_elem = root.find(f"./{{{PLCOPEN_NS}}}contentHeader") + content_header = dict(content_elem.attrib) if content_elem is not None else {} + + plc_type = "" + for data in root.findall( + f"./{{{PLCOPEN_NS}}}addData/{{{PLCOPEN_NS}}}data" + ): + if data.get("name") != _PLC_TYPE_DATA_NAME: + continue + type_elem = data.find(f".//{{{PLCOPEN_NS}}}projectPlcType") + if type_elem is not None: + # The PLC type is stored in the 'name' attribute (e.g. FX5U). + plc_type = type_elem.get("name", "") + return ProjectMetadata( + file_header=file_header, + content_header=content_header, + project_add_data={"projectPlcType": plc_type}, + ) \ No newline at end of file diff --git a/src/mitsubishi_label_converter/sdt_parser.py b/src/mitsubishi_label_converter/sdt_parser.py new file mode 100644 index 0000000..db4651f --- /dev/null +++ b/src/mitsubishi_label_converter/sdt_parser.py @@ -0,0 +1,141 @@ +"""Parse Mitsubishi SDT (structured data type) XML exports. + +An SDT file is another GX Works3 PLCopen export that contains the project's +user-defined data types in ``types > dataTypes > dataType``. Each ``dataType`` +has a ``baseType > struct`` with one ```` per member, carrying the +member's ```` — primitives, ``string``, ``array``, or nested ``derived`` +references to other data types. +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from pathlib import Path + +from .models import GlobalVarSection, StructMember, Variable + +PLCOPEN_NS = "http://www.plcopen.org/xml/tc6_0201" + +# Standard IEC 61131-3 primitive types; anything else in a member is +# a derived (user-defined) type reference. +_PRIMITIVES = { + "BOOL", "SINT", "INT", "DINT", "LINT", "USINT", "UINT", "UDINT", "ULINT", + "BYTE", "WORD", "DWORD", "LWORD", "REAL", "LREAL", + "TIME", "DATE", "TIME_OF_DAY", "DATE_AND_TIME", "STRING", "WSTRING", +} + + +def _localname(tag: str) -> str: + """Strip the XML namespace from a tag name.""" + return tag.rsplit("}", 1)[-1] if "}" in tag else tag + + +def _type_to_string(type_elem: ET.Element | None, _depth: int = 0) -> str: + """Render a ```` element (as used inside structs) to a string. + + - ```` -> ``BOOL`` + - ```` -> ``STRING[12]`` + - ```` -> ``ST_X`` + - ``... + `` -> ``ARRAY[1..16] OF WORD`` + """ + if type_elem is None or len(type_elem) == 0: + return "" + if _depth > 16: # guard against cyclic type references + return "" + child = type_elem[0] + kind = _localname(child.tag) + if kind == "string": + length = child.get("length", "") + return f"STRING[{length}]" if length else "STRING" + if kind == "derived": + return child.get("name", "") + if kind == "array": + dimensions: list[str] = [] + for dim in child.findall(f"./{{{PLCOPEN_NS}}}dimension"): + lower = dim.get("lower", "0") + upper = dim.get("upper", "?") + dimensions.append(f"{lower}..{upper}") + base = child.find(f"./{{{PLCOPEN_NS}}}baseType") + inner = _type_to_string(base, _depth + 1) + return f"ARRAY[{', '.join(dimensions)}] OF {inner}".rstrip() + return kind + + +def parse_struct_definitions(sdt_path: Path) -> dict[str, dict[str, str]]: + """Parse all struct definitions from an SDT XML export. + + Returns a mapping ``{data_type_name: {member_name: type_string}}`` for + every ``dataType`` whose baseType is a ``struct``. Non-struct data types + (e.g. plain enums/aliases) are skipped. + """ + root = ET.parse(sdt_path).getroot() + definitions: dict[str, dict[str, str]] = {} + for data_type in root.iter(f"{{{PLCOPEN_NS}}}dataType"): + name = data_type.get("name", "") + if not name: + continue + struct = data_type.find( + f"./{{{PLCOPEN_NS}}}baseType/{{{PLCOPEN_NS}}}struct" + ) + if struct is None: + continue + members: dict[str, str] = {} + for member in struct.findall(f"./{{{PLCOPEN_NS}}}variable"): + member_name = member.get("name", "") + if not member_name: + continue + type_string = _type_to_string( + member.find(f"./{{{PLCOPEN_NS}}}type") + ) + members[member_name] = type_string + definitions[name] = members + return definitions + + +def member_type( + definitions: dict[str, dict[str, str]], + struct_name: str, + member_name: str, +) -> str | None: + """Look up a single member's type; None when unknown.""" + members = definitions.get(struct_name) + if members is None: + return None + return members.get(member_name) + + +def apply_struct_types( + sections: list[GlobalVarSection], + definitions: dict[str, dict[str, str]], +) -> None: + """Populate ``StructMember.type_name`` from SDT definitions. + + Mutates the sections in place by replacing structured variables whose + members can be typed. Member types only appear in the SDT export; the + Global.xml device assignments carry names and addresses only. + """ + for section in sections: + for position, var in enumerate(section.variables): + if not var.members: + continue + members = definitions.get(var.type_name) + if members is None: + continue + typed = tuple( + StructMember( + name=m.name, + address=m.address, + type_name=members.get(m.name, m.type_name), + ) + for m in var.members + ) + section.variables[position] = Variable( + name=var.name, + address=var.address, + type_name=var.type_name, + comment=var.comment, + line_number=var.line_number, + members=typed, + retain=var.retain, + ) \ No newline at end of file diff --git a/src/mitsubishi_label_converter/validator.py b/src/mitsubishi_label_converter/validator.py new file mode 100644 index 0000000..33041f2 --- /dev/null +++ b/src/mitsubishi_label_converter/validator.py @@ -0,0 +1,480 @@ +"""Validation checks for parsed global label data. + +Produces a report of suspicious findings across the label tables: + +- duplicate variable names (error) +- exact duplicate device addresses (error) +- a bit address falling inside another variable's word range (warning) +- overlapping word ranges of two variables (warning) +- struct members not present in the SDT definition (warning) +- struct members defined in the SDT but lacking a device address (info) +- derived types that are not defined in the SDT file (info) +- structured variables without any device assignment (info) +- addresses in an unrecognized format (warning) + +Addresses use the Mitsubishi device notation ``DEVICE[.]`` +where the bit suffix is written in hexadecimal (``D6005.0`` .. ``D6005.F``). +""" + +from __future__ import annotations + +import re +from dataclasses import dataclass, field + +from .models import GlobalVarSection, Variable + +SEVERITY_ERROR = "error" +SEVERITY_WARNING = "warning" +SEVERITY_INFO = "info" + +_ADDRESS_RE = re.compile(r"^([A-Za-z]+)(\d+)(?:\.([0-9A-Fa-f]+))?$") + +# Conservative word sizes (16-bit registers) per IEC type, used for +# word-range overlap detection. +_WORD_SIZES = { + "BOOL": 1, "BIT": 1, + "SINT": 1, "INT": 1, "UINT": 1, "USINT": 1, "WORD": 1, "BYTE": 1, + "DINT": 2, "UDINT": 2, "DWORD": 2, "REAL": 2, + "LINT": 4, "ULINT": 4, "LWORD": 4, "LREAL": 4, +} + +_STRING_RE = re.compile(r"^STRING\[(\d+)\]$") +_WSTRING_RE = re.compile(r"^WSTRING\[(\d+)\]$") + + +@dataclass(frozen=True) +class Finding: + """One entry in the validation report.""" + + severity: str + code: str + message: str + section: str = "" + variable: str = "" + + @property + def location(self) -> str: + if self.section and self.variable: + return f"{self.section}/{self.variable}" + return self.section or self.variable or "-" + + +@dataclass(frozen=True) +class _DeviceAddress: + """A parsed device address (word- or bit-level).""" + + device: str + register: int + bit: int | None # None -> word-level address + + @property + def is_bit(self) -> bool: + return self.bit is not None + + +@dataclass +class ValidationReport: + """Aggregated validation findings.""" + + findings: list[Finding] = field(default_factory=list) + + def add(self, finding: Finding) -> None: + self.findings.append(finding) + + @property + def errors(self) -> list[Finding]: + return [f for f in self.findings if f.severity == SEVERITY_ERROR] + + @property + def warnings(self) -> list[Finding]: + return [f for f in self.findings if f.severity == SEVERITY_WARNING] + + @property + def infos(self) -> list[Finding]: + return [f for f in self.findings if f.severity == SEVERITY_INFO] + + @property + def has_errors(self) -> bool: + return bool(self.errors) + + @property + def error_count(self) -> int: + return len(self.errors) + + @property + def warning_count(self) -> int: + return len(self.warnings) + + @property + def info_count(self) -> int: + return len(self.infos) + + def format_text(self) -> str: + """Render the report as human-readable text.""" + lines: list[str] = [] + lines.append( + f"Validation report: " + f"{self.error_count} error(s), " + f"{self.warning_count} warning(s), " + f"{self.info_count} info" + ) + if not self.findings: + lines.append("No findings.") + return "\n".join(lines) + + for severity, label in ( + (SEVERITY_ERROR, "ERRORS"), + (SEVERITY_WARNING, "WARNINGS"), + (SEVERITY_INFO, "INFO"), + ): + group = [f for f in self.findings if f.severity == severity] + if not group: + continue + lines.append("") + lines.append(f"== {label} ({len(group)}) ==") + for finding in group: + lines.append( + f" [{finding.code}] {finding.location}: " + f"{finding.message}" + ) + return "\n".join(lines) + + +def parse_address(text: str) -> _DeviceAddress | None: + """Parse a Mitsubishi device address; None when unparseable. + + Handles ``D450``, ``R1000``, ``SD200`` (word level) and ``D6005.0``, + ``D201.A`` (bit level, hex bit suffix). + """ + match = _ADDRESS_RE.match(text.strip()) + if match is None: + return None + device = match.group(1).upper() + register = int(match.group(2)) + bit_text = match.group(3) + if bit_text is None: + return _DeviceAddress(device=device, register=register, bit=None) + return _DeviceAddress(device=device, register=register, bit=int(bit_text, 16)) + + +def word_size(type_name: str) -> int | None: + """Conservative word size of a type; None when unknown. + + ``STRING[n]`` occupies ``ceil((n + 1) / 2)`` 16-bit registers + (payload + length byte), mirroring GX Works3 device layout. + """ + if not type_name: + return None + if type_name in _WORD_SIZES: + return _WORD_SIZES[type_name] + string_match = _STRING_RE.match(type_name) + if string_match: + return (int(string_match.group(1)) + 1 + 1) // 2 + wstring_match = _WSTRING_RE.match(type_name) + if wstring_match: + return int(wstring_match.group(1)) + 1 + return None # derived/unknown types: no size assumption + + +def _word_span(address: _DeviceAddress, size: int) -> tuple[str, int, int]: + """Half-open [start, end) register span of a word-level address.""" + return (address.device, address.register, address.register + max(size, 1)) + + +def _check_addresses( + sections: list[GlobalVarSection], + report: ValidationReport, +) -> None: + """Duplicate-address and overlap checks across all variables. + + Members of the same structured variable legitimately share word + registers (e.g. ``D0.0`` and ``D0.1``), so entries are grouped by + variable and only cross-variable collisions are reported. + """ + + @dataclass(frozen=True) + class _Entry: + section: str + variable: str + address_text: str + address: _DeviceAddress + size: int | None + + entries: list[_Entry] = [] + + for section in sections: + for var in section.variables: + # The variable's own address (if any). + if var.address: + parsed = parse_address(var.address) + if parsed is None: + report.add( + Finding( + severity=SEVERITY_WARNING, + code="ADDR_FORMAT", + message=f"unrecognized address format: {var.address!r}", + section=section.name, + variable=var.name, + ) + ) + continue + entries.append( + _Entry( + section=section.name, + variable=var.name, + address_text=var.address, + address=parsed, + size=word_size(var.type_name), + ) + ) + # Member addresses of structured variables. + for member in var.members: + if not member.address: + continue + parsed = parse_address(member.address) + if parsed is None: + report.add( + Finding( + severity=SEVERITY_WARNING, + code="ADDR_FORMAT", + message=( + f"unrecognized member address format: " + f"{member.address!r} (member {member.name})" + ), + section=section.name, + variable=var.name, + ) + ) + continue + entries.append( + _Entry( + section=section.name, + variable=var.name, + address_text=member.address, + address=parsed, + size=word_size(member.type_name), + ) + ) + + # Exact duplicate addresses across different variables (error). + by_address: dict[tuple, list[_Entry]] = {} + for entry in entries: + key = (entry.address.device, entry.address.register, entry.address.bit) + by_address.setdefault(key, []).append(entry) + duplicate_keys: set[tuple] = set() + for key, group in sorted( + by_address.items(), + key=lambda item: (item[0][0], item[0][1], item[0][2] is not None), + ): + distinct_vars = {e.variable for e in group} + if len(distinct_vars) > 1: + duplicate_keys.add(key) + names = ", ".join(sorted(distinct_vars)) + report.add( + Finding( + severity=SEVERITY_ERROR, + code="ADDR_DUPLICATE", + message=( + f"address {group[0].address_text} used by " + f"{len(distinct_vars)} variables: {names}" + ), + section=group[0].section, + variable=group[0].variable, + ) + ) + + # Bit-in-word and word-range overlaps across different variables. + word_entries = [e for e in entries if not e.address.is_bit] + bit_entries = [e for e in entries if e.address.is_bit] + + word_spans: list[tuple[str, int, int, _Entry]] = [] + for entry in word_entries: + device, start, end = _word_span( + entry.address, entry.size if entry.size else 1 + ) + word_spans.append((device, start, end, entry)) + + for bit_entry in bit_entries: + bit_key = ( + bit_entry.address.device, + bit_entry.address.register, + bit_entry.address.bit, + ) + if bit_key in duplicate_keys: + continue # already reported as an exact duplicate + for device, start, end, owner in word_spans: + if ( + owner.variable != bit_entry.variable + and device == bit_entry.address.device + and start <= bit_entry.address.register < end + ): + report.add( + Finding( + severity=SEVERITY_WARNING, + code="ADDR_BIT_IN_WORD", + message=( + f"bit {bit_entry.address_text} overlaps word " + f"range of '{owner.variable}' " + f"({owner.address_text})" + ), + section=bit_entry.section, + variable=bit_entry.variable, + ) + ) + + for i, (device, start, end, owner) in enumerate(word_spans): + for device2, start2, end2, other in word_spans[i + 1 :]: + if owner.variable == other.variable: + continue + if device != device2: + continue + owner_key = ( + owner.address.device, + owner.address.register, + owner.address.bit, + ) + other_key = ( + other.address.device, + other.address.register, + other.address.bit, + ) + # skip pairs already reported as exact duplicates + if owner_key in duplicate_keys or other_key in duplicate_keys: + continue + if start < end2 and start2 < end: + report.add( + Finding( + severity=SEVERITY_WARNING, + code="ADDR_OVERLAP", + message=( + f"word range {owner.address_text}.." + f"(+{end - start}) overlaps " + f"'{other.variable}' at {other.address_text}" + ), + section=owner.section, + variable=owner.variable, + ) + ) + + +def _check_names( + sections: list[GlobalVarSection], + report: ValidationReport, +) -> None: + """Duplicate variable-name check across all sections.""" + seen: dict[str, GlobalVarSection] = {} + for section in sections: + for var in section.variables: + if var.name in seen: + report.add( + Finding( + severity=SEVERITY_ERROR, + code="NAME_DUPLICATE", + message=( + f"variable name declared multiple times " + f"(also in '{seen[var.name].name}')" + ), + section=section.name, + variable=var.name, + ) + ) + else: + seen[var.name] = section + + +def _check_structures( + sections: list[GlobalVarSection], + definitions: dict[str, dict[str, str]], + report: ValidationReport, +) -> None: + """Struct-vs-SDT consistency checks (only when SDT data is given).""" + # Aggregated: one finding per unknown derived type, not per variable. + unknown_types: dict[str, list[str]] = {} + for section in sections: + for var in section.variables: + if not var.type_name: + continue + definition = definitions.get(var.type_name) + if definition is None: + # Only report when it is a derived (non-primitive) type and + # actually looks like a type name we should know. + if var.type_name not in _WORD_SIZES and not _STRING_RE.match( + var.type_name + ): + unknown_types.setdefault(var.type_name, []).append(var.name) + continue + if not var.members: + # Structured type with a definition but no device assignment. + report.add( + Finding( + severity=SEVERITY_INFO, + code="STRUCT_UNASSIGNED", + message=( + f"variable of type '{var.type_name}' has no " + f"device address assignment" + ), + section=section.name, + variable=var.name, + ) + ) + continue + assigned = {m.name for m in var.members} + # Members with an address that are not part of the struct. + for member in var.members: + if member.name not in definition: + report.add( + Finding( + severity=SEVERITY_WARNING, + code="MEMBER_UNKNOWN", + message=( + f"member '{member.name}' " + f"({member.address}) is not defined in " + f"struct '{var.type_name}'" + ), + section=section.name, + variable=var.name, + ) + ) + # Struct members that lack a device address. + for member_name in definition: + if member_name not in assigned: + report.add( + Finding( + severity=SEVERITY_INFO, + code="MEMBER_UNASSIGNED", + message=( + f"struct member '{member_name}' of " + f"'{var.type_name}' has no device address" + ), + section=section.name, + variable=var.name, + ) + ) + + for type_name, variable_names in sorted(unknown_types.items()): + count = len(variable_names) + preview = ", ".join(sorted(variable_names)[:3]) + more = f", ... ({count} total)" if count > 3 else "" + report.add( + Finding( + severity=SEVERITY_INFO, + code="TYPE_UNKNOWN", + message=( + f"derived type '{type_name}' is not defined in the " + f"SDT file (used by {count} variable(s): {preview}{more})" + ), + ) + ) + + +def validate( + sections: list[GlobalVarSection], + definitions: dict[str, dict[str, str]] | None = None, +) -> ValidationReport: + """Run all validation checks and return the aggregated report.""" + report = ValidationReport() + _check_names(sections, report) + _check_addresses(sections, report) + if definitions: + _check_structures(sections, definitions, report) + return report \ No newline at end of file diff --git a/src/mitsubishi_label_converter/xml_writer.py b/src/mitsubishi_label_converter/xml_writer.py new file mode 100644 index 0000000..8b37a66 --- /dev/null +++ b/src/mitsubishi_label_converter/xml_writer.py @@ -0,0 +1,239 @@ +"""Generate GX Works3-compatible PLCopen TC6 XML from parsed label data. + +Rebuilds the full document structure: +``project > fileHeader, contentHeader, types, instances, addData`` with the +``configurations/configuration/globalVars`` variable lists. Encoded as UTF-16 +LE with BOM, matching the GX Works3 export format. +""" + +from __future__ import annotations + +import xml.etree.ElementTree as ET +from pathlib import Path +from xml.etree.ElementTree import Element, SubElement, tostring + +from .models import GlobalVarSection, ProjectMetadata, Variable + +PLCOPEN_NS = "http://www.plcopen.org/xml/tc6_0201" +XSI_NS = "http://www.w3.org/2001/XMLSchema-instance" +XSD_NS = "http://www.w3.org/2001/XMLSchema" +XHTML_NS = "http://www.w3.org/1999/xhtml" +XML_NS = "http://www.w3.org/XML/1998/namespace" + +# Serialize with the PLCopen namespace as the DEFAULT namespace (no ns0: +# prefixes), matching the GX Works3 export format. +ET.register_namespace("", PLCOPEN_NS) +ET.register_namespace("html", XHTML_NS) + +ET_Element = Element + +_LINE_NUMBER_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/variableLineNumber" +) +_STRUCT_ASSIGNMENT_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/variableStructDeviceAssignment" +) +_COMMENT_DATA_NAME = "http://www.mitsubishielectric.com/xml/variableComments" +_PLC_TYPE_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/projectPlcType" +) +_VAR_LIST_PROPERTIES_DATA_NAME = ( + "http://www.mitsubishielectric.com/xml/varListProperties" +) + +_DATA_HANDLE = "implementation" + +# Attributes that are set unconditionally on generated elements; everything +# else from the metadata dicts is copied verbatim. +_FILE_HEADER_ORDER = [ + "companyName", + "companyURL", + "productName", + "productVersion", + "creationDateTime", +] +_CONTENT_HEADER_ORDER = ["name", "modificationDateTime"] + + +def qn(tag: str) -> str: + """Build a namespaced tag name.""" + return f"{{{PLCOPEN_NS}}}{tag}" + + +def _parse_type_string(type_name: str) -> tuple[str, str | None]: + """Split a compact type string into (kind, length). + + ``STRING[12]`` -> ``("string", "12")``; ``FB_Din`` -> ``("derived", None)`` + with the name carried in ``kind`` context; primitives pass through. + """ + if type_name.startswith("STRING[") and type_name.endswith("]"): + return "string", type_name[len("STRING[") : -1] + return type_name, None + + +def _add_type_element(variable: ET_Element, type_name: str) -> None: + """Append the ```` child for a compact type string.""" + type_elem = SubElement(variable, qn("type")) + kind, length = _parse_type_string(type_name) + if not kind: + return + if kind == "string": + string_elem = SubElement(type_elem, qn("string")) + if length is not None: + string_elem.set("length", length) + elif _is_derived(kind): + derived = SubElement(type_elem, qn("derived")) + derived.set("name", kind) + else: + SubElement(type_elem, qn(kind)) + + +def _is_derived(kind: str) -> bool: + """Heuristic: derived types are user/FB/struct type names. + + In GX Works3 exports the primitive IEC types are a fixed set; anything + else (FB_Din, ST_Din_STA, Timer, ...) is a derived type reference. + """ + primitives = { + "BOOL", "SINT", "INT", "DINT", "LINT", "USINT", "UINT", "UDINT", + "ULINT", "BYTE", "WORD", "DWORD", "LWORD", "REAL", "LREAL", + "TIME", "DATE", "TIME_OF_DAY", "DATE_AND_TIME", "STRING", "WSTRING", + "ANY", "ANY_BIT", + } + return kind not in primitives + + +def _add_add_data(variable: ET_Element, var: Variable) -> None: + add_data = SubElement(variable, qn("addData")) + if var.line_number is not None: + data = SubElement(add_data, qn("data")) + data.set("name", _LINE_NUMBER_DATA_NAME) + data.set("handleUnknown", _DATA_HANDLE) + line = SubElement(data, qn("variableLineNumber")) + line.set("number", str(var.line_number)) + if var.members: + data = SubElement(add_data, qn("data")) + data.set("name", _STRUCT_ASSIGNMENT_DATA_NAME) + data.set("handleUnknown", _DATA_HANDLE) + assignment = SubElement(data, qn("variableStructDeviceAssignment")) + for member in var.members: + member_elem = SubElement(assignment, qn("member")) + member_elem.set("name", member.name) + member_elem.set("address", member.address) + if var.comment: + data = SubElement(add_data, qn("data")) + data.set("name", _COMMENT_DATA_NAME) + data.set("handleUnknown", _DATA_HANDLE) + comments = SubElement(data, qn("variableComments")) + comment = SubElement(comments, qn("comment")) + comment.set("number", "1") + html = SubElement(comment, f"{{{XHTML_NS}}}html") + html.set(f"{{{XML_NS}}}space", "preserve") + html.text = var.comment + + +def _build_variable(var: Variable) -> ET_Element: + variable = Element(qn("variable")) + variable.set("name", var.name) + if var.address: + variable.set("address", var.address) + _add_type_element(variable, var.type_name) + _add_add_data(variable, var) + return variable + + +def _build_global_vars(section: GlobalVarSection) -> ET_Element: + global_vars = Element(qn("globalVars")) + global_vars.set("name", section.name) + if section.retain: + global_vars.set("retain", "true") + for var in section.variables: + global_vars.append(_build_variable(var)) + add_data = SubElement(global_vars, qn("addData")) + data = SubElement(add_data, qn("data")) + data.set("name", _VAR_LIST_PROPERTIES_DATA_NAME) + data.set("handleUnknown", _DATA_HANDLE) + var_list_properties = SubElement(data, qn("varListProperties")) + if section.modification_date_time: + var_list_properties.set( + "modificationDateTime", section.modification_date_time + ) + return global_vars + + +def _build_project( + sections: list[GlobalVarSection], metadata: ProjectMetadata | None +) -> ET_Element: + # ElementTree declares the plcopen namespace automatically from the + # tags used; the xsi/xsd declarations present in GX Works3 exports are + # not required for re-import. + project = Element(qn("project")) + + file_header = SubElement(project, qn("fileHeader")) + for key in _FILE_HEADER_ORDER: + value = (metadata.file_header.get(key) if metadata else None) or "" + if value: + file_header.set(key, value) + if "companyName" not in file_header.attrib: + file_header.set("companyName", "MITSUBISHI ELECTRIC CORPORATION") + if "productName" not in file_header.attrib: + file_header.set("productName", "GX Works3") + if "creationDateTime" not in file_header.attrib: + file_header.set( + "creationDateTime", + "1970-01-01T00:00:00", + ) + + content_header = SubElement(project, qn("contentHeader")) + for key in _CONTENT_HEADER_ORDER: + value = (metadata.content_header.get(key) if metadata else None) or "" + if value: + content_header.set(key, value) + if "name" not in content_header.attrib: + content_header.set("name", "Global") + + coordinate_info = SubElement(content_header, qn("coordinateInfo")) + for language in ("fbd", "ld", "sfc"): + lang_elem = SubElement(coordinate_info, qn(language)) + scaling = SubElement(lang_elem, qn("scaling")) + scaling.set("x", "1") + scaling.set("y", "1") + + types = SubElement(project, qn("types")) + SubElement(types, qn("dataTypes")) + pous = SubElement(types, qn("pous")) + + instances = SubElement(project, qn("instances")) + configurations = SubElement(instances, qn("configurations")) + configuration = SubElement(configurations, qn("configuration")) + for section in sections: + configuration.append(_build_global_vars(section)) + + project_add_data = SubElement(project, qn("addData")) + plc_data = SubElement(project_add_data, qn("data")) + plc_data.set("name", _PLC_TYPE_DATA_NAME) + plc_data.set("handleUnknown", _DATA_HANDLE) + plc_type = SubElement(plc_data, qn("projectPlcType")) + plc_type_value = ( + (metadata.project_add_data.get("projectPlcType") if metadata else "") + or "" + ) + if plc_type_value: + plc_type.set("name", plc_type_value) + + return project + + +def write_xml( + sections: list[GlobalVarSection], + metadata: ProjectMetadata | None, + output_path: Path, +) -> None: + """Write the PLCopen XML document, UTF-16 LE encoded with BOM.""" + project = _build_project(sections, metadata) + xml_bytes = tostring(project, encoding="unicode") + xml_text = ( + '\r\n' + xml_bytes + "\r\n" + ) + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_bytes(xml_text.encode("utf-16")) \ No newline at end of file diff --git a/tests/test_excel.py b/tests/test_excel.py new file mode 100644 index 0000000..cd297fe --- /dev/null +++ b/tests/test_excel.py @@ -0,0 +1,262 @@ +"""Tests for the Excel writer and reader (including round-trip).""" + +from __future__ import annotations + +from pathlib import Path + +from openpyxl import load_workbook + +from mitsubishi_label_converter.excel import ( + HEADER_ROW, + META_SHEET_NAME, + _unique_sheet_names, + merge_sections, + write_workbook, +) +from mitsubishi_label_converter.excel_reader import read_workbook +from mitsubishi_label_converter.models import ( + GlobalVarSection, + ProjectMetadata, + StructMember, + Variable, +) + + +def make_sections() -> list[GlobalVarSection]: + plain = GlobalVarSection( + name="Global", + retain=False, + variables=[ + Variable("version", "D450", "STRING[12]", ""), + Variable( + "PLC_FEL_STA", + "", + "ST_Din_STA", + "", + line_number=39, + members=( + StructMember("Value", "D0.0"), + StructMember("Status", "D1"), + ), + ), + Variable("PLC_FEL", "", "FB_Din", "PLC error active", line_number=36), + ], + ) + retained = GlobalVarSection( + name="Global", + retain=True, + variables=[ + Variable("PulseTot", "D600", "WORD", "", line_number=31, retain=True), + ], + ) + return [plain, retained] + + +def make_metadata() -> ProjectMetadata: + return ProjectMetadata( + file_header={ + "companyName": "MITSUBISHI ELECTRIC CORPORATION", + "companyURL": "http://www.mitsubishielectric.com", + "productName": "GX Works3", + "productVersion": "1.126G", + "creationDateTime": "2026-10-01T20:59:19+02:00", + }, + content_header={ + "name": "21014JON-B14_20260930", + "modificationDateTime": "2026-10-01T20:57:58+02:00", + }, + project_add_data={"projectPlcType": "FX5U"}, + ) + + +def test_merge_sections() -> None: + """Two Global sections (plain + retain) merge into one section.""" + merged = merge_sections(make_sections()) + assert len(merged) == 1 + assert merged[0].name == "Global" + assert [v.name for v in merged[0].variables] == [ + "version", + "PLC_FEL_STA", + "PLC_FEL", + "PulseTot", + ] + # per-variable retain flags survive the merge + assert [v.retain for v in merged[0].variables] == [False, False, False, True] + + +def test_unique_sheet_names_deduplicates() -> None: + sections = [ + GlobalVarSection("Same", False, []), + GlobalVarSection("Same", True, []), + GlobalVarSection("Same", False, []), + ] + merged = merge_sections(sections) + titles = _unique_sheet_names(merged) + assert len(titles) == len(set(titles)) + assert len(titles) == 1 + + +def test_unique_sheet_names_after_merge() -> None: + """Different names never collide after merging.""" + sections = [ + GlobalVarSection("Alpha", False, []), + GlobalVarSection("Beta", True, []), + ] + titles = _unique_sheet_names(merge_sections(sections)) + assert titles == ["Alpha", "Beta"] + + +def test_write_workbook_creates_expected_sheets(tmp_path: Path) -> None: + """Plain + retain sections merge into ONE worksheet plus Meta.""" + out = tmp_path / "out.xlsx" + write_workbook(make_sections(), out, metadata=make_metadata()) + wb = load_workbook(out) + assert wb.sheetnames == ["Global", META_SHEET_NAME] + + +def test_sheet_content_with_member_rows(tmp_path: Path) -> None: + out = tmp_path / "out.xlsx" + write_workbook(make_sections(), out, metadata=make_metadata()) + wb = load_workbook(out) + ws = wb["Global"] + rows = list(ws.iter_rows(values_only=True)) + assert rows[0] == tuple(HEADER_ROW) + # openpyxl reads cells that were written as "" back as None + assert rows[1] == ("version", "D450", "STRING[12]", None, None, "FALSE") + # structured variable row followed by its member rows + assert rows[2][:3] == ("PLC_FEL_STA", None, "ST_Din_STA") + assert rows[2][5] == "FALSE" + # members inherit the parent's retain flag (for filtering) + assert rows[3] == ( + "PLC_FEL_STA.Value", "D0.0", None, None, None, "FALSE" + ) + assert rows[4] == ( + "PLC_FEL_STA.Status", "D1", None, None, None, "FALSE" + ) + # next variable continues after the members + assert rows[5] == ("PLC_FEL", None, "FB_Din", "PLC error active", 36, "FALSE") + # the retained variable from the second section follows, flagged TRUE + assert rows[6] == ("PulseTot", "D600", "WORD", None, 31, "TRUE") + + +def test_meta_sheet_content(tmp_path: Path) -> None: + out = tmp_path / "out.xlsx" + write_workbook(make_sections(), out, metadata=make_metadata()) + wb = load_workbook(out) + ws = wb[META_SHEET_NAME] + rows = list(ws.iter_rows(values_only=True)) + meta = {k: v for k, v in rows[1:]} + assert meta["fileHeader.productName"] == "GX Works3" + assert meta["contentHeader.name"] == "21014JON-B14_20260930" + assert meta["project.projectPlcType"] == "FX5U" + # retain flags live in the Retain column now, not in Meta + assert meta["sheet.Global.name"] == "Global" + assert "sheet.Global.retain" not in meta + + +def test_round_trip_excel_reader(tmp_path: Path) -> None: + """Write then read: sections and metadata must survive intact. + + The reader must split the merged worksheet back into the original two + sections (plain first, retain second), matching the input. Member rows + carrying the inherited retain flag must still regroup correctly. + """ + out = tmp_path / "out.xlsx" + sections = make_sections() + metadata = make_metadata() + write_workbook(sections, out, metadata=metadata) + + read_sections, read_meta = read_workbook(out) + assert read_sections == sections + assert read_meta == metadata + + +def test_member_rows_carry_retain_for_filtering(tmp_path: Path) -> None: + """Every row — variable or member — must have a non-empty Retain cell + so the Retain column can be used for Excel filtering.""" + out = tmp_path / "out.xlsx" + write_workbook(make_sections(), out, metadata=make_metadata()) + wb = load_workbook(out) + ws = wb["Global"] + for row in ws.iter_rows(min_row=2, values_only=True): + assert row[5] in ("TRUE", "FALSE"), ( + f"row {row[0]!r} has no retain flag" + ) + + +def test_retain_members_of_retained_variable(tmp_path: Path) -> None: + """A structured variable with retain=True flags all its members TRUE.""" + out = tmp_path / "out.xlsx" + retained_struct = GlobalVarSection( + name="Global", + retain=True, + variables=[ + Variable( + "SUML_DO", + "", + "ST_Dout_STA", + "", + line_number=5, + retain=True, + members=( + StructMember("Value", "D100.0"), + StructMember("Status", "D101"), + ), + ), + ], + ) + write_workbook([retained_struct], out) + wb = load_workbook(out) + rows = list(wb["Global"].iter_rows(values_only=True)) + assert rows[1] == ("SUML_DO", None, "ST_Dout_STA", None, 5, "TRUE") + assert rows[2] == ("SUML_DO.Value", "D100.0", None, None, None, "TRUE") + assert rows[3] == ("SUML_DO.Status", "D101", None, None, None, "TRUE") + + +def test_round_trip_regroups_interleaved_retain_rows(tmp_path: Path) -> None: + """Manually edited sheets: retain rows moved to the top must still + regroup into the correct sections.""" + out = tmp_path / "out.xlsx" + write_workbook(make_sections(), out, metadata=make_metadata()) + + # Simulate an edit: move the retained variable to the top of the sheet. + from openpyxl import load_workbook as lwb + + wb = lwb(out) + ws = wb["Global"] + header = [c.value for c in ws[1]] + rows = [r for r in ws.iter_rows(min_row=2, values_only=True)] + pulse_row = next(r for r in rows if r[0] == "PulseTot") + others = [r for r in rows if r[0] != "PulseTot"] + # delete all and rewrite with PulseTot first + ws.delete_rows(1, ws.max_row) + for row in [header] + [pulse_row] + others: + ws.append(list(row)) + wb.save(out) + + read_sections, _ = read_workbook(out) + # First section in the sheet is now the retain one (first-appearance order) + assert len(read_sections) == 2 + assert [v.name for v in read_sections[0].variables] == ["PulseTot"] + assert read_sections[0].retain is True + assert [v.name for v in read_sections[1].variables] == [ + "version", + "PLC_FEL_STA", + "PLC_FEL", + ] + assert read_sections[1].retain is False + + +def test_sheet_name_max_31_chars() -> None: + long_name = "A" * 40 + sections = [GlobalVarSection(long_name, False, [])] + titles = _unique_sheet_names(sections) + assert len(titles[0]) <= 31 + assert titles[0].startswith("A") + + +def test_forbidden_characters_are_replaced() -> None: + sections = [GlobalVarSection("Global/Vars[1]:x", False, [])] + titles = _unique_sheet_names(sections) + for forbidden in "[]:*?/\\": + assert forbidden not in titles[0] \ No newline at end of file diff --git a/tests/test_parser.py b/tests/test_parser.py new file mode 100644 index 0000000..88a1c92 --- /dev/null +++ b/tests/test_parser.py @@ -0,0 +1,175 @@ +"""Tests for the PLCopen XML parser.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from mitsubishi_label_converter.parser import ( + parse_global_vars, + parse_project_metadata, +) + +NS = "http://www.plcopen.org/xml/tc6_0201" + +SAMPLE_XML = f""" + + + + + + + + + + + + + + + + + + + + + + + PLC error active + + + + + + + + + + + + + + + + + + + + + + + + + + +""" + + +@pytest.fixture() +def sample_file(tmp_path: Path) -> Path: + """Write the sample XML as UTF-16 LE, like GX Works3 does.""" + xml_path = tmp_path / "Global.xml" + xml_path.write_bytes(SAMPLE_XML.encode("utf-16")) + return xml_path + + +def test_parses_two_sections(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + assert len(sections) == 2 + assert all(section.name == "Global" for section in sections) + + +def test_retain_flag_is_parsed(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + assert sections[0].retain is False + assert sections[1].retain is True + + +def test_primitive_type(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + first = sections[0].variables[0] + assert first.name == "Always_ON" + assert first.address == "SM400" + assert first.type_name == "BOOL" + assert first.comment == "" + + +def test_string_type_includes_length(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + version = sections[0].variables[1] + assert version.name == "version" + assert version.type_name == "STRING[12]" + + +def test_derived_type_uses_name(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + plc_fel = sections[0].variables[2] + assert plc_fel.address == "" # derived FB variable without device address + assert plc_fel.type_name == "FB_Din" + sta = sections[0].variables[3] + assert sta.type_name == "ST_Din_STA" + assert [(m.name, m.address) for m in sta.members] == [ + ("Value", "D0.0"), + ("Status", "D1"), + ] + + +def test_comment_extraction(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + plc_fel = sections[0].variables[2] + assert plc_fel.comment == "PLC error active" + + +def test_struct_members_are_parsed(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + # PLC_FEL (plain derived FB) has no member assignments + assert sections[0].variables[2].members == () + + +def test_struct_members_from_real_file() -> None: + asset = Path(__file__).parents[1] / "assets" / "Global.xml" + if not asset.is_file(): + pytest.skip("assets/Global.xml not present") + sections = parse_global_vars(asset) + structured = [v for s in sections for v in s.variables if v.members] + assert len(structured) == 189 + sta = next(v for v in structured if v.name == "PLC_FEL_STA") + assert sta.type_name == "ST_Din_STA" + assert sta.line_number == 39 + assert (sta.members[0].name, sta.members[0].address) == ("Value", "D0.0") + + +def test_project_metadata_from_real_file() -> None: + asset = Path(__file__).parents[1] / "assets" / "Global.xml" + if not asset.is_file(): + pytest.skip("assets/Global.xml not present") + meta = parse_project_metadata(asset) + assert meta.file_header["productName"] == "GX Works3" + assert meta.content_header["name"] == "21014JON-B14_20260930" + assert meta.project_add_data["projectPlcType"] == "FX5U" + + +def test_line_number_is_parsed(sample_file: Path) -> None: + sections = parse_global_vars(sample_file) + always_on = sections[0].variables[0] + assert always_on.line_number == 3 + + +def test_real_world_global_xml() -> None: + """End-to-end check against the real asset in the repository.""" + asset = Path(__file__).parents[1] / "assets" / "Global.xml" + if not asset.is_file(): + pytest.skip("assets/Global.xml not present") + sections = parse_global_vars(asset) + assert len(sections) == 2 + # 174/118 raw child elements include one (varListProperties) each + assert [len(s.variables) for s in sections] == [173, 117] + version_row = sections[0].variables[0] + assert version_row.name == "version" + assert version_row.address == "D450" + assert version_row.type_name == "STRING[12]" \ No newline at end of file diff --git a/tests/test_sdt.py b/tests/test_sdt.py new file mode 100644 index 0000000..3ddf235 --- /dev/null +++ b/tests/test_sdt.py @@ -0,0 +1,197 @@ +"""Tests for the SDT (structured data type) parser and type enrichment.""" + +from __future__ import annotations + +from pathlib import Path + +import pytest + +from mitsubishi_label_converter.excel_reader import read_workbook +from mitsubishi_label_converter.excel import write_workbook +from mitsubishi_label_converter.models import ( + GlobalVarSection, + ProjectMetadata, + StructMember, + Variable, +) +from mitsubishi_label_converter.parser import parse_global_vars +from mitsubishi_label_converter.sdt_parser import ( + apply_struct_types, + parse_struct_definitions, +) + +NS = "http://www.plcopen.org/xml/tc6_0201" + +SDT_XML = f""" + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +""" + +GLOBAL_XML = f""" + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + + +""" + + +@pytest.fixture() +def sdt_file(tmp_path: Path) -> Path: + path = tmp_path / "SDT.xml" + path.write_bytes(SDT_XML.encode("utf-16")) + return path + + +@pytest.fixture() +def global_file(tmp_path: Path) -> Path: + path = tmp_path / "Global.xml" + path.write_bytes(GLOBAL_XML.encode("utf-16")) + return path + + +def test_parse_struct_definitions(sdt_file: Path) -> None: + definitions = parse_struct_definitions(sdt_file) + assert definitions["ST_Din_STA"] == { + "Value": "BOOL", + "Status": "WORD", + "Name": "STRING[16]", + "Levels": "ARRAY[1..4] OF INT", + "Sub": "ST_Sub", + } + assert definitions["ST_Sub"] == {"Flag": "BOOL"} + + +def test_apply_struct_types(global_file: Path, sdt_file: Path) -> None: + sections = parse_global_vars(global_file) + definitions = parse_struct_definitions(sdt_file) + apply_struct_types(sections, definitions) + + di4 = sections[0].variables[0] + assert [(m.name, m.type_name) for m in di4.members] == [ + ("Value", "BOOL"), + ("Status", "WORD"), + ("Name", "STRING[16]"), + ("Levels", "ARRAY[1..4] OF INT"), + ("Sub", "ST_Sub"), + ("Unknown", ""), # not part of the struct definition + ] + # FB_Din is not defined in the SDT file: members stay untyped + fb = sections[0].variables[1] + assert [(m.name, m.type_name) for m in fb.members] == [("Value", "")] + + +def test_excel_round_trip_with_member_types( + global_file: Path, sdt_file: Path, tmp_path: Path +) -> None: + """Members with types survive the Excel write/read round-trip.""" + sections = parse_global_vars(global_file) + definitions = parse_struct_definitions(sdt_file) + apply_struct_types(sections, definitions) + + out = tmp_path / "out.xlsx" + write_workbook(sections, out, metadata=ProjectMetadata()) + read_sections, _ = read_workbook(out) + assert read_sections == sections + + +def test_real_files_sdt_enrichment() -> None: + """Enrich the real Global.xml with the real SDT.xml and verify.""" + assets = Path(__file__).parents[1] / "assets" + global_xml = assets / "Global.xml" + sdt_xml = assets / "SDT.xml" + if not global_xml.is_file() or not sdt_xml.is_file(): + pytest.skip("assets not present") + + sections = parse_global_vars(global_xml) + definitions = parse_struct_definitions(sdt_xml) + + # All 31 struct types carrying device assignments are ST_* and defined + # in SDT.xml (FB_* instances have no device-assigned members). + used = {v.type_name for s in sections for v in s.variables if v.members} + assert len(used) == 31 + assert all(name.startswith("ST_") for name in used) + assert used <= set(definitions) + + apply_struct_types(sections, definitions) + typed = sum( + 1 + for s in sections + for v in s.variables + for m in v.members + if m.type_name + ) + total = sum(len(v.members) for s in sections for v in s.variables) + # Every device-assigned member belongs to an ST_* type defined in the + # SDT file, so every member gets a type. + assert typed == total + + +def test_real_sdt_definition_sample() -> None: + assets = Path(__file__).parents[1] / "assets" + sdt_xml = assets / "SDT.xml" + if not sdt_xml.is_file(): + pytest.skip("assets/SDT.xml not present") + definitions = parse_struct_definitions(sdt_xml) + assert len(definitions) == 164 + assert definitions["ST_Din_STA"]["Value"] == "BOOL" + assert definitions["ST_Din_STA"]["Status"] == "WORD" \ No newline at end of file diff --git a/tests/test_validator.py b/tests/test_validator.py new file mode 100644 index 0000000..53901a7 --- /dev/null +++ b/tests/test_validator.py @@ -0,0 +1,347 @@ +"""Tests for the validation report.""" + +from __future__ import annotations + +from mitsubishi_label_converter.models import ( + GlobalVarSection, + StructMember, + Variable, +) +from mitsubishi_label_converter.validator import ( + SEVERITY_ERROR, + SEVERITY_INFO, + SEVERITY_WARNING, + parse_address, + validate, + word_size, +) + + +def var( + name: str, + address: str = "", + type_name: str = "BOOL", + members: tuple[StructMember, ...] = (), + **kwargs, +) -> Variable: + return Variable( + name=name, + address=address, + type_name=type_name, + comment="", + members=members, + **kwargs, + ) + + +def section(variables: list[Variable], name: str = "Global") -> GlobalVarSection: + return GlobalVarSection(name=name, retain=False, variables=variables) + + +def codes(report, severity: str) -> set[str]: + return {f.code for f in report.findings if f.severity == severity} + + +# ---------------------------------------------------------------- parse_address + +def test_parse_address_word_level() -> None: + addr = parse_address("D450") + assert addr is not None + assert (addr.device, addr.register, addr.bit) == ("D", 450, None) + + +def test_parse_address_bit_decimal() -> None: + addr = parse_address("D6005.1") + assert addr is not None + assert (addr.device, addr.register, addr.bit) == ("D", 6005, 1) + + +def test_parse_address_bit_hex() -> None: + addr = parse_address("D201.A") + assert addr is not None + assert (addr.device, addr.register, addr.bit) == ("D", 201, 10) + + +def test_parse_address_special_devices() -> None: + assert parse_address("SM400") is not None + assert parse_address("SD8005") is not None + assert parse_address("R5699.0") is not None + + +def test_parse_address_invalid() -> None: + assert parse_address("") is None + assert parse_address("D") is None + assert parse_address("D45.0.1") is None + assert parse_address("not an address") is None + + +# ---------------------------------------------------------------- word_size + +def test_word_size_primitives() -> None: + assert word_size("BOOL") == 1 + assert word_size("INT") == 1 + assert word_size("WORD") == 1 + assert word_size("DINT") == 2 + assert word_size("REAL") == 2 + assert word_size("LREAL") == 4 + + +def test_word_size_strings() -> None: + assert word_size("STRING[12]") == 7 # (12 + 1 + 1) // 2 + assert word_size("STRING[11]") == 6 + assert word_size("STRING[0]") == 1 + assert word_size("WSTRING[10]") == 11 + + +def test_word_size_unknown() -> None: + assert word_size("ST_Din_STA") is None # derived: no assumption + assert word_size("") is None + + +# ---------------------------------------------------------------- name checks + +def test_duplicate_names_reported_as_error() -> None: + report = validate( + [ + section( + [ + var("Always_ON", "SM400"), + var("Always_ON", "SM401"), + ] + ) + ] + ) + assert "NAME_DUPLICATE" in codes(report, SEVERITY_ERROR) + assert report.has_errors + + +def test_unique_names_clean() -> None: + report = validate([section([var("A", "D1"), var("B", "D2")])]) + assert "NAME_DUPLICATE" not in codes(report, SEVERITY_ERROR) + assert not report.has_errors + + +# ---------------------------------------------------------------- address checks + +def test_duplicate_addresses_reported_as_error() -> None: + report = validate( + [section([var("A", "D100"), var("B", "D100")])] + ) + assert "ADDR_DUPLICATE" in codes(report, SEVERITY_ERROR) + + +def test_same_variable_member_bits_not_flagged() -> None: + """Bits of the same variable sharing a word are normal, not findings.""" + report = validate( + [ + section( + [ + var( + "STA", + type_name="ST_Din_STA", + members=( + StructMember("Value", "D0.0"), + StructMember("AutoMan", "D0.1"), + StructMember("Status", "D1"), + ), + ) + ] + ) + ] + ) + assert "ADDR_DUPLICATE" not in codes(report, SEVERITY_ERROR) + assert "ADDR_BIT_IN_WORD" not in codes(report, SEVERITY_WARNING) + assert "ADDR_OVERLAP" not in codes(report, SEVERITY_WARNING) + + +def test_bit_inside_other_variables_word_reported() -> None: + report = validate( + [ + section( + [ + var("Word", "D100", "WORD"), + var( + "Sta", + type_name="ST_X", + members=(StructMember("Flag", "D100.3"),), + ), + ] + ) + ] + ) + assert "ADDR_BIT_IN_WORD" in codes(report, SEVERITY_WARNING) + + +def test_overlapping_word_ranges_reported() -> None: + report = validate( + [ + section( + [ + var("Str", "D100", "STRING[12]"), # occupies D100..D106 + var("Other", "D104", "WORD"), + ] + ) + ] + ) + assert "ADDR_OVERLAP" in codes(report, SEVERITY_WARNING) + + +def test_different_devices_do_not_overlap() -> None: + report = validate( + [ + section( + [ + var("A", "D100"), + var("B", "R100"), + var("C", "SD200"), + ] + ) + ] + ) + assert "ADDR_OVERLAP" not in codes(report, SEVERITY_WARNING) + assert "ADDR_DUPLICATE" not in codes(report, SEVERITY_ERROR) + + +def test_unparseable_address_reported() -> None: + report = validate([section([var("Weird", "???")])]) + assert "ADDR_FORMAT" in codes(report, SEVERITY_WARNING) + + +# ---------------------------------------------------------------- struct checks + +def test_unknown_member_reported() -> None: + definitions = {"ST_X": {"Flag": "BOOL"}} + report = validate( + [ + section( + [ + var( + "Sta", + type_name="ST_X", + members=(StructMember("Typo", "D0.0"),), + ) + ] + ) + ], + definitions, + ) + assert "MEMBER_UNKNOWN" in codes(report, SEVERITY_WARNING) + + +def test_unassigned_struct_member_reported_as_info() -> None: + definitions = {"ST_X": {"Flag": "BOOL", "Missing": "WORD"}} + report = validate( + [ + section( + [ + var( + "Sta", + type_name="ST_X", + members=(StructMember("Flag", "D0.0"),), + ) + ] + ) + ], + definitions, + ) + assert "MEMBER_UNASSIGNED" in codes(report, SEVERITY_INFO) + + +def test_struct_without_device_assignment_reported() -> None: + definitions = {"ST_X": {"Flag": "BOOL"}} + report = validate( + [section([var("Sta", type_name="ST_X")])], + definitions, + ) + assert "STRUCT_UNASSIGNED" in codes(report, SEVERITY_INFO) + + +def test_unknown_derived_type_aggregated_once() -> None: + """One finding per unknown type, with the variable count — not per var.""" + # a non-empty SDT (one known type) enables the struct checks + definitions = {"ST_Known": {"Flag": "BOOL"}} + report = validate( + [ + section( + [ + var("A", type_name="FB_Din"), + var("B", type_name="FB_Din"), + var("C", type_name="FB_Din"), + ] + ) + ], + definitions, + ) + type_findings = [ + f for f in report.findings if f.code == "TYPE_UNKNOWN" + ] + assert len(type_findings) == 1 + assert "used by 3 variable(s)" in type_findings[0].message + + +def test_without_sdt_no_struct_findings() -> None: + """Without SDT definitions, struct checks are skipped entirely.""" + report = validate( + [ + section( + [ + var( + "Sta", + type_name="ST_X", + members=(StructMember("Flag", "D0.0"),), + ) + ] + ) + ] + ) + assert not report.findings + + +# ---------------------------------------------------------------- report text + +def test_format_text_renders_groups() -> None: + report = validate( + [ + section( + [ + var("A", "D1"), + var("A", "D2"), # duplicate name (different addresses) + var("B", "D1"), # duplicate address of var A + ] + ) + ] + ) + text = report.format_text() + assert "2 error(s)" in text + assert "ERRORS (2)" in text + assert "[NAME_DUPLICATE]" in text + assert "[ADDR_DUPLICATE]" in text + + +def test_clean_report_text() -> None: + report = validate([section([var("A", "D1")])]) + assert "No findings." in report.format_text() + + +def test_real_file_is_clean() -> None: + """The shipped Global.xml + SDT.xml must produce no errors/warnings.""" + from pathlib import Path + + from mitsubishi_label_converter.parser import parse_global_vars + from mitsubishi_label_converter.sdt_parser import parse_struct_definitions + + assets = Path(__file__).parents[1] / "assets" + global_xml = assets / "Global.xml" + sdt_xml = assets / "SDT.xml" + if not global_xml.is_file() or not sdt_xml.is_file(): + import pytest + + pytest.skip("assets not present") + + sections = parse_global_vars(global_xml) + definitions = parse_struct_definitions(sdt_xml) + report = validate(sections, definitions) + assert report.error_count == 0 + assert report.warning_count == 0 + # only the known informational findings remain + assert report.info_count == 13 \ No newline at end of file diff --git a/tests/test_xml_writer.py b/tests/test_xml_writer.py new file mode 100644 index 0000000..f27cef6 --- /dev/null +++ b/tests/test_xml_writer.py @@ -0,0 +1,154 @@ +"""Tests for the XML writer (Excel data -> GX Works3 XML).""" + +from __future__ import annotations + +from pathlib import Path + +from mitsubishi_label_converter.models import ( + GlobalVarSection, + ProjectMetadata, + StructMember, + Variable, +) +from mitsubishi_label_converter.parser import ( + parse_global_vars, + parse_project_metadata, +) +from mitsubishi_label_converter.xml_writer import write_xml + + +def make_sections() -> list[GlobalVarSection]: + return [ + GlobalVarSection( + name="Global", + retain=False, + variables=[ + Variable("Always_ON", "SM400", "BOOL", "", line_number=3), + Variable("version", "D450", "STRING[12]", ""), + Variable( + "PLC_FEL_STA", + "", + "ST_Din_STA", + "", + line_number=39, + members=( + StructMember("Value", "D0.0"), + StructMember("Status", "D1"), + ), + ), + Variable( + "Switch_status", + "SD200", + "INT", + "CPU switch status (0: RUN, 1: STOP)", + line_number=14, + ), + ], + ), + GlobalVarSection( + name="Global", + retain=True, + variables=[ + Variable( + "PulseTot", "D600", "WORD", "", line_number=31, retain=True + ), + ], + ), + ] + + +def make_metadata() -> ProjectMetadata: + return ProjectMetadata( + file_header={ + "companyName": "MITSUBISHI ELECTRIC CORPORATION", + "companyURL": "http://www.mitsubishielectric.com", + "productName": "GX Works3", + "productVersion": "1.126G", + "creationDateTime": "2026-10-01T20:59:19+02:00", + }, + content_header={ + "name": "21014JON-B14_20260930", + "modificationDateTime": "2026-10-01T20:57:58+02:00", + }, + project_add_data={"projectPlcType": "FX5U"}, + ) + + +def test_full_round_trip(tmp_path: Path) -> None: + """XML -> sections -> XML: the parsed data must be identical.""" + out = tmp_path / "regenerated.xml" + sections = make_sections() + write_xml(sections, make_metadata(), out) + + # The regenerated file must parse back to the same sections. + reparsed = parse_global_vars(out) + assert reparsed == sections + + meta = parse_project_metadata(out) + assert meta.file_header == make_metadata().file_header + assert meta.content_header == make_metadata().content_header + assert meta.project_add_data == make_metadata().project_add_data + + +def test_output_is_utf16_le(tmp_path: Path) -> None: + out = tmp_path / "regenerated.xml" + write_xml(make_sections(), make_metadata(), out) + raw = out.read_bytes() + # UTF-16 LE BOM + assert raw[:2] == b"\xff\xfe" + # XML declaration must declare utf-16 + assert raw.decode("utf-16").startswith('') + + +def test_retain_flag_written(tmp_path: Path) -> None: + out = tmp_path / "regenerated.xml" + write_xml(make_sections(), make_metadata(), out) + text = out.read_text(encoding="utf-16") + assert '' in text + # the plain section: retain attribute must NOT be present + assert '' in text + + +def test_member_addresses_written(tmp_path: Path) -> None: + out = tmp_path / "regenerated.xml" + write_xml(make_sections(), make_metadata(), out) + text = out.read_text(encoding="utf-16") + assert 'member name="Value" address="D0.0"' in text + assert 'member name="Status" address="D1"' in text + + +def test_comment_written(tmp_path: Path) -> None: + out = tmp_path / "regenerated.xml" + write_xml(make_sections(), make_metadata(), out) + reparsed = parse_global_vars(out) + switch_status = reparsed[0].variables[3] + assert switch_status.comment == "CPU switch status (0: RUN, 1: STOP)" + + +def test_line_numbers_written(tmp_path: Path) -> None: + out = tmp_path / "regenerated.xml" + write_xml(make_sections(), make_metadata(), out) + reparsed = parse_global_vars(out) + assert [v.line_number for v in reparsed[0].variables] == [3, None, 39, 14] + + +def test_real_file_full_round_trip(tmp_path: Path) -> None: + """The real asset: XML -> sections -> XML -> sections must be stable.""" + asset = Path(__file__).parents[1] / "assets" / "Global.xml" + if not asset.is_file(): + import pytest + + pytest.skip("assets/Global.xml not present") + + original_sections = parse_global_vars(asset) + original_meta = parse_project_metadata(asset) + + out = tmp_path / "regenerated.xml" + write_xml(original_sections, original_meta, out) + + reparsed_sections = parse_global_vars(out) + reparsed_meta = parse_project_metadata(out) + assert reparsed_sections == original_sections + assert reparsed_meta.file_header == original_meta.file_header + assert reparsed_meta.content_header == original_meta.content_header + assert reparsed_meta.project_add_data == original_meta.project_add_data \ No newline at end of file