java-topology/defects/openfoam-0003/test/test_openfoam_0003.py

"""
openfoam-0003: DSMCCloud::initialise findIndex O(C*T*M*K) -> O(C*T*M) hash lookup

Stubs DSMCCloud molecule-type lookup in pure Python and measures the improvement.
"""
import time
import unittest


def initialise_defective(cells, tets_per_cell, molecules, typeIdList):
    """
    Defective: findIndex(typeIdList, name) -- O(K) string scan per (cell, tet, molecule).
    Total: O(C * T * M * K).
    """
    comparisons = 0

    def find_index(lst, val):
        nonlocal comparisons
        for item in lst:
            comparisons += 1
            if item == val:
                return lst.index(val)
        return -1

    for _ in range(cells):
        for _ in range(tets_per_cell):
            for mol in molecules:
                typeId = find_index(typeIdList, mol)
                assert typeId != -1, f"typeId {mol} not defined"

    return comparisons


def initialise_patched(cells, tets_per_cell, molecules, typeIdList):
    """
    Patched: pre-build HashTable<label, word> -- O(K) once, O(1) per lookup.
    Total: O(K + C * T * M).
    """
    comparisons = 0

    # Build lookup map once -- O(K)
    lookup = {}
    for idx, name in enumerate(typeIdList):
        comparisons += 1
        lookup[name] = idx

    for _ in range(cells):
        for _ in range(tets_per_cell):
            for mol in molecules:
                typeId = lookup[mol]
                assert typeId != -1, f"typeId {mol} not defined"

    return comparisons


class TestOpenFOAM0003(unittest.TestCase):

    def _run(self, cells, tets, molecules, typeIdList):
        def_ops = initialise_defective(cells, tets, molecules, typeIdList)
        pat_ops = initialise_patched(cells, tets, molecules, typeIdList)
        ratio = def_ops / max(pat_ops, 1)
        return def_ops, pat_ops, ratio

    def test_small_case(self):
        """Small case: 100 cells, 5 tets, 2 molecule species, 2 types."""
        cells, tets = 100, 5
        molecules = ["Ar", "N2"]
        typeIdList = ["Ar", "N2"]
        def_ops, pat_ops, ratio = self._run(cells, tets, molecules, typeIdList)
        self.assertGreater(ratio, 1.5,
            f"Expected speedup >1.5x, got {ratio:.1f}x "
            f"(defective={def_ops}, patched={pat_ops})")

    def test_medium_case(self):
        """Medium: 10000 cells, 5 tets, 3 molecule species, 5 types in list."""
        cells, tets = 10_000, 5
        molecules = ["Ar", "N2", "O2"]
        typeIdList = ["He", "Ar", "N2", "O2", "CO2"]
        def_ops, pat_ops, ratio = self._run(cells, tets, molecules, typeIdList)
        # Defective: 10000*5*3*5 = 750000. Patched: 5 + 10000*5*3 = 150005.
        # Ratio ~= 4.9x
        self.assertGreater(ratio, 3.0,
            f"Expected speedup >3x, got {ratio:.1f}x "
            f"(defective={def_ops}, patched={pat_ops})")

    def test_correctness_all_types_found(self):
        """Patched result must find all types correctly."""
        cells, tets = 10, 2
        molecules = ["He", "Ar"]
        typeIdList = ["Ar", "He", "N2"]
        lookup = {}
        for idx, name in enumerate(typeIdList):
            lookup[name] = idx
        self.assertEqual(lookup["He"], 1)
        self.assertEqual(lookup["Ar"], 0)

    def test_large_type_list(self):
        """Large type list magnifies the O(K) cost."""
        cells, tets = 1000, 5
        K = 20
        typeIdList = [f"mol_{j}" for j in range(K)]
        molecules = typeIdList[:3]  # only 3 species needed
        def_ops, pat_ops, ratio = self._run(cells, tets, molecules, typeIdList)
        # Worst-case scan: mol_0->1 compare, mol_1->2 compares, mol_2->3 compares
        # vs. hash: K + cells*tets*3 = 20 + 15000 = 15020
        self.assertGreater(ratio, 1.2,
            f"Expected speedup at K={K}, got {ratio:.2f}x")

    def test_timing(self):
        """Wall-clock: patched must be faster on a non-trivial workload."""
        cells, tets, K = 5000, 5, 5
        typeIdList = [f"species_{j}" for j in range(K)]
        molecules = typeIdList

        t0 = time.perf_counter()
        initialise_defective(cells, tets, molecules, typeIdList)
        dt_def = time.perf_counter() - t0

        t0 = time.perf_counter()
        initialise_patched(cells, tets, molecules, typeIdList)
        dt_pat = time.perf_counter() - t0

        self.assertGreater(dt_def, dt_pat * 0.5,
            f"Patched ({dt_pat:.4f}s) should be faster than defective ({dt_def:.4f}s)")


if __name__ == "__main__":
    import sys
    export_unbuffered = True  # PYTHONUNBUFFERED equivalent flag
    unittest.main(verbosity=2)