# -*- coding: utf-8 -*-
"""
Convert output from a generator of communities to one or several
community specification files.
"""
import json
import logging
from collections import OrderedDict
from math import ceil
from pathlib import Path
from . import ess
from .name import Name
from .types import Ecosystem
[docs]
def write_specs(
gen: Ecosystem,
out: Path,
eco: str,
models_per_spec: int | None = None,
catalog_dir: Path | None = None,
):
"""
Write specs from a generator of communities, possibly split into
several files.
"""
def write_one(com: dict, spec: Path, catalog_dir: None | Path = None):
"""Write community to spec file, possibly with catalog_dir hardlink."""
logging.debug("- write one %s, size %d", str(spec), len(com))
with open(spec, "w", encoding="UTF-8") as f:
json.dump(com, fp=f, indent=2)
if catalog_dir:
catalog_dir.mkdir(exist_ok=True, parents=True)
(catalog_dir / spec.name).unlink(missing_ok=True)
(catalog_dir / spec.name).hardlink_to(spec)
out.mkdir(exist_ok=True, parents=True)
# Collect communities from generator, so that we can count them and
# their sizes. Coerce community members to strings.
named = [(Name(com=x), x) if isinstance(x, list) else x for x in gen]
com = OrderedDict((str(name), [str(f) for f in coms]) for name, coms in named)
num_sbml = sum(len(x) for x in com.values())
num_com = len(com)
logging.info(
"Sample %s: spec %d com%s, %d models", eco, num_com, ess(com), num_sbml
)
# Write the spec, if necessary split into separate files.
if models_per_spec and 1 <= models_per_spec < num_sbml:
size = ceil(num_com / ceil(num_sbml / models_per_spec))
n_parts = ceil(num_com / size)
names = list(com.keys())
parts = [names[i::n_parts] for i in range(0, n_parts)]
for i in range(len(parts)):
subcom = OrderedDict((k, com[k]) for k in sorted(parts[i]))
write_one(subcom, out / f"{eco}_{i}.json", catalog_dir)
else:
write_one(com, out / f"{eco}.json")