"""Summarize the HED tags in collection of tabular files."""
import numpy as np
import pandas as pd
from hed.models.tabular_input import TabularInput
from hed.tools.analysis.hed_tag_counts import HedTagCounts
from hed.tools.analysis.event_manager import EventManager
from hed.tools.analysis.hed_tag_manager import HedTagManager
from remodeler.operations.base_op import BaseOp
from remodeler.operations.base_summary import BaseSummary
class HedTagSummary(BaseSummary):
"""Manager of the HED tag summaries."""
def __init__(self, sum_op):
"""Constructor for HED tag summary manager.
Parameters:
sum_op (SummarizeHedTagsOp): Operation associated with this summary.
"""
super().__init__(sum_op)
self.sum_op = sum_op
def update_summary(self, new_info):
"""Update the summary for a given tabular input file.
Parameters:
new_info (dict): A dictionary with the parameters needed to update a summary.
Notes:
- The summary needs a "name" str, a "schema", a "df, and a "Sidecar".
"""
counts = HedTagCounts(new_info["name"], total_events=len(new_info["df"]))
input_data = TabularInput(new_info["df"], sidecar=new_info["sidecar"], name=new_info["name"])
tag_man = HedTagManager(EventManager(input_data, new_info["schema"]), remove_types=self.sum_op.remove_types)
hed_objs = tag_man.get_hed_objs(
include_context=self.sum_op.include_context, replace_defs=self.sum_op.replace_defs
)
for hed in hed_objs:
counts.update_tag_counts(hed, new_info["name"])
self.summary_dict[new_info["name"]] = counts
def get_details_dict(self, tag_counts) -> dict:
"""Return the summary-specific information in a dictionary.
Parameters:
tag_counts (HedTagCounts): Contains the counts of tags in the dataset.
Returns:
dict: dictionary with the summary results.
"""
template, unmatched = tag_counts.organize_tags(self.sum_op.tags)
details = {}
for key, key_list in self.sum_op.tags.items():
details[key] = self._get_details(key_list, template, verbose=True)
leftovers = [value.get_info(verbose=True) for value in unmatched]
return {
"Name": tag_counts.name,
"Total events": tag_counts.total_events,
"Total files": len(tag_counts.files.keys()),
"Files": list(tag_counts.files.keys()),
"Specifics": {"Main tags": details, "Other tags": leftovers},
}
def _get_result_string(self, name, result, indent=BaseSummary.DISPLAY_INDENT):
"""Return a formatted string with the summary for the indicated name.
Parameters:
name (str): Identifier (usually the filename) of the individual file.
result (dict): The dictionary of the summary results indexed by name.
indent (str): A string containing spaces used for indentation (usually 3 spaces).
Returns:
str: The results in a printable format ready to be saved to a text file.
Notes:
This calls _get_dataset_string to get the overall summary string and
_get_individual_string to get an individual summary string.
"""
if name == "Dataset":
return self._get_dataset_string(result, indent=indent)
return self._get_individual_string(result, indent=indent)
def merge_all_info(self) -> "HedTagCounts":
"""Create a HedTagCounts containing the overall dataset HED tag summary.
Returns:
HedTagCounts: The overall dataset summary object for HED tag counts.
"""
all_counts = HedTagCounts("Dataset")
for _key, counts in self.summary_dict.items():
all_counts.merge_tag_dicts(counts.tag_dict)
for file_name in counts.files.keys():
all_counts.files[file_name] = ""
all_counts.total_events = all_counts.total_events + counts.total_events
return all_counts
@staticmethod
def summary_to_dict(specifics, transform=np.log10, scale_adjustment=7) -> dict:
"""Convert a HedTagSummary json specifics dict into the word cloud input format.
Parameters:
specifics (dict): Dictionary with keys "Main tags" and "Other tags".
transform (func): The function to transform the number of found tags.
Default log10
scale_adjustment (int): Value added after transform.
Returns:
dict: A dict of the words and their occurrence count.
Raises:
KeyError: A malformed dictionary was passed.
"""
if transform is None:
def transform(x):
return x
word_dict = {}
tag_dict = specifics.get("Main tags", {})
for tag, tag_sub_list in tag_dict.items():
if tag == "Exclude tags":
continue
for tag_sub_dict in tag_sub_list:
word_dict[tag_sub_dict["tag"]] = transform(tag_sub_dict["events"]) + scale_adjustment
other_dict = specifics.get("Other tags", [])
for tag_sub_list in other_dict:
word_dict[tag_sub_list["tag"]] = transform(tag_sub_list["events"]) + scale_adjustment
return word_dict
@staticmethod
def _get_dataset_string(result, indent=BaseSummary.DISPLAY_INDENT):
"""Return a string with the overall summary for all the tabular files.
Parameters:
result (dict): Dictionary of merged summary information.
indent (str): String of blanks used as the amount to indent for readability.
Returns:
str: Formatted string suitable for saving in a file or printing.
"""
sum_list = [f"Dataset: Total events={result.get('Total events', 0)} Total files={len(result.get('Files', []))}"]
sum_list = sum_list + HedTagSummary._get_tag_list(result, indent=indent)
return "\n".join(sum_list)
@staticmethod
def _get_individual_string(result, indent=BaseSummary.DISPLAY_INDENT):
"""Return a string with the summary for an individual tabular file.
Parameters:
result (dict): Dictionary of summary information for a particular tabular file.
indent (str): String of blanks used as the amount to indent for readability.
Returns:
str: Formatted string suitable for saving in a file or printing.
"""
sum_list = [f"Total events={result.get('Total events', 0)}"]
sum_list = sum_list + HedTagSummary._get_tag_list(result, indent=indent)
return "\n".join(sum_list)
@staticmethod
def _tag_details(tags):
"""Return a list of strings with the tag details.
Parameters:
tags (list): List of tags to summarize.
Returns:
list: Each entry has the summary details for a tag.
"""
tag_list = []
for tag in tags:
tag_list.append(f"{tag['tag']}[{tag['events']},{len(tag['files'])}]")
return tag_list
@staticmethod
def _get_tag_list(result, indent=BaseSummary.DISPLAY_INDENT):
"""Return a list lines to be output to summarize the tags as organized in the result.
Parameters:
result (dict): Dictionary with the results organized under key "Specifics".
indent (str): Spaces to indent each line.
Returns:
list: Each entry is a string representing a line to be printed.
"""
tag_info = result["Specifics"]
sum_list = [f"\n{indent}Main tags[events,files]:"]
for category, tags in tag_info["Main tags"].items():
sum_list.append(f"{indent}{indent}{category}:")
if tags:
sum_list.append(f"{indent}{indent}{indent}{' '.join(HedTagSummary._tag_details(tags))}")
if tag_info["Other tags"]:
sum_list.append(f"{indent}Other tags[events,files]:")
sum_list.append(f"{indent}{indent}{' '.join(HedTagSummary._tag_details(tag_info['Other tags']))}")
return sum_list
@staticmethod
def _get_details(key_list, template, verbose=False):
"""Organized a tag information from a list based on the template.
Parameters:
key_list (list): List of information to be organized based on the template.
template (dict): An input template derived from the input parameters.
verbose (bool): If False (the default) output minimal information about the summary.
"""
key_details = []
for item in key_list:
for tag_cnt in template[item.casefold()]:
key_details.append(tag_cnt.get_info(verbose=verbose))
return key_details