PALEOS-API live tabulation (producer)

def make_grid_at_resolution(
    p_lo: float,
    p_hi: float,
    t_lo: float,
    t_hi: float,
    pts_per_decade: int = DEFAULT_PTS_PER_DECADE,
) -> GridSpec:
    """Build a ``GridSpec`` with a uniform log-log resolution.

    Parameters
    ----------
    p_lo, p_hi : float
        Pressure bounds [Pa], both > 0.
    t_lo, t_hi : float
        Temperature bounds [K], both > 0.
    pts_per_decade : int
        Nodes per decade on each axis. Default
        ``DEFAULT_PTS_PER_DECADE`` = 600 (4x the shipped Zenodo resolution).
    """
    return GridSpec(
        p_lo=p_lo,
        p_hi=p_hi,
        n_p=_n_points_per_decade(p_lo, p_hi, pts_per_decade),
        t_lo=t_lo,
        t_hi=t_hi,
        n_t=_n_points_per_decade(t_lo, t_hi, pts_per_decade),
    )

def make_default_grid_iron() -> GridSpec:
    """Default Fe grid: 1e5 to 1e14 Pa, 300 to 20000 K at 600 pts/decade."""
    return make_grid_at_resolution(1e5, 1e14, 300.0, 2.0e4)

def make_default_grid_mgsio3() -> GridSpec:
    """Default MgSiO3 grid: 1e5 to 1e14 Pa, 300 to 11500 K at 600 pts/decade."""
    return make_grid_at_resolution(1e5, 1e14, 300.0, 1.15e4)

def make_default_grid_h2o() -> GridSpec:
    """Default H2O grid: 0.1 Pa to 1e14 Pa, 150 to 1e5 K at 600 pts/decade.

    Wider P range than Fe/MgSiO3 to match AQUA table coverage (extends to
    1 micro-bar). H2O is table-backed in PALEOS (``Haldemann20``) so per-point
    cost is low.
    """
    return make_grid_at_resolution(1e-1, 1e14, 150.0, 1.0e5)

def paleos_installed_sha() -> str:
    """Return the git SHA of the installed PALEOS checkout, or ``'unknown'``.

    PALEOS does not expose ``__commit_sha__``; we fall back to
    ``git rev-parse HEAD`` against the directory holding ``paleos``.
    If both fail, we use ``paleos.__version__`` as a coarse identifier.
    """
    try:
        import paleos as _paleos
    except ImportError:
        return 'paleos-not-installed'

    pkg_dir = Path(_paleos.__file__).resolve().parent
    repo_dir = pkg_dir.parent
    try:
        sha = _subprocess.check_output(
            ['git', '-C', str(repo_dir), 'rev-parse', 'HEAD'],
            stderr=_subprocess.DEVNULL,
            text=True,
        ).strip()
        if sha:
            return sha
    except (_subprocess.CalledProcessError, FileNotFoundError, OSError):
        pass

    return f'version-{getattr(_paleos, "__version__", "unknown")}'

