# -*- coding: utf-8 -*-
"""
CoCoMiCo Clingo answer objects.
"""
# mypy: disable-error-code="attr-defined"
import logging
from itertools import groupby
from typing import Tuple, cast
from clyngor.as_pyasp import Term
from cocomico.base import Biomolecule, Exchange
from cocomico.facts import to_biomolecule, to_exchange, to_metabolite, to_reaction
[docs]
class Answer:
"""
An answer to a clingo query is a collection of terms in the stable model.
The Answer object parses these terms into cocomico.base objects.
:param scope: community scope
:type scope: MetaboliteSet
:param exchange: metabolic exchanges
:type exchange: dict[Biomolecule,Exchange]
:param polyopsonist: exchanges with more than one consumer
:type scope: set[Biomolecule]
:param monopsonist: exchanges with exactly one consumer
:type scope: set[Biomolecule]
:param activated: activated reactions
:type activated: set[Reaction]
:param produced_seeds: seeds that are also produced
:type produced_seeds: set[Biomolecule]
:param consumed_seeds: seeds that are actually consumed
:type consumed_seeds: set[Biomolecule]
"""
# pylint: disable=R0902,R0903
def __init__(
# pylint: disable=R0913
self,
# scope: all, # type: ignore
scope_flat: all, # type: ignore
exchange: all, # type: ignore
polyopsonist: all, # type: ignore
monopsonist: all, # type: ignore
polypolist: all, # type: ignore
monopolist: all, # type: ignore
# activated: all, # type: ignore
activated_flat: all, # type: ignore
# produced_seed: all, # type: ignore
produced_seed_flat: all, # type: ignore
# consumed_seed: all, # type: ignore
consumed_seed_flat: all, # type: ignore
) -> None:
"""
:param scope: all `scope` terms
:param exchange: all `exchange` terms
:param polyopsonist: all `polyopsonist` terms
:param monopsonist: all `monopsonist` terms
"""
def groupby_community(terms: list[Term]) -> dict[str, list[Term]]:
"""Factor terms to group those that have the same community."""
def key(t):
"""Community atom is last in the relation."""
atom = t[-1].strip('"')
if atom == "all" or atom[0:4] == "self":
return atom
return f'self("{ atom }")'
return {
community.strip('"'): [*terms]
for community, terms in groupby(sorted(terms, key=key), key=key)
}
# Scope from scope/2, for example
# scope(metabolite("M_C_c","Com1Org1"),all)
# scope(metabolite("M_M_c","Com1Org2"),self("Com1Org2"))
self.scope = {
community: to_metabolite(terms)
for community, terms in groupby_community(scope_flat).items()
}
# Activated from activated/2, for example
# activated(reaction("R_B_JD","Com1Org1"),all)
# activated(reaction("R_C_PY","Com1Org2"),self("Com1Org1"))
# or from activated/3, for example
# activated("R_B_JD","Com1Org1",all)
# activated("R_C_PY","Com1Org2","Com1Org1")
self.activated = {
community: to_reaction(terms)
for community, terms in groupby_community(activated_flat).items()
}
# Exchange from exchange/3, for example
# exchange("M_B_c","Com1Org3","Com1Org1")
# exchange("M_C_c","Com1Org1","Com1Org2")
self.exchange: dict[Biomolecule, set[Exchange]] = {
Biomolecule(m.strip('"')): to_exchange(tuples)
for m, tuples in groupby(
sorted(cast(list[Tuple[Term, Term, Term]], exchange)),
key=lambda t: t[0],
)
}
# Exchange cardinalities, for example
# polyopsonist("M_C_c",2)
# monopolist("M_B_c",1)
self.polyopsonist = {Biomolecule(m.strip('"')): int(n) for m, n in polyopsonist}
self.monopsonist = {Biomolecule(m.strip('"')): int(n) for m, n in monopsonist}
self.polypolist = {Biomolecule(m.strip('"')): int(n) for m, n in polypolist}
self.monopolist = {Biomolecule(m.strip('"')): int(n) for m, n in monopolist}
# Produced and consumed seeds, for example
# consumed_seed("M_F_c",all)
# consumed_seed("M_X1_c",all)
# produced_seed("M_F_c",all)
self.produced_seeds = {
community: to_biomolecule(terms)
for community, terms in groupby_community(produced_seed_flat).items()
}
self.consumed_seeds = {
community: to_biomolecule(terms)
for community, terms in groupby_community(consumed_seed_flat).items()
}
logging.debug("Init %s", str(self))
[docs]
def __str__(self):
"""
String representation of an answer
"""
# pylint: disable=C0103
SC = " ".join(sorted(str(m) for m in self.scope.get("all", [])))
EX = " ".join(sorted(f"{ k }:{ v }" for k, v in self.exchange.items()))
C2 = " ".join(sorted(str(m) for m in self.polyopsonist.keys()))
C1 = " ".join(sorted(str(m) for m in self.monopsonist.keys()))
P2 = " ".join(sorted(str(m) for m in self.polypolist.keys()))
P1 = " ".join(sorted(str(m) for m in self.monopolist.keys()))
return (
f"Answer<{hex(id(self))}> <"
f"scope { SC }; exch { EX }; "
f"C>1 { C2 }; C=1 { C1 }; P>1 { P2 }; P=1 { P1 }"
">"
)