""" 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