def generate_paleos_api_unified_table(
    material: str,
    out_path: str | Path,
    grid: GridSpec,
    *,
    h2o_table_path: str | None = None,
    log_every: int = 50,
    n_workers: int = 1,
) -> dict:
    """Generate a unified PALEOS ``.dat`` for one material.

    Stable-phase dispatch: for each (P, T) the PALEOS top-level class
    (``IronEoS`` / ``MgSiO3EoS`` / ``WaterEoS``) chooses the thermodynamically
    stable phase and evaluates that phase's EoS. Phase label is PALEOS's
    intrinsic string (``'solid-hpcen'``, ``'liquid'``, ``'hcp-Fe'``, etc).

    Parameters
    ----------
    material : {'iron', 'mgsio3', 'h2o'}
        Material selector.
    out_path : path-like
        Destination ``.dat`` path. Parent directories are created.
    grid : GridSpec
        (P, T) grid. ``p_lo`` must be > 0.
    h2o_table_path : str or None
        AQUA table path. Passed to ``WaterEoS`` when ``material == 'h2o'``.
    log_every : int
        Emit a progress line every ``log_every`` pressure rows.
    n_workers : int
        Parallelism. ``1`` = serial (default, backward-compatible). ``-1`` =
        ``os.cpu_count()``. Any other positive int = that many workers.
        Workers parallelize over P-rows; each worker instantiates its own
        PALEOS EoS at init (Wolf18 sympy compile cost paid once per worker,
        not per task).

    Returns
    -------
    dict
        ``{'n_valid', 'n_skipped', 'sha'}``.
    """
    if grid.p_lo <= 0:
        raise ValueError('GridSpec.p_lo must be > 0 (log spacing)')
    if material not in _HUMAN_MATERIAL:
        raise ValueError(f"material must be one of 'iron', 'mgsio3', 'h2o'; got {material!r}")

    human_material = _HUMAN_MATERIAL[material]
    n_workers = _resolve_n_workers(n_workers)
    sha = paleos_installed_sha()
    P_axis, T_axis = grid.axes()
    T_axis_list = T_axis.tolist()  # avoid repeated numpy scalar conversions in workers

    out_path = Path(out_path)
    out_path.parent.mkdir(parents=True, exist_ok=True)
    tmp_path = out_path.with_suffix(out_path.suffix + '.tmp')

    tasks = [(i, float(P), T_axis_list) for i, P in enumerate(P_axis)]

    rows_by_i: dict[int, str] = {}
    n_valid = 0
    n_skipped = 0
    n_done = 0

    if n_workers == 1:
        # Serial path: no Pool, no worker init — tabulate in-process.
        _worker_init(material, h2o_table_path=h2o_table_path)
        iterator = (_worker_unified_row(t) for t in tasks)
    else:
        ctx = _mp.get_context('spawn')
        pool = ctx.Pool(
            processes=n_workers,
            initializer=_worker_init,
            initargs=(material, h2o_table_path),
        )
        # chunksize kept small so progress logs stay meaningful at high cost/cell.
        chunksize = max(1, len(tasks) // (n_workers * 16))
        iterator = pool.imap_unordered(_worker_unified_row, tasks, chunksize=chunksize)

    try:
        for i_P, rows_text, nv, ns in iterator:
            rows_by_i[i_P] = rows_text
            n_valid += nv
            n_skipped += ns
            n_done += 1
            if log_every and (n_done % log_every == 0):
                logger.info(
                    'paleos_api %s unified: %d / %d P rows done (valid=%d skipped=%d)',
                    human_material,
                    n_done,
                    grid.n_p,
                    n_valid,
                    n_skipped,
                )
    finally:
        if n_workers > 1:
            pool.close()
            pool.join()

    with open(tmp_path, 'w') as f:
        _write_header(
            f,
            human_material,
            'unified (stable-phase)',
            grid,
            sha,
            n_valid=n_valid,
            n_skipped=n_skipped,
        )
        for i in range(grid.n_p):
            f.write(rows_by_i[i])
    _os.replace(tmp_path, out_path)

    logger.info(
        'paleos_api %s unified: wrote %s (valid=%d skipped=%d sha=%s n_workers=%d)',
        human_material,
        out_path,
        n_valid,
        n_skipped,
        sha[:10],
        n_workers,
    )
    return {'n_valid': n_valid, 'n_skipped': n_skipped, 'sha': sha}

def generate_paleos_api_2phase_mgsio3_tables(
    out_solid: str | Path,
    out_liquid: str | Path,
    grid: GridSpec,
    *,
    log_every: int = 50,
    n_workers: int = 1,
) -> dict:
    """Generate solid and liquid MgSiO3 P-T tables with metastable extensions.

    Replaces the shipped ``paleos_mgsio3_tables_pt_proteus_solid.dat`` and
    ``paleos_mgsio3_tables_pt_proteus_liquid.dat`` — bit-format-compatible,
    consumed unchanged by ``eos_export.generate_spider_eos_tables`` via
    ``solid_eos_file`` / ``liquid_eos_file``.

    Parameters
    ----------
    out_solid, out_liquid : path-like
        Destinations for the two files. Parent directories are created.
    grid : GridSpec
        (P, T) grid; ``p_lo`` must be > 0.
    log_every : int
        Progress log cadence in pressure rows.
    n_workers : int
        Parallelism. ``1`` = serial (default, backward-compatible). ``-1`` =
        ``os.cpu_count()``. Workers parallelize over P-rows; each worker
        instantiates ``MgSiO3EoS`` + ``Wolf18`` once (Wolf18's sympy compile
        cost is paid once per worker, not per task).

    Returns
    -------
    dict
        Summary with per-side ``n_valid`` / ``n_skipped`` and the installed
        PALEOS ``sha``.

    Notes
    -----
    The liquid-side file evaluates ``paleos.mgsio3_eos.Wolf18`` on every
    grid node, including below the solidus (metastable liquid). The
    solid-side file dispatches through ``MgSiO3EoS()._phase_eos_map`` using
    ``_get_mgsio3_solid_phase`` (melting-curve test removed) so every node
    gets a solid-polymorph EoS evaluated metastably above the liquidus if
    needed.
    """
    if grid.p_lo <= 0:
        raise ValueError('GridSpec.p_lo must be > 0 (log spacing)')

    n_workers = _resolve_n_workers(n_workers)
    sha = paleos_installed_sha()
    P_axis, T_axis = grid.axes()
    T_axis_list = T_axis.tolist()

    out_solid = Path(out_solid)
    out_liquid = Path(out_liquid)
    out_solid.parent.mkdir(parents=True, exist_ok=True)
    out_liquid.parent.mkdir(parents=True, exist_ok=True)
    tmp_solid = out_solid.with_suffix(out_solid.suffix + '.tmp')
    tmp_liquid = out_liquid.with_suffix(out_liquid.suffix + '.tmp')

    tasks = [(i, float(P), T_axis_list) for i, P in enumerate(P_axis)]

    solid_by_i: dict[int, str] = {}
    liquid_by_i: dict[int, str] = {}
    n_valid_s = n_skip_s = 0
    n_valid_l = n_skip_l = 0
    n_done = 0

    if n_workers == 1:
        _worker_init('mgsio3')
        iterator = (_worker_2phase_row(t) for t in tasks)
    else:
        ctx = _mp.get_context('spawn')
        pool = ctx.Pool(
            processes=n_workers,
            initializer=_worker_init,
            initargs=('mgsio3',),
        )
        chunksize = max(1, len(tasks) // (n_workers * 16))
        iterator = pool.imap_unordered(_worker_2phase_row, tasks, chunksize=chunksize)

    try:
        for i_P, s_text, l_text, nv_s, ns_s, nv_l, ns_l in iterator:
            solid_by_i[i_P] = s_text
            liquid_by_i[i_P] = l_text
            n_valid_s += nv_s
            n_skip_s += ns_s
            n_valid_l += nv_l
            n_skip_l += ns_l
            n_done += 1
            if log_every and (n_done % log_every == 0):
                logger.info(
                    'paleos_api MgSiO3 2-phase: %d / %d P rows done '
                    '(solid valid=%d skipped=%d | liquid valid=%d skipped=%d)',
                    n_done,
                    grid.n_p,
                    n_valid_s,
                    n_skip_s,
                    n_valid_l,
                    n_skip_l,
                )
    finally:
        if n_workers > 1:
            pool.close()
            pool.join()

    with open(tmp_solid, 'w') as f:
        _write_header(
            f,
            'MgSiO3',
            'solid (metastable extension above liquidus)',
            grid,
            sha,
            n_valid=n_valid_s,
            n_skipped=n_skip_s,
            extra_lines=[
                'Solid polymorph picked by _get_mgsio3_solid_phase; melting-curve test suppressed.',
                'Phase column is the polymorph label even if (P, T) is above the liquidus.',
            ],
        )
        for i in range(grid.n_p):
            f.write(solid_by_i[i])
    _os.replace(tmp_solid, out_solid)

    with open(tmp_liquid, 'w') as f:
        _write_header(
            f,
            'MgSiO3',
            'liquid (Wolf18, metastable extension below solidus)',
            grid,
            sha,
            n_valid=n_valid_l,
            n_skipped=n_skip_l,
            extra_lines=[
                'Wolf18 RTpress evaluated everywhere; phase column is always "liquid".',
            ],
        )
        for i in range(grid.n_p):
            f.write(liquid_by_i[i])
    _os.replace(tmp_liquid, out_liquid)

    logger.info(
        'paleos_api MgSiO3 2-phase: wrote %s (valid=%d/%d) and %s (valid=%d/%d) '
        'sha=%s n_workers=%d',
        out_solid,
        n_valid_s,
        grid.n_p * grid.n_t,
        out_liquid,
        n_valid_l,
        grid.n_p * grid.n_t,
        sha[:10],
        n_workers,
    )
    return {
        'solid': {'n_valid': n_valid_s, 'n_skipped': n_skip_s, 'path': str(out_solid)},
        'liquid': {'n_valid': n_valid_l, 'n_skipped': n_skip_l, 'path': str(out_liquid)},
        'sha': sha,
    }