Source code for pycif.plugins.chemistries.chimere.read_chemistry

import pandas as pd

from logging import debug

from .utils import make_inout_react_graph




def read_chemicalscheme(chemistry, **kwargs):
    """Read the CHIMERE chemical scheme from pre-computed files.

    Parses the scheme files in
    ``{workdir}/chemical_scheme/{schemeid}/`` and populates the chemistry
    plugin with species lists and reaction counts:

    * ``acspecies`` — active species with transport flags.
    * ``outspecies`` — output species with wet/dry conversion flags.
    * ``emis_species`` / ``nemisspec`` — anthropogenic emission species.
    * ``bio_species`` / ``nemisspec_interp`` — biogenic emission species.
    * ``dep_species`` / ``ndepspecies`` — depositing species.
    * ``nreacs`` — number of chemical reactions.
    * ``nfamilies`` — number of species families.
    * ``inout_reaction_graph`` — transitively-closed reactant → product
      mapping (built by :func:`~pycif.plugins.chemistries.chimere.utils.make_inout_react_graph`).

    Args:
        chemistry: CHIMERE chemistry plugin instance with ``workdir`` and
            ``schemeid`` set.
        **kwargs: unused.
    """

    debug("Reading Chemistry")

    workdir = chemistry.workdir
    dirchem_ref = f"{workdir}/chemical_scheme/{chemistry.schemeid}/"

    # ACTIVE SPECIES
    file_chem = f"{dirchem_ref}/ACTIVE_SPECIES.{chemistry.schemeid}"
    acspecies = pd.read_csv(
        file_chem, header=None, sep=" ",
        names=["ID", "name", "htransport", "vtransport", "bound_dry"]
    )
    chemistry.acspecies = chemistry.from_dict(
        {acspecies.loc[k, "name"]: {
            "htransport": acspecies.loc[k, "htransport"],
            "vtransport": acspecies.loc[k, "vtransport"],
            "bound_dry": bool(acspecies.loc[k, "bound_dry"]),
        } for k in acspecies.index}
    )
    chemistry.nacspecies = len(acspecies)

    # OUTPUT SPECIES
    file_out = f"{dirchem_ref}/OUTPUT_SPECIES.{chemistry.schemeid}"
    outspecies = pd.read_csv(
        file_out, header=None, sep=r"\s+", names=["name", "output_frac"]
    )
    chemistry.outspecies = chemistry.from_dict(
        {outspecies.loc[k, "name"]: {
            "output_frac": outspecies.loc[k, "output_frac"],
        } for k in outspecies.index}
    )

    # ANTHROPIC
    file_chem = f"{dirchem_ref}/ANTHROPIC.{chemistry.schemeid}"
    emis_species = pd.read_csv(
        file_chem, header=None, sep=" ", usecols=[0, 1], names=["ID", "name"]
    )
    chemistry.emis_species = chemistry.from_dict(
        {s: None for s in emis_species["name"]}
    )
    chemistry.nemisspec = len(emis_species)

    # BIOGENIC
    file_chem = f"{dirchem_ref}/BIOGENIC.{chemistry.schemeid}"
    bio_species = pd.read_csv(
        file_chem, header=None, sep=" ", usecols=[0, 1], names=["ID", "name"]
    )
    chemistry.bio_species = chemistry.from_dict(
        {s: None for s in bio_species["name"]}
    )
    chemistry.nemisspec_interp = len(bio_species)

    # DEPO_SPEC
    file_chem = f"{dirchem_ref}/DEPO_SPEC.{chemistry.schemeid}"
    dep_species = pd.read_csv(
        file_chem, header=None, sep=" ", usecols=[0], names=["name"]
    )
    chemistry.dep_species = chemistry.from_dict(
        {s: None for s in dep_species["name"]}
    )
    chemistry.ndepspecies = len(dep_species)

    # CHEMISTRY
    with open(dirchem_ref + "CHEMISTRY." + chemistry.schemeid, "r") as fsp:
        ln = fsp.readlines()
        chemistry.nreacs = len(ln)

    with open(dirchem_ref + "FAMILIES." + chemistry.schemeid, "r") as fsp:
        ln = fsp.readlines()
        chemistry.nfamilies = len(ln)

    # MAKE GRAPH OF PRODUCTS AND REACTIVE SPECIES
    chemistry.inout_reaction_graph = make_inout_react_graph(
        f"{dirchem_ref}/CHEMISTRY.{chemistry.schemeid}")

    # TODO: generalize number of prescribed species
    chemistry.nprspecies = 4