"""
Visualization Module - Charts and Metrics
"""
from typing import Any
import pandas as pd
import plotly.express as px
import plotly.graph_objects as go
import streamlit as st
def calculate_summary_metrics(df: pd.DataFrame) -> dict[str, Any]:
"""
Calculate summary metrics for display
Args:
df: Input DataFrame
Returns:
Dictionary of metrics
"""
if df.empty:
return {
'Total Events': '0',
'Most Recent Change': 'N/A',
'Most Active Profile': 'N/A',
'Unique Profiles': '0',
}
# Total events
total_events = len(df)
# Most recent change - shorter format
most_recent = df['event_time'].max().strftime('%Y-%m-%d %H:%M')
# Most active profile (by profile_id)
profile_counts = df['profile_id'].value_counts()
if not profile_counts.empty:
most_active_profile_id = profile_counts.idxmax()
most_active_profile_type = df[df['profile_id'] == most_active_profile_id][
'profile_type'
].iloc[0]
most_active = f'{most_active_profile_type} ({most_active_profile_id})'
else:
most_active = 'N/A'
# Unique profiles
unique_profiles = df['profile_uuid'].nunique()
return {
'Total Events': f'{total_events:,}',
'Most Recent Change': most_recent,
'Most Active Profile': most_active,
'Unique Profiles': f'{unique_profiles:,}',
}
def render_visualizations(df: pd.DataFrame):
"""
Render all visualization charts
Args:
df: Input DataFrame
"""
if df.empty:
st.warning('No data available for visualization')
return
# Create three columns for charts
col1, col2 = st.columns(2)
with col1:
render_operations_bar_chart(df)
st.write('') # Spacing
render_profile_type_chart(df)
with col2:
render_target_type_chart(df)
def render_operations_timeline(df: pd.DataFrame):
"""
Render timeline chart showing operations over time
Args:
df: Input DataFrame
"""
st.markdown('**📈 Operations Timeline**')
# Group by date and operation
timeline_data = (
df.groupby([df['event_time'].dt.date, 'operation']).size().reset_index(name='count')
)
timeline_data.columns = ['date', 'operation', 'count']
# Create line chart with Plotly
fig = px.line(
timeline_data,
x='date',
y='count',
color='operation',
title='',
labels={'date': 'Date', 'count': 'Number of Events', 'operation': 'Operation'},
color_discrete_map={'created': '#28a745', 'updated': '#007bff', 'deleted': '#dc3545'},
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='right', x=1),
hovermode='x unified',
)
st.plotly_chart(fig, use_container_width=True)
@st.cache_data(show_spinner=False)
def aggregate_operation_counts(df_hash: int, operation_series: tuple) -> pd.DataFrame:
"""
Aggregate operation counts (cached for performance)
Args:
df_hash: Hash of dataframe for cache key
operation_series: Tuple of operation values for aggregation
Returns:
DataFrame with operation counts
"""
operation_counts = pd.Series(operation_series).value_counts().reset_index()
operation_counts.columns = ['operation', 'count']
return operation_counts
def render_operations_bar_chart(df: pd.DataFrame):
"""
Render bar chart showing operation counts by type
Uses cached aggregations for performance
Args:
df: Input DataFrame
"""
st.markdown('**📊 Operation Counts**')
# Aggregate with caching (pass tuple for hashability)
df_hash = hash(tuple(df.index))
operation_counts = aggregate_operation_counts(df_hash, tuple(df['operation'].tolist()))
# Create bar chart with Plotly (small aggregated data)
fig = px.bar(
operation_counts,
x='operation',
y='count',
title='',
labels={'operation': 'Operation Type', 'count': 'Number of Events'},
color='operation',
color_discrete_map={'created': '#28a745', 'updated': '#007bff', 'deleted': '#dc3545'},
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
showlegend=False,
xaxis_title='',
yaxis_title='Count',
)
st.plotly_chart(fig, use_container_width=True)
@st.cache_data(show_spinner=False)
def aggregate_profile_type_counts(df_hash: int, profile_type_series: tuple) -> pd.DataFrame:
"""
Aggregate profile type counts (cached for performance)
Args:
df_hash: Hash of dataframe for cache key
profile_type_series: Tuple of profile_type values for aggregation
Returns:
DataFrame with profile_type counts
"""
profile_counts = pd.Series(profile_type_series).value_counts().reset_index()
profile_counts.columns = ['profile_type', 'count']
return profile_counts
def render_profile_type_chart(df: pd.DataFrame):
"""
Render bar chart showing events by profile type
Uses cached aggregations for performance
Args:
df: Input DataFrame
"""
st.markdown('**👤 Events by Profile Type**')
# Aggregate with caching (pass tuple for hashability)
df_hash = hash(tuple(df.index))
profile_counts = aggregate_profile_type_counts(df_hash, tuple(df['profile_type'].tolist()))
# Create horizontal bar chart with Plotly (small aggregated data)
fig = px.bar(
profile_counts,
x='count',
y='profile_type',
title='',
labels={'profile_type': 'Profile Type', 'count': 'Number of Events'},
orientation='h',
color='count',
color_continuous_scale='Blues',
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
showlegend=False,
xaxis_title='Count',
yaxis_title='',
)
st.plotly_chart(fig, use_container_width=True)
@st.cache_data(show_spinner=False)
def aggregate_tenant_type_counts(df_hash: int, tenant_type_series: tuple) -> pd.DataFrame:
"""
Aggregate tenant type counts (cached for performance)
Args:
df_hash: Hash of dataframe for cache key
tenant_type_series: Tuple of tenant_type values for aggregation
Returns:
DataFrame with tenant_type counts
"""
resource_counts = pd.Series(tenant_type_series).value_counts().reset_index()
resource_counts.columns = ['tenant_type', 'count']
return resource_counts
def render_target_type_chart(df: pd.DataFrame):
"""
Render pie chart showing events by resource type
Uses cached aggregations for performance
Args:
df: Input DataFrame
"""
st.markdown('**🎯 Events by Tenant Type**')
# Aggregate with caching (pass tuple for hashability)
df_hash = hash(tuple(df.index))
resource_counts = aggregate_tenant_type_counts(df_hash, tuple(df['tenant_type'].tolist()))
# Create pie chart with Plotly (small aggregated data)
fig = px.pie(
resource_counts,
values='count',
names='tenant_type',
title='',
color_discrete_sequence=px.colors.qualitative.Set3,
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
showlegend=True,
legend=dict(orientation='v', yanchor='middle', y=0.5, xanchor='left', x=1.05),
)
fig.update_traces(textposition='inside', textinfo='percent+label')
st.plotly_chart(fig, use_container_width=True)
def render_hourly_activity_heatmap(df: pd.DataFrame):
"""
Render heatmap showing activity by day of week and hour
Args:
df: Input DataFrame
"""
st.markdown('**🕒 Activity Heatmap (Day of Week vs Hour)**')
# Use SQL-provided column names (event_day_of_week, event_hour)
if 'event_day_of_week' not in df.columns or 'event_hour' not in df.columns:
st.info('Heatmap data not available')
return
# Create pivot table using SQL column names
heatmap_data = df.groupby(['event_day_of_week', 'event_hour']).size().reset_index(name='count')
# Order days of week
day_order = ['Monday', 'Tuesday', 'Wednesday', 'Thursday', 'Friday', 'Saturday', 'Sunday']
heatmap_data['event_day_of_week'] = pd.Categorical(
heatmap_data['event_day_of_week'], categories=day_order, ordered=True
)
# Create pivot table
pivot_data = heatmap_data.pivot(
index='event_day_of_week', columns='event_hour', values='count'
).fillna(0)
# Create heatmap with Plotly
fig = go.Figure(
data=go.Heatmap(
z=pivot_data.values,
x=pivot_data.columns,
y=pivot_data.index,
colorscale='Blues',
showscale=True,
)
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
xaxis_title='Hour of Day',
yaxis_title='Day of Week',
)
st.plotly_chart(fig, use_container_width=True)
def render_activity_by_user(df: pd.DataFrame):
"""
Render chart showing activity by user (changed_by)
Args:
df: Input DataFrame
"""
st.markdown('**👥 Activity by User**')
# Count by user
user_counts = df['changed_by'].value_counts().head(10).reset_index()
user_counts.columns = ['user', 'count']
# Create horizontal bar chart
fig = px.bar(
user_counts,
x='count',
y='user',
title='',
labels={'user': 'User', 'count': 'Number of Events'},
orientation='h',
color='count',
color_continuous_scale='Greens',
)
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=20, b=20),
showlegend=False,
xaxis_title='Count',
yaxis_title='',
)
st.plotly_chart(fig, use_container_width=True)
def render_tenant_timeline_chart(
df: pd.DataFrame, tenant_type: str, tenant_id: str, full_df: pd.DataFrame
):
"""
Render interactive timeline chart for a specific tenant with click support
Args:
df: Events DataFrame
tenant_type: Type of tenant to filter
tenant_id: ID of tenant to filter
full_df: Full filtered dataframe (for mapping clicks to row indices)
"""
# Filter events for this resource (filtering creates a new DataFrame, no need for .copy())
filtered = df[
(df['tenant_type'] == tenant_type) & (df['tenant_id'].astype(str) == str(tenant_id))
]
if filtered.empty:
st.warning(f'No events found for {tenant_type} ID: {tenant_id}')
return
# Sort chronologically
filtered = filtered.sort_values('event_time', ascending=True)
# Add row indices from the full dataframe so we can map clicks back
full_df_indexed = full_df.reset_index(drop=True)
filtered['original_row_idx'] = filtered.index.map(
lambda idx: full_df_indexed.index[full_df_indexed.index == idx].tolist()[0]
if idx in full_df_indexed.index
else -1
)
# Prepare hover data
filtered['hover_text'] = filtered.apply(
lambda row: f'Profile: {row["profile_type"]} (ID: {row["profile_id"]})
'
f'Operation: {row["operation"]}
'
f'Changed By: {row["changed_by"]}
'
f'Relationship: {row.get("relationship_type", "N/A")}',
axis=1,
)
# Add jitter to y-axis to prevent overlapping points
# Use a small random offset so events at the same time are visible
import numpy as np
np.random.seed(42) # Consistent jitter for same data
jitter_amount = 0.15
filtered['timeline_y'] = np.random.uniform(-jitter_amount, jitter_amount, size=len(filtered))
# Create scatter plot for timeline - all events on same axis
fig = px.scatter(
filtered,
x='event_time',
y='timeline_y',
color='operation',
title='', # No title - header is shared with table
labels={
'event_time': 'Event Time',
'timeline_y': '',
},
color_discrete_map={'created': '#28a745', 'updated': '#007bff', 'deleted': '#dc3545'},
custom_data=['hover_text', 'original_row_idx'],
)
# Customize hover template to show clean tooltip without field labels
fig.update_traces(
hovertemplate='%{x|%Y-%m-%d %H:%M:%S}
%{customdata[0]}'
)
# Update layout
fig.update_layout(
height=300,
margin=dict(l=20, r=20, t=40, b=20),
hovermode='closest',
showlegend=True,
legend=dict(orientation='h', yanchor='bottom', y=1.02, xanchor='right', x=1),
yaxis=dict(
showticklabels=False, # Hide y-axis labels
showgrid=False, # Hide grid lines
zeroline=False, # Hide zero line
range=[-0.5, 0.5], # Set range to accommodate jitter
),
)
# Update markers for better visibility
fig.update_traces(marker=dict(size=12, line=dict(width=2, color='DarkSlateGrey')))
# Enable click selection and capture the event
event = st.plotly_chart(
fig,
use_container_width=True,
on_select='rerun',
selection_mode='points',
key=f'timeline_{tenant_type}_{tenant_id}',
)
# Handle click event
if event and event.selection and event.selection.points:
# Get the first clicked point
clicked_point = event.selection.points[0]
# Get the row index from custom data
if 'customdata' in clicked_point and len(clicked_point['customdata']) > 1:
row_idx = int(clicked_point['customdata'][1])
if row_idx >= 0:
st.session_state['selected_event_row'] = row_idx