caexfer 0.1.1

#!/usr/bin/env python3
"""Exercise supported CLI conversion routes using files from several formats.

Build ``target/debug/caexfer`` before running, or pass ``--binary`` with another
executable path. ``--op2-check`` requires pyNastran as a test-only dependency
and verifies Rust-written OP2 with an independent reader. All generated files
live in a temporary directory. Route counts, fields, IDs, assumptions, and
refusal behavior are checked explicitly.
"""
from __future__ import annotations
import argparse
import contextlib
import io
import json
from pathlib import Path
import subprocess
import sys
import tempfile
import xml.etree.ElementTree as ET

ROOT = Path(__file__).resolve().parents[1]
DEFAULT_BINARY = ROOT / 'target' / 'debug' / ('caexfer.exe' if sys.platform == 'win32' else 'caexfer')


def main() -> None:
    """Build source fixtures, check each writable route, and report the count."""
    parser = argparse.ArgumentParser()
    parser.add_argument('--op2-check', action='store_true', help='exercise native OP2 routes')
    parser.add_argument('--binary', type=Path, help='path to an existing caexfer binary')
    options = parser.parse_args()
    binary = (options.binary or DEFAULT_BINARY).resolve()
    if not binary.is_file():
        raise SystemExit(f'No executable at {binary}; build caexfer or pass --binary')

    def independent_op2(path: Path, *, assumed_zero: bool = False,
                        expected_rows: dict[int, list[float]] | None = None) -> None:
        """Check OP2 table identity, components, and values with pyNastran."""
        from pyNastran.op2.op2 import read_op2
        with contextlib.redirect_stdout(io.StringIO()):
            external = read_op2(str(path), build_dataframe=False, debug=False,
                                include_results=['displacements'])
        assert list(external.displacements) == [1]
        result = external.displacements[1]
        assert result.data.shape[0] == 1 and result.data.shape[2] == 6
        assert all(int(kind) == 1 for kind in result.node_gridtype[:, 1])
        if expected_rows is not None:
            actual_rows = {int(node): row.tolist() for node, row in
                           zip(result.node_gridtype[:, 0], result.data[0])}
            assert actual_rows.keys() == expected_rows.keys()
            for node, expected in expected_rows.items():
                assert all(abs(actual - wanted) < 1e-6 for actual, wanted in
                           zip(actual_rows[node], expected))
        if assumed_zero:
            assert 'CAEXFER ASSUMED ZERO DISPLACEMENT' in result.title
            assert (result.data == 0).all()

    def qualify_solver_plate(folder: Path) -> None:
        """Compare a solver-produced BDF/OP2 pair against independent readers."""
        from pyNastran.bdf.bdf import read_bdf
        from pyNastran.op2.op2 import read_op2

        source_bdf = ROOT / 'tests/fixtures/msc-plate.bdf'
        source_op2 = ROOT / 'tests/fixtures/msc-plate.op2'
        with contextlib.redirect_stdout(io.StringIO()):
            mesh = read_bdf(str(source_bdf), xref=False)
            results = read_op2(str(source_op2), build_dataframe=False, debug=False,
                               include_results=['displacements'])
        assert list(results.displacements) == [1]
        displacement = results.displacements[1]
        result_rows = {int(node): row for node, row in
                       zip(displacement.node_gridtype[:, 0], displacement.data[0])}

        target = folder / 'msc-plate.vtu'
        run('convert', source_op2, target, '--mesh', source_bdf, '--accept-all')
        piece = ET.parse(target).find('./UnstructuredGrid/Piece')
        assert piece is not None

        def array(xpath: str, cast: type = int) -> list:
            item = piece.find(xpath)
            assert item is not None
            return [cast(value) for value in (item.text or '').split()]

        node_ids = array("./PointData/DataArray[@Name='nastran_node_id']")
        cell_ids = array("./CellData/DataArray[@Name='nastran_element_id']")
        positions = array('./Points/DataArray', float)
        connectivity = array("./Cells/DataArray[@Name='connectivity']")
        offsets = array("./Cells/DataArray[@Name='offsets']")
        kinds = array("./Cells/DataArray[@Name='types']")
        values = array("./PointData/DataArray[@Name='DISPLACEMENT_SUBCASE_1']", float)

        assert node_ids == sorted(mesh.nodes) == sorted(result_rows)
        assert cell_ids == sorted(mesh.elements)
        assert len(positions) == 3 * len(node_ids)
        assert len(values) == 6 * len(node_ids)
        for index, node in enumerate(node_ids):
            expected_position = mesh.nodes[node].get_position()
            assert all(abs(actual - expected) < 1e-12 for actual, expected in
                       zip(positions[3 * index:3 * index + 3], expected_position))
            assert all(abs(actual - expected) < 1e-7 for actual, expected in
                       zip(values[6 * index:6 * index + 6], result_rows[node]))
        start = 0
        for index, element_id in enumerate(cell_ids):
            element = mesh.elements[element_id]
            assert element.type == 'CQUAD4' and kinds[index] == 9
            assert [node_ids[point] for point in connectivity[start:offsets[index]]] == element.nodes
            start = offsets[index]
        assert start == len(connectivity)

    def run(*args: object, code: int = 0) -> dict:
        """Run one JSON CLI command and require the expected process status."""
        result = subprocess.run([str(binary), *map(str, args), '--json'],
                                cwd=ROOT, capture_output=True, text=True)
        assert result.returncode == code, (args, result.returncode, result.stdout, result.stderr)
        return json.loads(result.stdout)

    with tempfile.TemporaryDirectory(prefix='caexfer-matrix-') as folder:
        folder = Path(folder)
        # Generate additional source dialects from a stable BDF fixture. Each
        # later route uses a real file rather than an in-memory Rust round trip.
        sources = {'bdf': ROOT / 'examples/plate.bdf',
                   'frd': ROOT / 'tests/fixtures/linear-results.frd'}
        for extension in ('vtu', 'vtk', 'msh', 'inp'):
            path = folder / f'source.{extension}'
            run('convert', sources['bdf'], path, '--accept-all')
            sources[extension] = path
        msh22 = folder / 'source22.msh'
        run('convert', sources['bdf'], msh22, '--msh-version', '2.2', '--accept-all')
        sources['msh22'] = msh22
        sources['pch'] = ROOT / 'tests/fixtures/pch-multiple.pch'
        if options.op2_check:
            qualify_solver_plate(folder)
            sources['op2'] = ROOT / 'tests/fixtures/solid_bending.op2'
        routes = 0
        for source_format, source in sources.items():
            # Result files need matching mesh IDs and sometimes subcase
            # selection; ordinary mesh files require no companion options.
            extra = []
            if source_format == 'op2':
                extra = ['--mesh', ROOT / 'tests/fixtures/solid_bending.bdf']
            elif source_format == 'pch':
                extra = ['--mesh', ROOT / 'tests/fixtures/pch-companion.bdf', '--subcase', '1']
            for target_format in ('bdf', 'vtu', 'vtk', 'msh', 'inp'):
                # Check basic identity and field counts for every route.
                target = folder / f'{source_format}-to-{target_format}.{target_format}'
                report = run('convert', source, target, '--accept-all', *extra)
                assert target.is_file() and report['points'] > 0 and report['cells'] > 0
                expected = (72, 186) if source_format == 'op2' else ((3, 1) if source_format == 'frd' else (2, 1) if source_format == 'pch' else (4, 1))
                assert (report['points'], report['cells']) == expected, (source_format, target_format, report)
                if target_format == 'bdf':
                    assert run('validate', target)['passed']
                else:
                    validation = run('validate', target)
                    assert (validation['points'], validation['cells']) == expected
                    expected_fields = 2 if source_format == 'frd' else (1 if source_format in ('op2', 'pch') else 0)
                    if target_format == 'inp':
                        expected_fields = 0
                    assert validation['fields'] == expected_fields, (source_format, target_format, validation)
                    if target_format == 'vtu' and source_format == 'frd':
                        piece = ET.parse(target).find('./UnstructuredGrid/Piece')
                        assert piece is not None
                        disp = piece.find("./PointData/DataArray[@Name='DISP']")
                        stress = piece.find("./PointData/DataArray[@Name='STRESS']")
                        assert disp is not None and stress is not None
                        assert [float(x) for x in disp.text.split()] == [0., 0., 0., 0.1, 0., 0., 0., 0.2, 0.]
                        assert [float(x) for x in stress.text.split()] == [1., 2., 3., 4., 5., 6., 2., 3., 4., 5., 6., 7., 3., 4., 5., 6., 7., 8.]
                    if target_format == 'vtu' and source_format == 'op2':
                        piece = ET.parse(target).find('./UnstructuredGrid/Piece')
                        assert piece is not None
                        field = piece.find("./PointData/DataArray[@Name='DISPLACEMENT_SUBCASE_1']")
                        assert field is not None
                        first = [float(x) for x in field.text.split()[:6]]
                        assert abs(first[0] - 0.007644693832844496) < 1e-9
                        assert first[3:] == [0., 0., 0.]
                routes += 1
            dialect = folder / f'{source_format}-to-msh22.msh'
            # MSH 2.2 is an explicit writer dialect of the same format family.
            report = run('convert', source, dialect, '--msh-version', '2.2',
                         '--accept-all', *extra)
            assert dialect.read_text().startswith('$MeshFormat\n2.2 0 8\n')
            assert (report['points'], report['cells']) == expected
            assert run('validate', dialect)['fields'] == (2 if source_format == 'frd' else
                                                         1 if source_format in ('op2', 'pch') else 0)
            routes += 1
            target = folder / f'{source_format}-to-frd.frd'
            report = run('convert', source, target, '--accept-all', *extra)
            assert target.is_file() and report['points'] == expected[0]
            validation = run('validate', target)
            assert (validation['points'], validation['cells']) == expected
            assert validation['fields'] == (2 if source_format == 'frd' else
                                            1 if source_format in ('op2', 'pch') else 0)
            routes += 1

            if source_format == 'frd':
                # Rotations are unknown in a three-component FRD displacement;
                # unattended OP2 output names the separate infill acceptance.
                result = folder / 'frd-to-op2.op2'
                refused = run('convert', source, result, '--accept-omissions', code=2)
                assert refused['error']['code'] == 'E_USAGE' and not result.exists()
                assert '--accept-zero-rotations' in refused['error']['message']

            if options.op2_check and source_format in ('frd', 'op2'):
                # Verify Rust OP2 output independently, then ingest it again
                # through the public CLI route with its companion mesh.
                result = folder / f'{source_format}-to-op2.op2'
                if source_format == 'frd':
                    mesh = folder / 'frd-for-op2.bdf'
                    report = run('convert', source, result, '--accept-all',
                                 '--accept-zero-rotations',
                                 '--mesh-out', mesh)
                    assert report['mesh_output']['path'] == str(mesh)
                    assert report['mesh_output']['format'] == 'bdf'
                    assert any('excluded from companion mesh' in item['detail']
                               for item in report['mesh_output']['omissions'])
                else:
                    report = run('convert', source, result, '--accept-all', *extra)
                assert result.is_file() and report['fields'] >= 1
                independent_op2(result, expected_rows={
                    1: [0., 0., 0., 0., 0., 0.],
                    2: [0.1, 0., 0., 0., 0., 0.],
                    3: [0., 0.2, 0., 0., 0., 0.],
                } if source_format == 'frd' else None)
                mesh = (ROOT / 'tests/fixtures/solid_bending.bdf' if source_format == 'op2'
                        else folder / 'frd-for-op2.bdf')
                if source_format == 'frd':
                    assert mesh.is_file() and run('validate', mesh)['passed']
                    assert any('typed float 0.0' in item['detail'] for item in report['omissions'])
                reread = folder / f'{source_format}-op2-reread.vtu'
                run('convert', result, reread, '--mesh', mesh, '--accept-all')
                piece = ET.parse(reread).find('./UnstructuredGrid/Piece')
                field = piece.find("./PointData/DataArray[@Name='DISPLACEMENT_SUBCASE_1']")
                assert field is not None
                first = [float(x) for x in field.text.split()[:6]]
                assert first[3:] == [0., 0., 0.]
                routes += 1
            elif source_format in ('bdf', 'inp', 'vtu', 'msh'):
                # Mesh-only BDF/INP can produce only explicitly accepted
                # synthetic results. Other mesh carriers still fail.
                target = folder / f'{source_format}-to-op2.op2'
                if source_format in ('bdf', 'inp'):
                    result = run('convert', source, target, '--accept-omissions', code=2)
                    assert result['error']['code'] == 'E_USAGE' and not target.exists()
                    assert '--accept-synthetic-zero' in result['error']['message']
                else:
                    result = run('convert', source, target, '--accept-all', code=1)
                    assert result['error']['code'] == 'E_OP2' and not target.exists()
                if options.op2_check and source_format in ('bdf', 'inp'):
                    report = run('convert', source, target, '--accept-all',
                                 '--accept-synthetic-zero')
                    assert target.is_file()
                    independent_op2(target, assumed_zero=True)
                    assert any('SYNTHETIC ASSUMPTION' in item['detail']
                               for item in report['omissions'])
                    mesh = (source if source_format == 'bdf'
                            else folder / 'inp-to-bdf.bdf')
                    reread = folder / f'{source_format}-zero-reread.vtu'
                    loaded = run('convert', target, reread, '--mesh', mesh, '--accept-all')
                    assert any('synthetic all-zero' in item['detail']
                               for item in loaded['omissions'])
                    piece = ET.parse(reread).find('./UnstructuredGrid/Piece')
                    field = piece.find("./PointData/DataArray[@Name='DISPLACEMENT_SUBCASE_1']")
                    assert field is not None and all(float(x) == 0.0 for x in field.text.split())
                    if source_format == 'bdf':
                        rewritten = folder / 'synthetic-op2-rewritten.op2'
                        run('convert', target, rewritten, '--mesh', mesh, '--accept-all')
                        reread_report = run('validate', rewritten, '--mesh', mesh)
                        assert any('synthetic all-zero' in item for item in reread_report['omissions'])
                    routes += 1
        if options.op2_check:
            # Exercise PCH transfer and the basic-frame/companion rules with
            # independently supplied OP2 results.
            pch_mesh = ROOT / 'tests/fixtures/pch-companion.bdf'
            pch_op2 = folder / 'pch-to-op2.op2'
            run('convert', sources['pch'], pch_op2, '--mesh', pch_mesh,
                '--subcase', '1', '--accept-all')
            independent_op2(pch_op2, expected_rows={
                10: [1., 2., 3., 4., 5., 6.],
                20: [7., 8., 9., 10., 11., 12.],
            })
            assert run('validate', pch_op2, '--mesh', pch_mesh)['fields'] == 1
            routes += 1
            op2_source = sources['op2']
            op2_mesh = ROOT / 'tests/fixtures/solid_bending.bdf'
            for extension in ('vtu', 'msh', 'inp', 'frd'):
                companion = folder / f'op2-companion.{extension}'
                run('convert', op2_mesh, companion, '--accept-all')
                destination = folder / f'op2-with-{extension}-mesh.vtu'
                refused = run('convert', op2_source, destination, '--mesh', companion,
                              '--accept-omissions', code=2)
                assert refused['error']['code'] == 'E_USAGE' and not destination.exists()
                assert '--accept-basic-frame' in refused['error']['message']
                report = run('convert', op2_source, destination, '--mesh', companion,
                             '--accept-basic-frame',
                             '--accept-all')
                assert (report['points'], report['cells'], report['fields']) == (72, 186, 1)
                assert any(item['stage'] == 'assumption' and 'basic-frame' in item['detail']
                           for item in report['omissions'])
                routes += 1
            enriched_mesh = folder / 'op2-enriched-companion.vtu'
            run('convert', op2_source, enriched_mesh, '--mesh', op2_mesh,
                '--accept-all')
            enriched_result = folder / 'op2-with-enriched-mesh.vtu'
            enriched_report = run('convert', op2_source, enriched_result,
                                  '--mesh', enriched_mesh, '--accept-basic-frame',
                                  '--accept-all')
            assert any('numeric field(s) in companion mesh ignored' in item['detail']
                       for item in enriched_report['omissions'])
            routes += 1
            mismatched = folder / 'op2-mismatched-mesh.vtu'
            refused = run('convert', op2_source, mismatched,
                          '--mesh', sources['vtu'], '--accept-basic-frame',
                          '--accept-all', code=1)
            assert refused['error']['code'] == 'E_OP2' and not mismatched.exists()
            existing = folder / 'existing-companion.bdf'
            # Paired output must leave an existing companion untouched and
            # avoid installing a partial OP2 result.
            existing.write_text('already here')
            failed_pair = folder / 'no-partial-pair.op2'
            refused = run('convert', sources['frd'], failed_pair, '--mesh-out', existing,
                          '--accept-zero-rotations',
                          '--accept-all', code=1)
            assert refused['error']['code'] == 'E_EXISTS' and not failed_pair.exists()
            assert existing.read_text() == 'already here'
            paired_op2 = folder / 'paired-with-vtu.op2'
            paired_vtu = folder / 'paired-mesh.vtu'
            paired_report = run('convert', sources['frd'], paired_op2,
                                '--mesh-out', paired_vtu, '--accept-zero-rotations',
                                '--accept-all')
            assert paired_report['mesh_output']['format'] == 'vtu'
            assert run('validate', paired_vtu)['fields'] == 0
            paired_readback = folder / 'paired-vtu-readback.msh'
            readback_report = run('convert', paired_op2, paired_readback,
                                  '--mesh', paired_vtu, '--accept-basic-frame',
                                  '--accept-all')
            assert readback_report['fields'] == 1
            routes += 1
            mesh = folder / 'result-carrier-mesh.bdf'
            run('convert', sources['frd'], mesh, '--accept-all')
            for carrier in ('vtu', 'msh'):
                enriched = folder / f'frd-fields.{carrier}'
                run('convert', sources['frd'], enriched, '--accept-all')
                target = folder / f'{carrier}-fields-to-op2.op2'
                report = run('convert', enriched, target,
                             '--accept-zero-rotations', '--accept-all')
                assert target.is_file()
                assert any('typed float 0.0' in item['detail'] for item in report['omissions'])
                assert run('validate', target, '--mesh', mesh)['fields'] == 1
                routes += 1
        print(json.dumps({'routes_checked': routes,
                          'sources': sorted(sources)}))


if __name__ == '__main__':
    main()