import argparse
import os
import sys
from gain import logging
from gain.genomic_resources.cli import (
_create_proto,
_find_resources,
)
from gain.genomic_resources.cli_errors import (
report_resource_failure,
)
from gain.genomic_resources.histogram import (
NullHistogram,
plot_histogram,
)
from gain.genomic_resources.repository import (
GR_CONTENTS_FILE_NAME,
GenomicResource,
ReadWriteRepositoryProtocol,
)
from gain.genomic_resources.repository_factory import (
build_resource_implementation,
)
from gain.genomic_resources.score_implementation import (
ScoreImplementationBase,
)
from gain.utils.fs_utils import find_directory_with_a_file
from gain.utils.verbosity_configuration import VerbosityConfiguration
logger = logging.getLogger("draw_score_histograms")
[docs]
class ScorelessResourceError(TypeError):
"""A resource whose type carries no scores at all.
Distinguished from the errors that mean a resource is *broken*: there
is nothing wrong with a genome, it simply has no histograms to draw.
A ``TypeError`` because that is what selecting such a resource by id
has always raised, and callers still get exactly that.
"""
[docs]
def parse_cli_arguments() -> argparse.ArgumentParser:
"""Create CLI parser."""
parser = argparse.ArgumentParser(
description="Draw histograms for genomic scores.")
VerbosityConfiguration.set_arguments(parser)
parser.add_argument(
"-R",
"--repository",
help="Optional URL to the genomic resources repository.",
)
parser.add_argument(
"-r",
"--resource",
help="Optional URL to the resource.",
)
return parser
[docs]
def main(
argv: list[str] | None = None,
) -> None:
"""Liftover dae variants tool main function."""
if argv is None:
argv = sys.argv[1:]
parser = parse_cli_arguments()
args = parser.parse_args(argv)
VerbosityConfiguration.set(args)
repo_path = find_directory_with_a_file(
GR_CONTENTS_FILE_NAME,
args.repository,
)
if repo_path is None:
current_path = args.repository
if current_path is None:
current_path = os.getcwd()
print("Can't find repository starting from: %s", current_path)
sys.exit(1)
repo_url = str(repo_path)
print(f"working with repository: {repo_url}")
proto = _create_proto(repo_url)
if not isinstance(proto, ReadWriteRepositoryProtocol):
raise TypeError(
f"resource management works with RW protocols; "
f"{proto.proto_id} ({proto.scheme}) is read only")
resourses = _find_resources(proto, repo_url, resource=args.resource)
if not resourses:
print("Resource not found...")
sys.exit(1)
failed: set[str] = set()
scoreless: list[tuple[str, str]] = []
found_a_score_resource = False
for res in resourses:
try:
_draw_resource_histograms(res)
except ScorelessResourceError:
# Not a failure: a genome carrying no scores is its normal
# state, and every real GRR holds one. Collected rather than
# raised so that a sweep is not stopped by the most ordinary
# resource in the repository (gain#537).
scoreless.append((res.resource_id, res.get_type()))
logger.info(
"nothing to draw for <%s>: a %s resource carries no scores",
res.resource_id, res.get_type())
except Exception as err: # ruff: ignore[blind-except]
# One resource the tool cannot draw costs the user that
# resource, not the rest of the repository -- the same
# bargain every `grr_manage` sweep already makes (gain#364,
# gain#537).
#
# `Exception`, not `RESOURCE_ERRORS`, because this loop is of
# the family that RENDERS a resource, like the info-page loop
# -- not the manifest loop, which only hashes files and can
# afford to enumerate what a resource may raise. Reading a
# histogram back off a resource raises `KeyError` or
# `TypeError` for a statistics file that is well-formed JSON
# of the wrong shape, and drawing runs a plot function the
# RESOURCE names, whose module body may raise anything at
# all. Every one of those is the fault of one resource, and
# a narrower catch let each of them abort the sweep -- the
# very bug this is meant to fix. `report_resource_failure`
# still separates the two tiers: what is not a
# `RESOURCE_ERRORS` is reported as an unexpected internal
# error, with a traceback.
report_resource_failure(
err, "could not draw histograms for", res.resource_id)
failed.add(res.resource_id)
else:
found_a_score_resource = True
if failed:
# Reported once at the end as well as per resource: a long sweep
# scrolls its individual failures out of sight, and the exit
# status alone does not say which resources to go and look at.
# Reported before the scoreless case below because a resource
# that BROKE is the more urgent of the two.
logger.error(
"resources whose histograms could not be drawn in GRR <%s>: %s",
repo_url, ", ".join(sorted(failed)))
sys.exit(1)
if args.resource is not None and scoreless and not found_a_score_resource:
# Naming resources and getting no histogram at all is a mistake
# worth reporting, and `-r` is the only way a user asserts that
# particular resources have some. Keyed on that assertion rather
# than on how many resources matched: `-r` takes a glob, so the
# match count is a property of the repository, not of the ask,
# and keying on it made the same command an error or a silent
# success depending on what else the repository happened to hold.
raise ScorelessResourceError(_no_scores_message(scoreless))
def _no_scores_message(scoreless: list[tuple[str, str]]) -> str:
"""Say that nothing in a selection carries scores, naming each one.
The one-resource wording is kept verbatim from when this could only
ever be about a single resource (#337): it is what a user pointing
the tool at one resource by id has always been told.
"""
if len(scoreless) == 1:
resource_id, resource_type = scoreless[0]
return (
f"can't draw histograms for resource <{resource_id}>: "
f"a {resource_type} resource carries no scores")
listed = ", ".join(
f"<{resource_id}> ({resource_type})"
for resource_id, resource_type in sorted(scoreless))
return (
f"can't draw histograms: no selected resource carries "
f"scores -- {listed}")
def _draw_resource_histograms(res: GenomicResource) -> None:
"""Draw every non-null score histogram of one resource."""
impl = build_resource_implementation(res)
if not isinstance(impl, ScoreImplementationBase):
raise ScorelessResourceError(
f"can't draw histograms for resource <{res.resource_id}>: "
f"a {res.get_type()} resource carries no scores")
score = impl.score
for score_id in score.get_all_scores():
hist = score.get_score_histogram(score_id)
if isinstance(hist, NullHistogram):
continue
score_def = score.score_definitions[score_id]
plot_histogram(
res,
score.get_histogram_image_filename(score_id),
hist,
score_id,
score_def.small_values_desc,
score_def.large_values_desc,
)