🎉 Initial commit
This commit is contained in:
Jonas Bengtsson
@@ -0,0 +1,9 @@
|
||||
.venv/
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*.egg-info/
|
||||
build/
|
||||
dist/
|
||||
.pytest_cache/
|
||||
output/
|
||||
*.xlsx
|
||||
@@ -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 `<globalVars>` 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 `<variable>` 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 (`<INT/>`, `<BOOL/>`, `<REAL/>`, `<WORD/>`, ...) are written as-is.
|
||||
- `<string length="12"/>` is written as `STRING[12]`.
|
||||
- Structured/derived types (`<derived name="FB_Din"/>`) 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 `<derived>` 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 `<member>` 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 <glob>` 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.
|
||||
Binary file not shown.
Binary file not shown.
Binary file not shown.
@@ -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"]
|
||||
@@ -0,0 +1,3 @@
|
||||
"""Convert Mitsubishi GX Works3 global label XML files to Excel."""
|
||||
|
||||
__version__ = "0.1.0"
|
||||
@@ -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 <globalVars> 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 <globalVars> 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())
|
||||
@@ -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)
|
||||
@@ -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()
|
||||
@@ -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 ``<member>`` 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 ``<variable>`` 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 ``<globalVars>`` 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)
|
||||
@@ -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 ``<type>`` element into a compact type string.
|
||||
|
||||
- ``<INT/>`` -> ``INT``
|
||||
- ``<string length="12"/>`` -> ``STRING[12]``
|
||||
- ``<derived name="FB_Din"/>`` -> ``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 ``<html xml:space="preserve">`` 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 ``<member name="..." address="..."/>`` 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 ``<globalVars>`` 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 ``<fileHeader>``
|
||||
- all attributes of ``<contentHeader>``
|
||||
- ``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},
|
||||
)
|
||||
@@ -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 ``<variable>`` per member, carrying the
|
||||
member's ``<type>`` — 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 <type> 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 ``<type>`` element (as used inside structs) to a string.
|
||||
|
||||
- ``<BOOL/>`` -> ``BOOL``
|
||||
- ``<string length="12"/>`` -> ``STRING[12]``
|
||||
- ``<derived name="ST_X"/>`` -> ``ST_X``
|
||||
- ``<array><dimension lower="1" upper="16"/><baseType>...</baseType>
|
||||
</array>`` -> ``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,
|
||||
)
|
||||
@@ -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<register>[.<bit>]``
|
||||
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
|
||||
@@ -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 ``<type>`` 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 = (
|
||||
'<?xml version="1.0" encoding="utf-16"?>\r\n' + xml_bytes + "\r\n"
|
||||
)
|
||||
output_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
output_path.write_bytes(xml_text.encode("utf-16"))
|
||||
@@ -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]
|
||||
@@ -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"""<?xml version="1.0" encoding="utf-16"?>
|
||||
<project xmlns="{NS}">
|
||||
<instances>
|
||||
<configurations>
|
||||
<configuration name="">
|
||||
<globalVars name="Global">
|
||||
<variable name="Always_ON" address="SM400">
|
||||
<type><BOOL/></type>
|
||||
<addData>
|
||||
<data name="http://www.mitsubishielectric.com/xml/variableLineNumber"
|
||||
handleUnknown="implementation">
|
||||
<variableLineNumber number="3"/>
|
||||
</data>
|
||||
</addData>
|
||||
</variable>
|
||||
<variable name="version" address="D450">
|
||||
<type><string length="12"/></type>
|
||||
</variable>
|
||||
<variable name="PLC_FEL">
|
||||
<type><derived name="FB_Din"/></type>
|
||||
<addData>
|
||||
<data name="http://www.mitsubishielectric.com/xml/variableComments"
|
||||
handleUnknown="implementation">
|
||||
<variableComments>
|
||||
<comment number="1">
|
||||
<html xml:space="preserve"
|
||||
xmlns="http://www.w3.org/1999/xhtml">PLC error active</html>
|
||||
</comment>
|
||||
</variableComments>
|
||||
</data>
|
||||
</addData>
|
||||
</variable>
|
||||
<variable name="PLC_FEL_STA">
|
||||
<type><derived name="ST_Din_STA"/></type>
|
||||
<addData>
|
||||
<data name="http://www.mitsubishielectric.com/xml/variableStructDeviceAssignment"
|
||||
handleUnknown="implementation">
|
||||
<variableStructDeviceAssignment>
|
||||
<member name="Value" address="D0.0"/>
|
||||
<member name="Status" address="D1"/>
|
||||
</variableStructDeviceAssignment>
|
||||
</data>
|
||||
</addData>
|
||||
</variable>
|
||||
</globalVars>
|
||||
<globalVars name="Global" retain="true">
|
||||
<variable name="PulseTot" address="D600">
|
||||
<type><WORD/></type>
|
||||
</variable>
|
||||
</globalVars>
|
||||
</configuration>
|
||||
</configurations>
|
||||
</instances>
|
||||
</project>
|
||||
"""
|
||||
|
||||
|
||||
@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 <addData> (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]"
|
||||
@@ -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"""<?xml version="1.0" encoding="utf-16"?>
|
||||
<project xmlns="{NS}">
|
||||
<types>
|
||||
<dataTypes>
|
||||
<dataType name="ST_Din_STA">
|
||||
<baseType>
|
||||
<struct>
|
||||
<variable name="Value"><type><BOOL/></type></variable>
|
||||
<variable name="Status"><type><WORD/></type></variable>
|
||||
<variable name="Name">
|
||||
<type><string length="16"/></type>
|
||||
</variable>
|
||||
<variable name="Levels">
|
||||
<type><array><dimension lower="1" upper="4"/>
|
||||
<baseType><INT/></baseType></array></type>
|
||||
</variable>
|
||||
<variable name="Sub"><type><derived name="ST_Sub"/></type></variable>
|
||||
</struct>
|
||||
</baseType>
|
||||
</dataType>
|
||||
<dataType name="ST_Sub">
|
||||
<baseType>
|
||||
<struct>
|
||||
<variable name="Flag"><type><BOOL/></type></variable>
|
||||
</struct>
|
||||
</baseType>
|
||||
</dataType>
|
||||
</dataTypes>
|
||||
</types>
|
||||
</project>
|
||||
"""
|
||||
|
||||
GLOBAL_XML = f"""<?xml version="1.0" encoding="utf-16"?>
|
||||
<project xmlns="{NS}">
|
||||
<instances>
|
||||
<configurations>
|
||||
<configuration name="">
|
||||
<globalVars name="Global">
|
||||
<variable name="DI4">
|
||||
<type><derived name="ST_Din_STA"/></type>
|
||||
<addData>
|
||||
<data name="http://www.mitsubishielectric.com/xml/variableStructDeviceAssignment"
|
||||
handleUnknown="implementation">
|
||||
<variableStructDeviceAssignment>
|
||||
<member name="Value" address="D0.0"/>
|
||||
<member name="Status" address="D1"/>
|
||||
<member name="Name" address="D10"/>
|
||||
<member name="Levels" address="D20"/>
|
||||
<member name="Sub" address="D30"/>
|
||||
<member name="Unknown" address="D40"/>
|
||||
</variableStructDeviceAssignment>
|
||||
</data>
|
||||
</addData>
|
||||
</variable>
|
||||
<variable name="FB_Instance">
|
||||
<type><derived name="FB_Din"/></type>
|
||||
<addData>
|
||||
<data name="http://www.mitsubishielectric.com/xml/variableStructDeviceAssignment"
|
||||
handleUnknown="implementation">
|
||||
<variableStructDeviceAssignment>
|
||||
<member name="Value" address="R0.0"/>
|
||||
</variableStructDeviceAssignment>
|
||||
</data>
|
||||
</addData>
|
||||
</variable>
|
||||
</globalVars>
|
||||
</configuration>
|
||||
</configurations>
|
||||
</instances>
|
||||
</project>
|
||||
"""
|
||||
|
||||
|
||||
@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"
|
||||
@@ -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
|
||||
@@ -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('<?xml version="1.0" encoding="utf-16"?>')
|
||||
|
||||
|
||||
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 '<globalVars name="Global" retain="true">' in text
|
||||
# the plain section: retain attribute must NOT be present
|
||||
assert '<globalVars name="Global">' 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
|
||||
Reference in New Issue
Block a user