wuenlp_tools.models.sentiment.harrymotions_aggregation

  1from __future__ import annotations
  2
  3from collections import Counter
  4from dataclasses import dataclass, field
  5from enum import Enum
  6
  7from loguru import logger
  8from wuenlp.impl.UIMANLPStructs import UIMADocument
  9from wuenlp.impl.uima import UIMAInteraction, UIMASpan, UIMASystemScene
 10
 11from wuenlp_tools.pipeline import PipelineProcessor, PipelineStep
 12
 13
 14class HarryMotionsCategoricalAggregator(str, Enum):
 15    MODE = "mode"
 16    DISTRIBUTION = "distribution"
 17
 18
 19class HarryMotionsNumericReducer(str, Enum):
 20    MEAN = "mean"
 21    ABS_MEAN = "abs_mean"
 22    MIN = "min"
 23    MAX = "max"
 24    NEG_RATIO = "neg_ratio"
 25    COUNT = "count"
 26
 27
 28class HarryMotionsEmptyPolicy(str, Enum):
 29    ZERO = "zero"
 30    NONE = "none"
 31
 32
 33@dataclass(frozen=True)
 34class HarryMotionsAggregationConfig:
 35    reducers: tuple[HarryMotionsNumericReducer, ...] = (HarryMotionsNumericReducer.MEAN,)
 36    categorical_aggregator: HarryMotionsCategoricalAggregator = HarryMotionsCategoricalAggregator.MODE
 37    empty_policy: HarryMotionsEmptyPolicy = HarryMotionsEmptyPolicy.ZERO
 38
 39
 40def _hm_empty_value(policy: HarryMotionsEmptyPolicy) -> float | None:
 41    return 0.0 if policy == HarryMotionsEmptyPolicy.ZERO else None
 42
 43
 44def _hm_setting_reducer_key(setting: str, reducer: str) -> str:
 45    return f"harrymotions_{setting}_agg_{reducer}"
 46
 47
 48def _hm_setting_categorical_key(setting: str, suffix: str) -> str:
 49    return f"harrymotions_{setting}_agg_{suffix}"
 50
 51
 52@dataclass
 53class HarryMotionsAggregationProcessor(PipelineProcessor):
 54    config: HarryMotionsAggregationConfig = field(default_factory=HarryMotionsAggregationConfig)
 55
 56    @staticmethod
 57    def _parse_harrymotions_features(interactions: list[UIMAInteraction]) -> dict[str, dict[str, str]]:
 58        settings: dict[str, dict[str, str]] = {}
 59        suffixes = ("sentiment_polarity", "sentiment_bin", "label")
 60        for inter in interactions:
 61            for key in inter.additional_features.keys():
 62                for suffix in suffixes:
 63                    marker = f"_{suffix}"
 64                    if key.startswith("harrymotions_") and key.endswith(marker):
 65                        setting = key[len("harrymotions_"):-len(marker)]
 66                        setting_map = settings.setdefault(setting, {})
 67                        setting_map[suffix] = key
 68                        break
 69        return settings
 70
 71    @staticmethod
 72    def _reduce_numeric(values: list[float], reducer: HarryMotionsNumericReducer) -> float:
 73        if reducer == HarryMotionsNumericReducer.MEAN:
 74            return sum(values) / len(values)
 75        if reducer == HarryMotionsNumericReducer.ABS_MEAN:
 76            return sum(abs(v) for v in values) / len(values)
 77        if reducer == HarryMotionsNumericReducer.MIN:
 78            return min(values)
 79        if reducer == HarryMotionsNumericReducer.MAX:
 80            return max(values)
 81        if reducer == HarryMotionsNumericReducer.NEG_RATIO:
 82            return sum(1 for v in values if v < 0) / len(values)
 83        if reducer == HarryMotionsNumericReducer.COUNT:
 84            return float(len(values))
 85        raise ValueError(
 86            "Unsupported reducer. Use one of: mean, abs_mean, min, max, neg_ratio, count"
 87        )
 88
 89    def __call__(self, doc: UIMADocument, unit_type: type[UIMASpan] | None, overwrite: bool = False, **kwargs) -> UIMADocument:
 90        if unit_type is None:
 91            raise ValueError("HarryMotionsAggregationProcessor requires a non-None unit_type.")
 92        interactions = list(doc.interactions)
 93        if not interactions:
 94            logger.warning("No interactions found; skipping HarryMotions aggregation.")
 95            return doc
 96        units = list(doc.get_annos_of_type(unit_type))
 97        if not units:
 98            logger.warning(f"No units found for {unit_type.__name__}; skipping HarryMotions aggregation.")
 99            return doc
