"""
Generate reports for communities, using results that are already computed.
"""
import csv
import json
import logging
from pathlib import Path
from typing import Callable, Dict, Iterable
from pathvalidate import sanitize_filepath
from cocomico import score
from cocomico.base import Exchange, ExchangeNM, MetaboliteSet, Seeds
from cocomico.community import Community
from cocomico.constants import COMP_NUM, COOP_CONS, COOP_NUM, COOP_PROD, CSV_HEADER
from cocomico.model import Model
# High-level reports
PRECISION = "{:.6g}" # in fact g default precision=6
[docs]
def write_json(
communities: Dict[str, Community],
output_dir: Path,
mode: str = "w+",
):
"""
Write a JSON-formatted file of computed results for each member of
a collection of communities.
"""
output_dir.mkdir(parents=True, exist_ok=True)
for name, community in sorted(communities.items()):
output_path = output_dir / sanitize_filepath(f"{ name }.json")
payload = from_community(community, with_results=True)
with open(output_path, mode=mode, encoding="UTF-8") as json_file:
json_file.write(payload)
[docs]
def to_json(payload: object) -> str:
"""
Serialize a cocomico object and convert to JSON string.
"""
return json.dumps(payload, indent=4, sort_keys=True)
[docs]
def write_tabular(
# pylint: disable=R0913
communities: Dict[str, Community],
output_file: Path,
mode: str = "w",
dialect="excel-tab",
header=True,
symbolic_seeds=False,
seeds_todo: Iterable[Seeds] | None = None,
):
"""
Write CSV-formatted table of computed results for a collection of
communities.
"""
with open(output_file, mode=mode, encoding="UTF-8") as output:
writer = csv.writer(output, dialect=dialect, lineterminator="\n")
if header:
logging.debug("Report tabular add header %s", str(header))
writer.writerow(CSV_HEADER)
for name, community in sorted(communities.items()):
logging.debug("Report tabular for [%s] %s", name, str(community))
for seeds in seeds_todo or community.seeds():
logging.debug(' seeds "%s" %s', str(seeds.name), str(seeds))
coop, coop_metrics = score.cooperation(community, seeds)
comp, comp_metrics = score.competition(community, seeds)
seeds_col = (
seeds.name
if symbolic_seeds
else ",".join(str(s) for s in sorted(seeds))
)
logging.debug("seeds_col = %s", str(seeds_col))
writer.writerow(
[
name,
seeds_col,
PRECISION.format(score.delta(community, seeds)[0]),
PRECISION.format(coop),
PRECISION.format(score.rho(community, seeds)[0]),
PRECISION.format(comp),
comp_metrics[COMP_NUM],
coop_metrics[COOP_NUM],
PRECISION.format(coop_metrics[COOP_PROD]),
PRECISION.format(coop_metrics[COOP_CONS]),
len(community),
]
)
[docs]
def read_tabular(
output_file: Path,
mode: str = "r",
dialect="excel-tab",
):
"""
Read CSV-formatted table of computed results.
"""
result: dict[str, list] = {}
with open(output_file, mode=mode, encoding="UTF-8") as input_csv:
reader = csv.reader(input_csv, dialect=dialect)
for row in reader:
if row != CSV_HEADER:
name = row.pop(0)
seeds = row.pop(0).split(",")
result[name] = [seeds, *row]
return result
# Decorator to associate serializers with CoCoMiCo types
serializers: dict[str, Callable] = {}
[docs]
def serializer(object_type: str):
"""Decorator to register serializers by name."""
def decorator(f):
"""Wrap a serializer."""
serializers[object_type] = f
return f
return decorator
[docs]
def lookup_serializer(coco: object):
"""Look up the serializer by type name."""
return serializers[type(coco).__name__]
# Serializers
[docs]
def serialize(coco: object, *args, **kwargs):
"""
Dispatch serialization based on object type
"""
return lookup_serializer(coco)(coco, *args, **kwargs)
[docs]
@serializer("int")
@serializer("float")
def serialize_number(obj):
"""Number do not need conversion."""
return obj if isinstance(obj, int) else float(PRECISION.format(obj))
[docs]
@serializer("str")
@serializer("Biomolecule")
@serializer("Metabolite")
@serializer("Reaction")
def serialize_str(obj):
"""Convert an object that has a __str__ method."""
return str(obj) if obj else None
[docs]
@serializer("set")
@serializer("list")
@serializer("tuple")
@serializer("frozenset")
def serialize_iterable(obj: Iterable, sort=True):
"""Convert iterable object."""
gen = (serialize(i) for i in obj)
return sorted(gen) if sort else list(gen)
[docs]
@serializer("dict")
def serialize_dict(obj: dict):
"""Convert dict object."""
return {k: serialize(v) for k, v in obj.items()}
[docs]
@serializer("Seeds")
def serialize_seeds(seeds: Seeds) -> list[str]:
"""Convert seeds."""
return sorted(str(biomolecule) for biomolecule in seeds)
[docs]
@serializer("Model")
def serialize_model(model: Model) -> list:
"""Convert model."""
return [
(str(pair[0]), str(pair[1]))
for kind in ["product", "reactant"]
for pair in model.relations[kind]
]
[docs]
@serializer("Exchange")
def serialize_exchange(exchange: Exchange) -> list:
"""Convert set of metabolites."""
return [serialize(exchange.producer), serialize(exchange.consumer)]
[docs]
def serialize_results(community: Community) -> list:
"""
Convert community analyses.
"""
analyses = ["scope", "exchange", "polyopsonist", "monopsonist"]
subsets = ["activated", "produced_seeds", "consumed_seeds"]
scores = ["cooperation", "competition", "rho", "delta"]
payload = [
{"seeds": serialize(seeds)}
| {k: serialize(getattr(community, k)(seeds)) for k in analyses + subsets}
| {
k: serialize(getattr(score, k)(community, seeds), sort=False)
for k in scores
}
for seeds in community.seeds()
]
# override exchange to use ExchangeNM
for result in payload:
for biomolecule, relations in result["exchange"].items():
nm = ExchangeNM(relations=relations)
result["exchange"][biomolecule] = {
"producers": sorted(nm.producers),
"consumers": sorted(nm.consumers),
}
return payload