"""Classes for managing counts of tags for one type tag such as Condition-variable or Task."""
[docs]
class HedTypeCount:
"""Manager of the counts of tags for one type tag such as Condition-variable or Task.
Parameters:
type_value (str): The value of the variable to be counted.
type_tag (str): The type of variable.
Examples:
HedTypeCounts('SymmetricCond', 'condition-variable') keeps counts of Condition-variable/Symmetric.
"""
[docs]
def __init__(self, type_value, type_tag, file_name=None):
self.type_value = type_value
self.type_tag = type_tag.casefold()
self.direct_references = 0
self.total_events = 0
self.events = 0
self.files = {}
if file_name:
self.files[file_name] = ""
self.events_with_multiple_refs = 0
self.max_refs_per_event = 0
self.level_counts = {}
[docs]
def update(self, type_sum, file_id):
"""Update the counts from a HedTypeValues.
Parameters:
type_sum (dict): Information about the contents for a particular data file.
file_id (str or None): Name of the file associated with the counts.
"""
self.direct_references += type_sum["direct_references"]
self.total_events += type_sum["total_events"]
self.events += type_sum["events"]
if file_id:
self.files[file_id] = ""
for file in type_sum.get("files", {}).keys():
self.files[file] = ""
self.events_with_multiple_refs += type_sum["events_with_multiple_refs"]
self.max_refs_per_event = max(self.max_refs_per_event, type_sum["max_refs_per_event"])
self._update_levels(type_sum.get("level_counts", {}))
[docs]
def to_dict(self):
"""Return count information as a dictionary."""
return {
"type_value": self.type_value,
"type_tag": self.type_tag,
"direct_references": self.direct_references,
"total_events": self.total_events,
"events": self.events,
"files": self.files,
"events_with_multiple_refs": self.events_with_multiple_refs,
"max_refs_per_event": self.max_refs_per_event,
"level_counts": self.level_counts,
}
def _update_levels(self, level_dict):
"""Helper for updating counts in a level dictionary.
Parameters:
level_dict (dict): A dictionary of level count information.
"""
for key, item in level_dict.items():
if key not in self.level_counts:
self.level_counts[key] = {"files": 0, "events": 0, "tags": "", "description": ""}
level_counts = self.level_counts[key]
if not isinstance(item, dict):
level_counts["files"] = level_counts["files"] + 1
level_counts["events"] = level_counts["events"] + item
continue
level_counts["files"] = level_counts["files"] + item["files"]
level_counts["events"] = level_counts["events"] + item["events"]
if not level_counts["tags"]:
level_counts["tags"] = item["tags"]
if not level_counts["description"]:
level_counts["description"] = item["description"]
[docs]
def get_summary(self):
"""Return the summary of one value of one type tag.
Returns:
dict: Count information for one tag of one type.
"""
summary = {
"type_value": self.type_value,
"type_tag": self.type_tag,
"levels": len(self.level_counts.keys()),
"direct_references": self.direct_references,
"total_events": self.total_events,
"events": self.events,
"events_with_multiple_refs": self.events_with_multiple_refs,
"max_refs_per_event": self.max_refs_per_event,
"files": list(self.files.keys()),
}
if self.level_counts:
summary["level_counts"] = self.level_counts
return summary
[docs]
class HedTypeCounts:
"""Manager for summaries of tag counts for columnar files."""
[docs]
def __init__(self, name, type_tag):
self.name = name
self.type_tag = type_tag
self.files = {}
self.total_events = 0
self.type_dict = {}
[docs]
def update_summary(self, type_sum, total_events=0, file_id=None):
"""Update this summary based on the type variable map.
Parameters:
type_sum (dict): Contains the information about the value of a type.
total_events (int): Total number of events processed.
file_id (str): Unique identifier for the associated file.
"""
for type_val, type_counts in type_sum.items():
if type_val not in self.type_dict:
self.type_dict[type_val] = HedTypeCount(type_val, self.type_tag, file_id)
val_counts = self.type_dict[type_val]
val_counts.update(type_counts, file_id)
self.files[file_id] = ""
self.total_events = self.total_events + total_events
[docs]
def add_descriptions(self, type_defs):
"""Update this summary based on the type variable map.
Parameters:
type_defs (HedTypeDefs): Contains the information about the value of a type.
"""
for type_val, type_count in self.type_dict.items():
if type_val not in type_defs.type_map:
continue
for level in type_defs.type_map[type_val]:
if level not in type_count.level_counts:
continue
level_dict = type_defs.def_map[level]
type_count.level_counts[level]["tags"] = level_dict["tags"]
type_count.level_counts[level]["description"] = level_dict["description"]
[docs]
def update(self, counts):
"""Update count information based on counts in another HedTypeCounts.
Parameters:
counts (HedTypeCounts): Information to use in the update.
"""
self.total_events = self.total_events + counts.total_events
for key, count in counts.type_dict.items():
if key not in self.type_dict:
self.type_dict[key] = HedTypeCount(count.type_value, count.type_tag, None)
this_count = self.type_dict[key]
this_count.update(count.to_dict(), None)
for file_id in counts.files.keys():
self.files[file_id] = ""
[docs]
def get_summary(self):
"""Return the information in the manager as a dictionary.
Returns:
dict: Dict with keys 'name', 'type_tag', 'files', 'total_events', and 'details'.
"""
details = {}
for type_value, count in self.type_dict.items():
details[type_value] = count.get_summary()
return {
"name": str(self.name),
"type_tag": self.type_tag,
"files": list(self.files.keys()),
"total_events": self.total_events,
"details": details,
}