100
101        settings = self._parse_harrymotions_features(interactions)
102        if not settings:
103            logger.warning("No HarryMotions interaction features found; skipping aggregation.")
104            return doc
105
106        for unit in units:
107            unit_interactions = list(unit.overlapping(UIMAInteraction))
108            for setting, feature_map in settings.items():
109                numeric_key = feature_map.get("sentiment_polarity") or feature_map.get("sentiment_bin")
110                label_key = feature_map.get("label")
111
112                if numeric_key:
113                    numeric_values: list[float] = []
114                    for inter in unit_interactions:
115                        value = inter.additional_features.get(numeric_key)
116                        try:
117                            if value is not None:
118                                numeric_values.append(float(value))
119                        except (TypeError, ValueError):
120                            continue
121                    for reducer in self.config.reducers:
122                        feature_name = _hm_setting_reducer_key(setting, reducer.value)
123                        if not overwrite and feature_name in unit.additional_features:
124                            continue
125                        if numeric_values:
126                            unit.additional_features[feature_name] = self._reduce_numeric(numeric_values, reducer)
127                        else:
128                            unit.additional_features[feature_name] = _hm_empty_value(self.config.empty_policy)
129
130                if label_key:
131                    labels = [
132                        str(inter.additional_features.get(label_key))
133                        for inter in unit_interactions
134                        if inter.additional_features.get(label_key)
135                    ]
136                    if self.config.categorical_aggregator == HarryMotionsCategoricalAggregator.MODE:
137                        mode_key = _hm_setting_categorical_key(setting, "label_mode")
138                        if overwrite or mode_key not in unit.additional_features:
139                            if labels:
140                                counts = Counter(labels)
141                                mode_label = sorted(counts.items(), key=lambda item: (-item[1], item[0]))[0][0]
142                                unit.additional_features[mode_key] = mode_label
143                            else:
144                                unit.additional_features[mode_key] = None
145                    elif self.config.categorical_aggregator == HarryMotionsCategoricalAggregator.DISTRIBUTION:
146                        base_key = _hm_setting_categorical_key(setting, "label_prob")
147                        counts = Counter(labels)
148                        total = sum(counts.values())
149                        for label, count in counts.items():
150                            feature_name = f"{base_key}_{label}"
151                            if overwrite or feature_name not in unit.additional_features:
152                                unit.additional_features[feature_name] = count / total if total > 0 else 0.0
153                    else:
154                        raise ValueError(f"Unknown categorical aggregator: {self.config.categorical_aggregator}")
155
156        return doc
157
158
159def HarryMotionsAggregationStep(
160        unit_type: type[UIMASpan] = UIMASystemScene,
161        config: HarryMotionsAggregationConfig | None = None,
162        name: str = "HarryMotions Aggregation",
163) -> PipelineStep[HarryMotionsAggregationProcessor]:
164    processor = HarryMotionsAggregationProcessor(config=config or HarryMotionsAggregationConfig())
165    return PipelineStep(
166        name=name,
167        processor=processor,
168        unit_type=unit_type,
169        modified_types=[unit_type],
170        added_additional_features=[
171            "harrymotions_*_agg_*",
172        ],
173    )