Skip to content

Commit

Permalink
Update and refactor pooling map EPP (#559)(patch)
Browse files Browse the repository at this point in the history
### Changed
- Added pool sorting to the pooling map
- Changed the color set
- Added new flag, '-r' / '--round-decimals' for determining the amount of rounding
- Made the coloring truly container-specific, and not pool-container as it was before
  • Loading branch information
Karl-Svard authored Nov 22, 2024
1 parent 5f5db08 commit 21e4640
Show file tree
Hide file tree
Showing 2 changed files with 54 additions and 37 deletions.
85 changes: 48 additions & 37 deletions cg_lims/EPPs/files/pooling_map/make_pooling_map.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
import logging
import sys
from datetime import date
from typing import List, Tuple
from typing import Dict, List, Optional, Tuple

import click
from cg_lims import options
Expand All @@ -18,20 +18,25 @@
)
from cg_lims.exceptions import LimsError
from cg_lims.get.artifacts import get_artifacts
from genologics.entities import Artifact, Process
from genologics.entities import Artifact, Process, Sample
from matplotlib import colors

COLORS = list(colors.TABLEAU_COLORS.values())
COLORS: List[str] = list(colors.XKCD_COLORS.values())
LOG = logging.getLogger(__name__)


def add_pool_info(pool_udfs: List[str], pool: Artifact) -> str:
def add_pool_info(
pool_udfs: List[str],
pool: Artifact,
round_decimals: Optional[int],
) -> str:
"""Adding info about the pool"""

html = []
html: List[str] = []
for udf in pool_udfs:
value = pool.udf.get(udf)
if value is not None:
if round_decimals and type(value) is float:
value: float = round(value, round_decimals)
html.append(
f'<tr><td class="group-field-label">{udf}: </td><td class="group-field-value">{value}</td></tr>'
)
Expand All @@ -40,18 +45,17 @@ def add_pool_info(pool_udfs: List[str], pool: Artifact) -> str:

def add_sample_info_headers(udfs_headers: List[str]) -> str:
"""Adding headers for more info about samples in the pool"""

html = [f'<th style="width: 7%;" class="">{header}</th>' for header in udfs_headers]

html: List[str] = [f'<th style="width: 7%;" class="">{header}</th>' for header in udfs_headers]
return "".join(html)


def add_sample_info(artifact: Artifact, udfs: List[str]) -> str:
def add_sample_info(artifact: Artifact, udfs: List[str], round_decimals: Optional[int]) -> str:
"""Adding info about samples in the pool"""

html = []
html: List[str] = []
for udf in udfs:
value = artifact.udf.get(udf, "")
value = artifact.udf.get(udf)
if round_decimals and type(value) is float:
value: float = round(value, round_decimals)
html.append(f'<td class="" style="width: 7%;">{value}</td>')
return "".join(html)

Expand All @@ -61,13 +65,13 @@ def make_html(
process: Process,
pool_udfs: List[str],
sample_udfs: List[str],
round_decimals: Optional[int],
) -> str:
"""Building the html for the pooling map"""

html = []
html: List[str] = []
header_info = PlacementMapHeader(process_type=process.type.name, date=date.today().isoformat())
html.append(PLACEMENT_MAP_HEADER.format(**header_info.dict()))

source_containers: Dict[str, str] = {}
for pool in pools:
artifacts: List[Tuple[str, Artifact]] = [
(artifact.location[1], artifact) for artifact in pool.input_artifact_list()
Expand All @@ -79,54 +83,61 @@ def make_html(
nr_samples=len(artifacts),
pool_name=pool.name,
pool_id=pool.id,
pools_information=add_pool_info(pool_udfs, pool),
pools_information=add_pool_info(
pool_udfs=pool_udfs, pool=pool, round_decimals=round_decimals
),
)
html.append(POOL_HEADER.format(**pool_info.dict()))
extra_sample_columns: str = add_sample_info_headers(sample_udfs)
html.append(SAMPLE_COLUMN_HEADERS.format(extra_sample_columns=extra_sample_columns))
html.append("""</thead><tbody>""")
source_containers = []
for location, artifact in artifacts:
sample = artifact.samples[0]
sample_warning_color = "#F08080" if artifact.udf.get("Warning") else "#FFFFFF"
if artifact.container.name not in source_containers:
source_container_color = COLORS[0]
COLORS.pop(0)
source_containers.append(artifact.container.name)
sample: Sample = artifact.samples[0]
sample_warning_color: str = "#F08080" if artifact.udf.get("Warning") else "#FFFFFF"
if artifact.container.name not in source_containers.keys():
source_container_color: str = COLORS.pop(0)
source_containers[artifact.container.name] = source_container_color
sample_table_values = SampleTableSection(
sample_id=sample.id,
sample_warning_color=sample_warning_color,
source_well=location,
source_container=artifact.container.name,
source_container_color=source_container_color,
source_container_color=source_containers[artifact.container.name],
pool_name=pool.name,
extra_sample_values=add_sample_info(artifact, sample_udfs),
extra_sample_values=add_sample_info(
artifact=artifact, udfs=sample_udfs, round_decimals=round_decimals
),
)
html.append(SAMPLE_COLUMN_VALUES.format(**sample_table_values.dict()))
html.append("""</tbody></table><br><br></html>""")

html = "".join(html)
return html
return "".join(html)


@click.command()
@options.file_placeholder(help="File placeholder name.")
@options.pool_udfs(help="Pool UDFs to show in the placement map.")
@options.sample_udfs(help="Sample UDFs to show in the placement map.")
@options.round_decimals()
@click.pass_context
def pool_map(ctx, file: str, sample_udfs: List[str], pool_udfs: List[str]):
def pool_map(
ctx, file: str, sample_udfs: List[str], pool_udfs: List[str], round_decimals: str = None
):
"""Create a pool placement map."""

LOG.info(f"Running {ctx.command_path} with params: {ctx.params}")
process = ctx.obj["process"]
process: Process = ctx.obj["process"]

try:
pools = get_artifacts(process=process, input=False)
html = make_html(
pools,
process,
pool_udfs,
sample_udfs,
if round_decimals:
round_decimals: int = int(round_decimals)
pools: List[Artifact] = get_artifacts(process=process, input=False)
pools.sort(key=lambda x: x.id)
html: str = make_html(
pools=pools,
process=process,
pool_udfs=pool_udfs,
sample_udfs=sample_udfs,
round_decimals=round_decimals,
)
with open(f"{file}.html", "w") as file:
file.write(html)
Expand Down
6 changes: 6 additions & 0 deletions cg_lims/options.py
Original file line number Diff line number Diff line change
Expand Up @@ -717,3 +717,9 @@ def minimum_volume(
help: str = "Minimum volume",
) -> click.option:
return click.option("--min-volume", required=False, default=0, help=help)


def round_decimals(
help: str = "The number of decimals you want to round to.",
) -> click.option:
return click.option("-r", "--round-decimals", required=False, help=help)

0 comments on commit 21e4640

Please sign in to comment.