"""ACRO: Output storage and serialization."""
from __future__ import annotations
import datetime
import hashlib
import json
import logging
import os
import shutil
from pathlib import Path
from typing import Any
import pandas as pd
from pandas import DataFrame
from .constants import ARTIFACTS_DIR
from .summary import generate_session_summary
from .utils import is_blocked_extension
from .version import __version__
logger = logging.getLogger("acro:records")
def _copy_output_file_if_needed(filename: str, dest_dir: str) -> str | None:
"""Copy a file to the output directory if it exists and is not already there.
Parameters
----------
filename : str
Path to the source file.
dest_dir : str
Destination directory path.
Returns
-------
str | None
The filename (not full path) if successful, None if file doesn't exist.
"""
if not os.path.exists(filename):
logger.info(
"WARNING: Unable to add %s because the file does not exist", filename
)
return None
dest_path = os.path.normpath(os.path.join(dest_dir, Path(filename).name))
src_path = os.path.normpath(filename)
# Only copy if source and destination are different
if src_path != dest_path:
shutil.copy(filename, dest_dir)
return Path(filename).name
[docs]
def load_outcome(outcome: dict[str, Any]) -> DataFrame:
"""Return a DataFrame from an outcome dictionary.
Parameters
----------
outcome : dict
The outcome to load as a DataFrame.
"""
return pd.DataFrame.from_dict(outcome)
[docs]
def load_output(path: str, output: list[str]) -> list[str] | list[DataFrame]:
"""Return a loaded output.
Parameters
----------
path : str
The path to the output folder (with results.json).
output : list[str]
The output to load.
Returns
-------
list[str] | list[DataFrame]
The loaded output field.
"""
if len(output) < 1:
raise ValueError("error loading output")
loaded: list[DataFrame] = []
for filename in output:
_, ext = os.path.splitext(filename)
if ext == ".csv":
filename = os.path.normpath(f"{path}/{filename}")
loaded.append(pd.read_csv(filename))
if len(loaded) < 1: # output is path(s) to custom file(s)
return output
return loaded
[docs]
class Record:
"""Stores data related to a single output record.
Attributes
----------
uid : str
Unique identifier.
status : str
SDC status: {"pass", "fail", "review"}
output_type : str
Type of output, e.g., "regression"
properties : dict
Dictionary containing structured output data.
sdc : dict
Dictionary containing SDC results.
fair : dict
Dictionary containing FAIR description of SDC process
command : str
String representation of the operation performed.
summary : str
String summarising the ACRO checks.
outcome : DataFrame
DataFrame describing the details of ACRO checks.
output : Any
List of output DataFrames.
comments : list[str]
List of strings entered by the user to add comments to the output.
exception : str
Description of why an exception to fail/review should be granted.
timestamp : str
Time the record was created in ISO format.
"""
[docs]
def __init__(
self,
uid: str,
status: str,
output_type: str,
properties: dict,
sdc: dict,
fair: dict,
command: str,
summary: str,
outcome: DataFrame,
output: list[str] | list[DataFrame],
comments: list[str] | None = None,
) -> None:
"""Construct a new output record.
Parameters
----------
uid : str
Unique identifier.
status : str
SDC status: {"pass", "fail", "review"}
output_type : str
Type of output, e.g., "regression"
properties : dict
Dictionary containing structured output data.
sdc : dict
Dictionary containing SDC results.
fair : dict
Dictionary containing FAIR description of SDC process
command : str
String representation of the operation performed.
summary : str
String summarising the ACRO checks.
outcome : DataFrame
DataFrame describing the details of ACRO checks.
output : list[str] | list[DataFrame]
List of output DataFrames.
comments : list[str] | None, default None
List of strings entered by the user to add comments to the output.
"""
self.uid: str = uid
self.status: str = status
self.output_type: str = output_type
self.properties: dict = properties
self.sdc: dict = sdc
self.fair = fair
self.command: str = command
self.summary: str = summary
self.outcome: DataFrame = outcome
self.output: Any = output
self.comments: list[str] = [] if comments is None else comments
self.exception: str = ""
now = datetime.datetime.now()
self.timestamp: str = now.isoformat()
[docs]
def serialize_output(self, path: str = "outputs") -> list[str]:
"""Serialize outputs.
Parameters
----------
path : str, default 'outputs'
Name of the folder that outputs are to be written.
Returns
-------
list[str]
List of filepaths of the written outputs.
"""
output: list[str] = []
# check if the outputs directory was already created
try: # pragma: no cover
os.makedirs(path)
logger.debug("Directory %s created successfully", path)
except FileExistsError:
logger.debug("Directory %s already exists", path)
# save each output DataFrame to a different csv
if all(isinstance(obj, DataFrame) for obj in self.output):
for i, data in enumerate(self.output):
filename = f"{self.uid}_{i}.csv"
output.append(filename)
filename = os.path.normpath(f"{path}/{filename}")
with open(filename, mode="w", newline="", encoding="utf-8") as file:
file.write(data.to_csv())
# move custom files or plot files to the output folder
if self.output_type in ["custom", "survival plot", "histogram", "pie chart"]:
for filename in self.output:
copied_filename = _copy_output_file_if_needed(filename, path)
if copied_filename:
output.append(copied_filename)
return output
def __str__(self) -> str:
"""Return a string representation of a record.
Returns
-------
str
The record.
"""
return (
f"uid: {self.uid}\n"
f"status: {self.status}\n"
f"type: {self.output_type}\n"
f"properties: {self.properties}\n"
f"sdc: {self.sdc}\n"
f"fair: {self.fair}\n"
f"command: {self.command}\n"
f"summary: {self.summary}\n"
f"outcome: {self.outcome}\n"
f"output: {self.output}\n"
f"timestamp: {self.timestamp}\n"
f"comments: {self.comments}\n"
f"exception: {self.exception}\n"
)
[docs]
class Records:
"""Stores data related to a collection of output records."""
[docs]
def __init__(self, blocked_extensions: list[str] | None = None) -> None:
"""Construct a new object for storing multiple records."""
self.results: dict[str, Record] = {}
self.output_id: int = 0
self.blocked_extensions: list[str] = [
ext.lower() for ext in (blocked_extensions or [])
]
[docs]
def add(
self,
status: str = "",
output_type: str = "",
properties: dict | None = None,
sdc: dict | None = None,
fair: dict | None = None,
command: str = "",
summary: str = "",
outcome: DataFrame | None = None,
output: list[str] | list[DataFrame] | None = None,
comments: list[str] | None = None,
) -> None:
"""Add an output to the results.
Parameters
----------
status : str
SDC status: {"pass", "fail", "review"}
output_type : str
Type of output, e.g., "regression"
properties : dict
Dictionary containing structured output data.
sdc : dict
Dictionary containing SDC results.
fair : dict
Dictionary containing FAIR description of analysis
command : str
String representation of the operation performed.
summary : str
String summarising the ACRO checks.
outcome : DataFrame
DataFrame describing the details of ACRO checks.
output : list[str | list[DataFrame]
List of output DataFrames.
comments : list[str] | None, default None
List of strings entered by the user to add comments to the output.
"""
if outcome is None:
outcome = pd.DataFrame()
if output is None:
output = []
if properties is None:
properties = {}
if sdc is None:
sdc = {}
if fair is None:
fair = {}
new = Record(
uid=f"output_{self.output_id}",
status=status,
output_type=output_type,
properties=properties,
sdc=sdc,
fair=fair,
command=command,
summary=summary,
outcome=outcome,
output=output,
comments=comments,
)
self.results[new.uid] = new
self.output_id += 1
logger.info("add(): %s", new.uid)
[docs]
def remove(self, key: str) -> None:
"""Remove an output from the results.
Parameters
----------
key : str
Key specifying which output to remove, e.g., 'output_0'.
"""
if key not in self.results:
raise ValueError(f"unable to remove {key}, key not found")
del self.results[key]
logger.info("remove(): %s removed", key)
[docs]
def get(self, key: str) -> Record:
"""Return a specified output from the results.
Parameters
----------
key : str
Key specifying which output to return, e.g., 'output_0'.
Returns
-------
Record
The requested output.
"""
logger.debug("get(): %s ", key)
return self.results[key]
[docs]
def get_keys(self) -> list[str]:
"""Return the list of available output keys.
Returns
-------
list[str]
List of output names.
"""
logger.debug("get_keys()")
return list(self.results.keys())
[docs]
def get_index(self, index: int) -> Record:
"""Return the output at the specified position.
Parameters
----------
index : int
Position of the output to return.
Returns
-------
Record
The requested output.
"""
logger.debug("get_index(): %s", index)
key = list(self.results.keys())[index]
return self.results[key]
[docs]
def add_custom(self, filename: str, comment: str | None = None) -> bool:
"""Add an unsupported output to the results dictionary.
Parameters
----------
filename : str
The name of the file that will be added to the list of the outputs.
comment : str | None, default None
An optional comment.
Returns
-------
bool
False if the file extension is blocked, True otherwise.
"""
if is_blocked_extension(filename, self.blocked_extensions):
return False
if os.path.exists(filename):
output = Record(
uid=f"output_{self.output_id}",
status="review",
output_type="custom",
properties={},
sdc={},
fair={},
command="custom",
summary="review",
outcome=DataFrame(),
output=[os.path.normpath(filename)],
comments=None if comment is None else [comment],
)
self.results[output.uid] = output
self.output_id += 1
logger.info("add_custom(): %s", output.uid)
else:
logger.info(
"WARNING: Unable to add %s because the file does not exist", filename
) # pragma: no cover
return True
[docs]
def rename(self, old: str, new: str) -> None:
"""Rename an output.
Parameters
----------
old : str
The old name of the output.
new : str
The new name of the output.
"""
if old not in self.results:
raise ValueError(f"unable to rename {old}, key not found")
if new in self.results:
raise ValueError(f"unable to rename, {new} already exists")
self.results[new] = self.results[old]
self.results[new].uid = new
del self.results[old]
logger.info("rename_output(): %s renamed to %s", old, new)
[docs]
def add_exception(self, output: str, reason: str) -> None:
"""Add an exception request to an output.
Parameters
----------
output : str
The name of the output.
reason : str
The reason the output should be released.
"""
if output not in self.results:
raise ValueError(f"unable to add exception: {output} not found")
self.results[output].exception = reason
logger.info("exception request was added to %s", output)
[docs]
def print(self) -> str:
"""Print the current results.
Returns
-------
str
String representation of all outputs.
"""
logger.debug("print()")
outputs: str = ""
for _, record in self.results.items():
outputs += str(record) + "\n"
print(outputs) # noqa: T201
return outputs
[docs]
def validate_outputs(self) -> None:
"""Prompt researcher to complete any required fields."""
for _, record in self.results.items():
if record.status != "pass" and record.exception == "":
logger.info(
"\n%s\n"
"The status of the record above is: %s.\n"
"Please explain why an exception should be granted.\n",
str(record),
record.status,
)
record.exception = input("")
[docs]
def finalise(self, path: str, ext: str, interactive: bool = False) -> None:
"""Create a results file for checking.
Parameters
----------
path : str
Name of a folder to save outputs.
ext : str
Extension of the results file. Valid extensions: {json, xlsx}.
interactive : Bool
Whether to prompt the user to request exceptions for failing outputs.
"""
logger.debug("finalise()")
if interactive:
self.validate_outputs()
if ext not in ["json", "xlsx"]:
raise ValueError("Invalid file extension. Options: {json, xlsx}")
try:
generate_session_summary(self, path)
except Exception as e:
logger.warning("Failed to generate session summary: %s", str(e))
if ext == "json":
self.finalise_json(path)
elif ext == "xlsx":
self.finalise_excel(path)
self.write_checksums(path)
# check if the artifacts directory exists and delete it
if os.path.exists(ARTIFACTS_DIR):
shutil.rmtree(ARTIFACTS_DIR)
logger.info("outputs written to: %s", path)
def _tool_metadata(self) -> dict[str, str]:
"""Return tool metadata for 5s-crate Ro-CRATE compliance.
Returns
-------
dict
Tool metadata with @type, name, url, and version.
"""
return {
"@type": "SoftwareApplication",
"name": "ACRO",
"url": "https://github.com/AI-SDC/ACRO",
"version": __version__,
}
[docs]
def finalise_json(self, path: str) -> None:
"""Write outputs to a JSON file.
Parameters
----------
path : str
Name of a folder to save outputs.
"""
outputs: dict[str, Any] = {}
for key, val in self.results.items():
outputs[key] = {
"uid": val.uid,
"status": val.status,
"type": val.output_type,
"properties": val.properties,
"files": [],
"outcome": json.loads(val.outcome.to_json()),
"command": val.command,
"summary": val.summary,
"timestamp": val.timestamp,
"comments": val.comments,
"exception": val.exception,
"fair": val.fair,
}
files: list[str] = val.serialize_output(path)
for file in files:
outputs[key]["files"].append({"name": file, "sdc": val.sdc})
results: dict[str, str | dict] = {
"version": __version__,
"tool": self._tool_metadata(),
"results": outputs,
}
filename: str = os.path.normpath(f"{path}/results.json")
try:
with open(filename, "w", newline="", encoding="utf-8") as handle:
json.dump(results, handle, indent=4, sort_keys=False)
except FileNotFoundError: # pragma: no cover
logger.info(
"You don't have any output in the acro object. "
"Directory %s will not be created.",
path,
)
[docs]
def finalise_excel(self, path: str) -> None:
"""Write outputs to an excel spreadsheet.
Parameters
----------
path : str
Name of a folder to save outputs.
"""
filename: str = os.path.normpath(f"{path}/results.xlsx")
try: # check if the directory was already created
os.makedirs(path, exist_ok=True)
logger.debug("Directory %s created successfully", path)
except FileExistsError: # pragma: no cover
logger.debug("Directory %s already exists", path)
with pd.ExcelWriter(filename, engine="openpyxl") as writer:
# description sheet
sheet: list[str] = []
summary: list[str] = []
command: list[str] = []
for output_id, output in self.results.items():
if output.output_type == "custom":
continue # avoid writing custom outputs
sheet.append(output_id)
command.append(output.command)
summary.append(output.summary)
tmp_df = pd.DataFrame(
{"Sheet": sheet, "Command": command, "Summary": summary}
)
tmp_df.to_excel(writer, sheet_name="description", index=False, startrow=0)
# individual sheets
for output_id, output in self.results.items():
if output.output_type == "custom":
continue # avoid writing custom outputs
# command and summary
start = 0
tmp_df = pd.DataFrame(
[output.command, output.summary], index=["Command", "Summary"]
)
tmp_df.to_excel(writer, sheet_name=output_id, startrow=start)
# outcome
if output.outcome is not None:
output.outcome.to_excel(writer, sheet_name=output_id, startrow=4)
# output
for table in output.output:
start = 1 + writer.sheets[output_id].max_row
table.to_excel(writer, sheet_name=output_id, startrow=start)
[docs]
def finalise_evidence(self, path: str, evidence_store: dict | None = None) -> dict:
"""Serialise federated evidence to CSV files and return the manifest dict.
Each interim table (DataFrame) is saved as a separate CSV file in *path*.
The returned dictionary is suitable for writing to ``evidence.json``.
Parameters
----------
path : str
Directory where CSV files and ``evidence.json`` will be written.
evidence_store : dict, optional
The evidence dictionary to serialise. When ``None`` an empty dict
is used, producing an empty manifest. Callers should pass
``getattr(self_acro, "_federated_evidence", {})``.
Returns
-------
dict
Manifest describing every output's evidence and the CSV filenames.
"""
os.makedirs(path, exist_ok=True)
outputs: dict[str, Any] = {}
evidence_store = evidence_store if evidence_store is not None else {}
for uid, entry in evidence_store.items():
table_files: dict[str, str] = {}
for table_name, csv_text in entry.get("interim_tables", {}).items():
filename = f"{uid}_{table_name}.csv"
filepath = os.path.normpath(f"{path}/{filename}")
with open(filepath, "w", newline="", encoding="utf-8") as fh:
fh.write(csv_text)
table_files[table_name] = filename
dof = entry.get("dof")
dof_file: str | None = None
if isinstance(dof, str) and "\n" in dof:
dof_file = f"{uid}_dof.csv"
with open(
os.path.normpath(f"{path}/{dof_file}"),
"w",
newline="",
encoding="utf-8",
) as fh:
fh.write(dof)
dof_val: Any = dof_file
else:
dof_val = dof
outputs[uid] = {
"command": entry.get("command", ""),
"analysis_names": entry.get("analysis_names", []),
"variable_types": entry.get("variable_types", {}),
"dof": dof_val,
"interim_tables": table_files,
}
return {
"version": __version__,
"tool": self._tool_metadata(),
"outputs": outputs,
}
[docs]
def write_checksums(self, path: str) -> None:
"""Write checksums for each file to checksums folder.
Parameters
----------
path : str
Name of a folder to save outputs.
"""
if os.path.exists(path):
checksums: dict[str, str] = {}
for name in os.listdir(path):
filename = os.path.join(path, name)
if os.path.isfile(filename):
with open(filename, "rb") as file:
read = file.read()
checksums[name] = hashlib.sha256(read).hexdigest()
checksums_dir: str = os.path.normpath(f"{path}/checksums")
os.makedirs(checksums_dir, exist_ok=True)
for name, sha256 in checksums.items():
filename = os.path.join(checksums_dir, name + ".txt")
with open(filename, "w", encoding="utf-8") as file:
file.write(sha256)
else:
logger.debug("There is no file to do the checksums") # pragma: no cover
[docs]
def load_records(path: str) -> Records:
"""Load outputs from a JSON file.
Parameters
----------
path : str
Name of an output folder containing results.json.
Returns
-------
Records
The loaded records.
"""
records = Records()
filename = os.path.normpath(f"{path}/results.json")
with open(filename, newline="", encoding="utf-8") as handle:
data = json.load(handle)
if data["version"] != __version__: # pragma: no cover
raise ValueError("error loading output")
for key, val in data["results"].items():
files: list[dict] = val["files"]
filenames: list[str] = []
sdcs: list[dict] = []
for file in files:
filenames.append(file["name"])
sdcs.append(file["sdc"])
records.results[key] = Record(
uid=val["uid"],
status=val["status"],
output_type=val["type"],
properties=val["properties"],
sdc=sdcs[0],
fair=val["fair"],
command=val["command"],
summary=val["summary"],
outcome=load_outcome(val["outcome"]),
output=load_output(path, filenames),
comments=val["comments"],
)
records.results[key].exception = val["exception"]
records.results[key].timestamp = val["timestamp"]
return records