wuenlp_tools.visualize.plot

  1from pathlib import Path
  2from typing import Type
  3
  4import numpy as np
  5import pandas as pd
  6import plotly.express as px
  7from loguru import logger
  8
  9from wuenlp.impl.UIMANLPStructs import UIMADocument, UIMAAnnotation, UIMASystemScene
 10
 11# Function to format text with newlines
 12from wuenlp_tools.utils.summarize import summarize_doc
 13
 14
 15def format_text_scenes(segment: UIMAAnnotation, line_length=20, max_lines=50, summary=True):
 16    # Split long texts into lines
 17    text = segment.additional_features["llama_summary"] if summary else segment.text
 18    split = text.split(" ")
 19    lines = [' '.join(split[i:i + line_length]) for i in range(0, len(split), line_length)]
 20
 21    if len(lines) > max_lines:
 22        lines = lines[:max_lines // 2] + [f"[{len(lines) - max_lines} more lines]"] + lines[-max_lines // 2:]
 23    return "<br>".join(lines)
 24
 25
 26def format_text_danger(segment: UIMAAnnotation, line_length=20, max_lines=75, summary=True):
 27    return f"{format_text_scenes(segment, line_length, max_lines, summary)}<br>{segment.additional_features['danger_score']}; {segment.additional_features['danger_reason']}"
 28
 29
 30def _signed_power_mean(values: list[float], power: float) -> float:
 31    if not values:
 32        return 0.0
 33    transformed = [np.sign(value) * (abs(value) ** power) for value in values]
 34    mean_transformed = sum(transformed) / len(transformed)
 35    return float(np.sign(mean_transformed) * (abs(mean_transformed) ** (1.0 / power)))
 36
 37
 38def _smooth_series(
 39    sentiments: list[float],
 40    smooth: int,
 41    method: str = "moving",
 42    signed_power: float = 2.0,
 43    bilateral_sigma_t: float = 2.5,
 44    bilateral_sigma_x: float = 0.08,
 45) -> list[float]:
 46    if smooth <= 0:
 47        return sentiments
 48    if method not in {"moving", "signed-power", "bilateral"}:
 49        raise ValueError(f"Unknown smoothing method: {method}")
 50    if method == "signed-power" and signed_power <= 0:
 51        raise ValueError("signed_power must be > 0")
 52    if method == "bilateral" and (bilateral_sigma_t <= 0 or bilateral_sigma_x <= 0):
 53        raise ValueError("bilateral sigma parameters must be > 0")
 54
 55    arr = np.asarray(sentiments, dtype=float)
 56    num_sentiments = len(arr)
 57    smoothed_sentiments = []
 58    for i in range(num_sentiments):
 59        start = max(0, i - smooth)
 60        end = min(num_sentiments, i + smooth + 1)
 61        window = arr[start:end]
 62        if method == "moving":
 63            smoothed = float(window.mean())
 64        elif method == "signed-power":
 65            smoothed = _signed_power_mean(window.tolist(), signed_power)
 66        else:
 67            # Bilateral: weight by temporal distance and value similarity to the center point.
 68            js = np.arange(start, end)
 69            dt = (i - js).astype(float)
 70            dx = (arr[i] - window).astype(float)
 71            weights = np.exp(-(dt * dt) / (2.0 * bilateral_sigma_t ** 2) - (dx * dx) / (2.0 * bilateral_sigma_x ** 2))
 72            weight_sum = float(weights.sum())
 73            smoothed = float((weights * window).sum() / weight_sum) if weight_sum > 0 else float(arr[i])
 74        smoothed_sentiments.append(smoothed)
 75    return smoothed_sentiments
 76
 77
 78def plot_sentiment_interactive(doc: UIMADocument, segment_type: Type[UIMAAnnotation] = UIMASystemScene, smooth: int = 0,
 79                               line_length=20, feature_name: str = "llama_sentiment",
 80                               summary: bool = False, call_summarizer_if_necessary: bool = False,
 81                               format_text_func=format_text_scenes, output_path: Path = None,
 82                               disable_markers: bool = False, doc_name: str = None,
 83                               segment_type_clean_name: str = None, show_title: bool = True,
 84                               y_label: str = 'Sentiment', smooth_only: bool = False,
 85                               smoothing_method: str = "moving", smoothing_signed_power: float = 2.0,
 86                               bilateral_sigma_t: float = 2.5, bilateral_sigma_x: float = 0.08):
 87    sentiments = []
 88    texts = []
 89    text_lengths = []
 90    keywords = []
 91
 92    segments = doc._get_annos_of_type(segment_type)
 93    if not segments:
 94        raise ValueError(f"No {segment_type.__name__} found in {doc.path.stem}")
 95
 96    if call_summarizer_if_necessary and not segments[0].additional_features.get('llama_summary', None):
 97        logger.info("Summarising segments")
 98        doc = summarize_doc(doc, unit_type=segment_type)
 99
100    # Extracting scenes (assuming 'document' is UIMADocument and has a method to get system scenes)
101    for segment in segments:
102        llama_sentiment = segment.additional_features.get(feature_name, None)
103        if llama_sentiment is not None:
104            sentiments.append(llama_sentiment)
105
106            formatted_text = format_text_func(segment, line_length, summary=summary)
107            texts.append(formatted_text)
108            keyword = "<br>".join(segment.additional_features.get('llama_keywords', []))
109            logger.info(f"Keywords: {keyword}")
110            keywords.append(keyword)
111            text_lengths.append(len(segment.text))  # Record the length of the original text
112
113    smoothed_sentiments = []
114    if smooth:
115        smoothed_sentiments = _smooth_series(
116            sentiments,
117            smooth=smooth,
118            method=smoothing_method,
119            signed_power=smoothing_signed_power,
120            bilateral_sigma_t=bilateral_sigma_t,
121            bilateral_sigma_x=bilateral_sigma_x,
122        )
123        if smooth_only:
124            sentiments = smoothed_sentiments
125
126    # Create a DataFrame for Plotly
127    df = pd.DataFrame({'Sentiment': sentiments, 'Scene': texts, 'Text Length': text_lengths, 'Keywords': keywords})
128    if smooth and not smooth_only:
129        df['Smoothed Sentiment'] = smoothed_sentiments
130
131    # Calculate the mean and standard deviation
132    mean_sentiment = df['Sentiment'].mean()
133    std_sentiment = df['Sentiment'].std()
134
135    # Define threshold for outliers (for example: mean ± 1.5 * std deviation)
136    threshold_upper = mean_sentiment + 1.5 * std_sentiment
137    threshold_lower = mean_sentiment - 1.5 * std_sentiment
138
139    size_scale = 1000  # Adjust this scaling factor for marker sizes
140    df['Marker Size'] = np.array(df['Text Length']) / size_scale
141
142    # Create an interactive plot
143    doc_name = doc_name or (doc.path.stem if doc.path else "Unknown")
144    segment_type_clean_name = segment_type_clean_name or segment_type.uima_type.split(".")[-1].replace("UIMA", "")
145    base_color = ['lightgray'] if smooth and not smooth_only else None
146
147    fig = px.line(
148        df,
149        x=df.index,
150        y='Sentiment',
151        hover_data=['Scene'],
152        labels={'Sentiment': y_label, 'Scene': 'Scene Text', 'index': segment_type_clean_name.title()},
153        title=f'{y_label} per {segment_type_clean_name}<br>{doc_name}' if show_title else None,
154        color_discrete_sequence=base_color
155    )
156
157    # 2. If 'smooth' is True, layer the smoothed line on top
158    if smooth and not smooth_only:
159        logger.info(
160            f"Adding smoothed line with window size {smooth}, "
161            f"method={smoothing_method}, signed_power={smoothing_signed_power}, "
162            f"bilateral_sigma_t={bilateral_sigma_t}, bilateral_sigma_x={bilateral_sigma_x}"
163        )
164        # Rename the first trace so the legend makes sense (Optional)
165        fig.data[0].name = 'Raw Sentiment'
166        fig.data[0].showlegend = True
167
168        # Add the Smoothed column
169        fig.add_scatter(
170            x=df.index,
171            y=df['Smoothed Sentiment'],
172            mode='lines',
173            name='Smoothed Sentiment',
174            line=dict(color='blue', width=3),  # Make it thicker and prominent
175            hovertext=df['Scene'],
176            hovertemplate='%{hovertext}<extra></extra>',
177        )
178
179    if len(sentiments) > 1000:
180        logger.warning("Trajectory is very long. Omitting additional annotations.")
181        fig.show()
182        write_to_path(fig, output_path)
183        return
184
185    fig.update_traces(line=dict(width=2),
186                      hovertemplate='%{hovertext}<extra></extra>',
187                      mode='lines+markers+text',
188                      marker=dict(size=df['Marker Size'], opacity=0.7),
189                      hovertext=df['Scene'],
190                      )
191
192    if not disable_markers:
193        for i in range(len(df)):
194            if df['Sentiment'].iloc[i] > threshold_upper:
195                # High outlier annotation
196                fig.add_annotation(
197                    x=df.index[i],
198                    y=df['Sentiment'].iloc[i] + 0.01,  # Adjust for space above
199                    text=df['Keywords'].iloc[i],
200                    showarrow=True,
201                    arrowhead=2,
202                    ax=0,
203                    ay=-30,  # Arrow pointing downwards
204                    textangle=0,  # Straight text for high points
205                    font=dict(size=10)
206                )
207            elif df['Sentiment'].iloc[i] < threshold_lower:
208                # Low outlier annotation
209                fig.add_annotation(
210                    x=df.index[i],
211                    y=df['Sentiment'].iloc[i] - 0.01,  # Adjust for space below
212                    text=df['Keywords'].iloc[i],
213                    showarrow=True,
214                    arrowhead=2,
215                    ax=0,
216                    ay=30,  # Arrow pointing upwards
217                    textangle=0,  # Straight text for low points
218                    font=dict(size=10)
219                )
220
221    fig.show()
222    write_to_path(fig, output_path)
223
224
225def write_to_path(fig, output_path):
226    if output_path:
227        output_path.parent.mkdir(parents=True, exist_ok=True)
228        if output_path.suffix == ".html":
229            fig.write_html(output_path)
230        else:
231            fig.write_image(str(output_path))