Coverage for src/common/telemetry_analysis.py: 6%
761 statements
« prev ^ index » next coverage.py v7.8.2, created at 2025-05-25 13:51 -0400
1# Copyright (c) CloudZero - ALL RIGHTS RESERVED - PROPRIETARY AND CONFIDENTIAL
2# Unauthorized copying of this file and/or project, via any medium is strictly prohibited.
3# Direct all questions to legal@cloudzero.com
5"""
6CloudZero Telemetry Analysis Module
8Interactive analysis to help users map CSV data to CloudZero telemetry records.
9Supports both Allocation Telemetry and Unit Cost Telemetry mapping.
10"""
12import json
13import polars as pl
14from pathlib import Path
15from typing import Dict, List, Any
16from datetime import datetime
17import re
19from rich.table import Table
20from rich import box
22# Import utility modules
23import sys
24sys.path.append(str(Path(__file__).parent.parent))
26from utils.console_helpers import (
27 console, print_success, print_analysis_header, prompt_with_ctrl_c_reminder, int_prompt_with_ctrl_c_reminder,
28 confirm_with_ctrl_c_reminder
29)
30from rich.prompt import Prompt
31from utils.table_helpers import (
32 create_selection_table, create_multi_select_table, add_selection_row, add_multi_select_row,
33 create_basic_table, TableColumn, create_temporal_selection_table, create_temporal_summary_table
34)
35from utils.dataframe_helpers import (
36 get_column_summary
37)
40class TelemetryAnalyzer:
41 """Analyzes CSV data to identify CloudZero telemetry mapping opportunities."""
43 def __init__(
44 self,
45 df: pl.DataFrame,
46 input_file,
47 costformation_file=None,
48 config_file=None,
49 ):
50 self.df = df
51 self.input_file = Path(input_file) if input_file else None
52 self.costformation_file = (
53 Path(costformation_file) if costformation_file else None
54 )
55 self.config_file = Path(config_file) if config_file else None
56 self.costformation_dimensions = []
57 self.config = {}
58 self.analysis_results = {
59 "input_file": str(input_file),
60 "analyzed_at": datetime.now().isoformat(),
61 "row_count": len(df),
62 "column_count": len(df.columns),
63 "telemetry_type": None,
64 "temporal_analysis": {},
65 "column_analysis": {},
66 "selected_mapping": {},
67 "sample_records": [],
68 }
70 # Load costformation if provided
71 if self.costformation_file and self.costformation_file.exists():
72 self._load_costformation()
74 # Load config if provided
75 if self.config_file and self.config_file.exists():
76 self._load_config()
78 def _load_costformation(self):
79 """Load and parse costformation.yaml file for dimension suggestions."""
80 try:
81 import yaml
83 with open(self.costformation_file, "r") as f:
84 costformation = yaml.safe_load(f)
86 # Extract dimensions from costformation
87 if "Dimensions" in costformation:
88 self.costformation_dimensions = list(costformation["Dimensions"].keys())
89 console.print(
90 f"📋 [green]Loaded {len(self.costformation_dimensions)} dimensions from costformation.yaml[/green]"
91 )
93 except Exception as e:
94 console.print(f"⚠️ [yellow]Could not load costformation.yaml: {e}[/yellow]")
96 def _load_config(self):
97 """Load and parse configuration file for automation."""
98 try:
99 if self.config_file.suffix.lower() in [".yaml", ".yml"]:
100 import yaml
102 with open(self.config_file, "r") as f:
103 self.config = yaml.safe_load(f)
104 elif self.config_file.suffix.lower() == ".json":
105 with open(self.config_file, "r") as f:
106 self.config = json.load(f)
107 else:
108 # Try to detect format from content
109 with open(self.config_file, "r") as f:
110 content = f.read()
111 if content.strip().startswith("{"):
112 self.config = json.loads(content)
113 else:
114 import yaml
116 self.config = yaml.safe_load(content)
118 console.print(
119 f"📋 [green]Loaded configuration from {self.config_file}[/green]"
120 )
121 console.print("🤖 [cyan]Running in automated mode[/cyan]")
123 except Exception as e:
124 console.print(f"⚠️ [yellow]Could not load config file: {e}[/yellow]")
125 console.print("💡 [cyan]Falling back to interactive mode[/cyan]")
126 self.config = {}
128 def analyze_columns(self):
129 """Analyze all columns to understand their characteristics."""
130 print_analysis_header("Analyzing column characteristics...")
132 for col in self.df.columns:
133 # Use utility function to get comprehensive column summary
134 column_summary = get_column_summary(self.df, col)
136 # Detect data patterns
137 col_data = self.df.select(col).to_series()
138 patterns = self._detect_patterns(col_data, column_summary["dtype"])
140 # Store complete analysis results
141 self.analysis_results["column_analysis"][col] = {
142 **column_summary, # Include all summary data
143 "patterns": patterns, # Add pattern detection
144 }
146 def _detect_patterns(self, col_data: pl.Series, dtype: str) -> Dict[str, Any]:
147 """Detect patterns in column data (datetime, numeric, etc.)."""
148 patterns = {
149 "is_datetime": False,
150 "is_numeric": False,
151 "is_identifier": False,
152 "datetime_format": None,
153 "supports_hourly": False,
154 "supports_daily": False,
155 }
157 # Check if it's already a datetime type
158 if "Datetime" in dtype or "Date" in dtype:
159 patterns["is_datetime"] = True
160 patterns = self._analyze_temporal_granularity(col_data, patterns)
162 # Check for numeric types
163 elif "Float" in dtype or "Int" in dtype:
164 patterns["is_numeric"] = True
166 # For string columns, check for datetime patterns
167 elif dtype == "String":
168 non_null = col_data.drop_nulls()
169 if len(non_null) > 0:
170 sample_vals = non_null.head(10).to_list()
172 # Check for datetime-like strings
173 datetime_patterns = [
174 r"\d{4}-\d{2}-\d{2}", # YYYY-MM-DD
175 r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}", # ISO datetime
176 r"\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}", # Standard datetime
177 r"\d{2}/\d{2}/\d{4}", # MM/DD/YYYY
178 r"\d{4}/\d{2}/\d{2}", # YYYY/MM/DD
179 ]
181 for pattern in datetime_patterns:
182 if any(re.search(pattern, str(val)) for val in sample_vals):
183 patterns["is_datetime"] = True
184 patterns["datetime_format"] = pattern
185 break
187 # Check for ID-like patterns
188 id_patterns = [
189 r"^[a-zA-Z0-9-]{8,}$", # UUID-like
190 r"^[A-Z]{2,4}-\d+$", # Code-number
191 r"^\d{6,}$", # Long number
192 ]
194 for pattern in id_patterns:
195 if any(re.search(pattern, str(val)) for val in sample_vals):
196 patterns["is_identifier"] = True
197 break
199 return patterns
201 def _analyze_temporal_granularity(
202 self, col_data: pl.Series, patterns: Dict
203 ) -> Dict:
204 """Analyze temporal data to determine if it supports hourly or daily granularity."""
205 try:
206 # Check if source data actually contains time information
207 has_time_data = False
209 # Convert to datetime if it's not already
210 if col_data.dtype == pl.Datetime:
211 datetime_col = col_data
212 has_time_data = True # Datetime columns have time data
213 elif col_data.dtype == pl.Date:
214 # Date columns only have date, no time data
215 datetime_col = col_data.dt.combine(pl.time(0, 0, 0))
216 has_time_data = False
217 elif "Datetime" in str(col_data.dtype):
218 # Already a datetime variant
219 datetime_col = col_data
220 has_time_data = True
221 else:
222 # Try to parse string as datetime
223 datetime_col = col_data.str.to_datetime(format=None, strict=False)
224 # Check if any non-null parsed values contain actual time data (not just 00:00:00)
225 non_null_dt = datetime_col.drop_nulls()
226 if len(non_null_dt) > 0:
227 # Check if any times are not midnight (00:00:00)
228 times = non_null_dt.dt.time()
229 has_time_data = not all(t == pl.time(0, 0, 0) for t in times.to_list())
231 # Remove nulls
232 datetime_col = datetime_col.drop_nulls()
234 if len(datetime_col) < 2:
235 return patterns
237 # Calculate time differences
238 sorted_times = datetime_col.sort()
239 time_diffs = sorted_times.diff().drop_nulls()
241 if len(time_diffs) == 0:
242 return patterns
244 # Convert to total seconds for analysis
245 diff_seconds = time_diffs.dt.total_seconds()
247 # Analyze the distribution of time differences
248 unique_diffs = diff_seconds.unique().sort()
250 # Check for hourly patterns (3600 seconds = 1 hour)
251 hourly_threshold = 7200 # 2 hours
252 hourly_count = sum(1 for diff in unique_diffs if diff <= hourly_threshold)
254 # Check for daily patterns (86400 seconds = 1 day)
255 daily_threshold = 172800 # 2 days
256 daily_count = sum(
257 1 for diff in unique_diffs if 21600 <= diff <= daily_threshold
258 ) # 6 hours to 2 days
260 total_unique = len(unique_diffs)
262 if total_unique > 0:
263 # Only support hourly if source data actually contains time information
264 if has_time_data and (
265 hourly_count / total_unique >= 0.3
266 ): # 30% of time diffs suggest hourly
267 patterns["supports_hourly"] = True
269 if daily_count / total_unique >= 0.3: # 30% of time diffs suggest daily
270 patterns["supports_daily"] = True
272 # If we have very regular intervals, be more confident
273 if len(unique_diffs) <= 3: # Very few unique intervals
274 # Only support hourly if source data actually contains time information
275 if has_time_data and any(3600 <= diff <= 3660 for diff in unique_diffs): # ~1 hour
276 patterns["supports_hourly"] = True
277 if any(86400 <= diff <= 86460 for diff in unique_diffs): # ~1 day
278 patterns["supports_daily"] = True
280 except Exception as e:
281 console.print(
282 f"⚠️ [yellow]Error analyzing temporal granularity: {e}[/yellow]"
283 )
285 return patterns
287 def analyze_temporal_columns(self):
288 """Analyze temporal columns and determine supported granularity."""
289 console.print("\n📅 [cyan]Analyzing temporal columns...[/cyan]")
291 temporal_cols = []
292 for col, analysis in self.analysis_results["column_analysis"].items():
293 if analysis["patterns"]["is_datetime"]:
294 temporal_cols.append(col)
296 if not temporal_cols:
297 console.print(
298 "❌ [red]No temporal columns detected. CloudZero telemetry requires timestamp data.[/red]"
299 )
300 self.analysis_results["temporal_analysis"] = {
301 "temporal_columns": [],
302 "supported_granularity": [],
303 "recommended_granularity": None,
304 }
305 return
307 # Analyze each temporal column
308 granularity_support = {}
309 for col in temporal_cols:
310 patterns = self.analysis_results["column_analysis"][col]["patterns"]
312 supports = []
313 if patterns["supports_hourly"]:
314 supports.append("HOURLY")
315 if patterns["supports_daily"]:
316 supports.append("DAILY")
318 granularity_support[col] = supports
320 # Determine overall recommendation
321 all_supported = set()
322 for supports in granularity_support.values():
323 all_supported.update(supports)
325 # Prefer HOURLY if available, fallback to DAILY
326 if "HOURLY" in all_supported:
327 recommended = "HOURLY"
328 elif "DAILY" in all_supported:
329 recommended = "DAILY"
330 else:
331 recommended = "DAILY" # Default fallback
333 self.analysis_results["temporal_analysis"] = {
334 "temporal_columns": temporal_cols,
335 "granularity_support": granularity_support,
336 "supported_granularity": list(all_supported),
337 "recommended_granularity": recommended,
338 }
340 # Brief summary only - detailed analysis will be shown during selection
341 print_success(f"Found {len(temporal_cols)} temporal column(s)")
342 if recommended:
343 console.print(
344 f"💡 [green]Recommended granularity: {recommended}[/green]"
345 )
347 # Interactive selection of temporal column and granularity
348 selected_temporal = self._select_temporal_column_and_granularity(
349 temporal_cols, granularity_support, recommended
350 )
352 # Update analysis results with user selection
353 self.analysis_results["temporal_analysis"]["selected_column"] = (
354 selected_temporal["column"]
355 )
356 self.analysis_results["temporal_analysis"]["selected_granularity"] = (
357 selected_temporal["granularity"]
358 )
360 def _select_temporal_column_and_granularity(
361 self,
362 temporal_cols: List[str],
363 granularity_support: Dict[str, List[str]],
364 recommended: str,
365 ) -> Dict[str, str]:
366 """Interactive selection of temporal column and granularity."""
368 # Check if selections are specified in config
369 config_timestamp = self.config.get("selections", {}).get("timestamp")
370 config_granularity = self.config.get("selections", {}).get("granularity")
372 if config_timestamp and config_granularity:
373 if config_timestamp in temporal_cols:
374 # Validate that the configured granularity is supported by the column
375 supported_granularities = granularity_support.get(config_timestamp, [])
376 if (
377 config_granularity.upper() in supported_granularities
378 or not supported_granularities
379 ):
380 console.print(
381 f"🤖 [cyan]Using configured temporal selection: {config_timestamp} with {config_granularity.upper()} granularity[/cyan]"
382 )
383 return {
384 "column": config_timestamp,
385 "granularity": config_granularity.upper(),
386 }
387 else:
388 console.print(
389 f"⚠️ [yellow]Configured granularity '{config_granularity}' not supported by column '{config_timestamp}'. Using interactive mode.[/yellow]"
390 )
391 else:
392 console.print(
393 f"⚠️ [yellow]Configured timestamp column '{config_timestamp}' not found. Using interactive mode.[/yellow]"
394 )
396 console.print("\n⏰ [bold]Select Temporal Column and Granularity[/bold]")
397 console.print(
398 "💡 [dim]Choose which date/time column to use and at what granularity for your telemetry data.[/dim]"
399 )
401 # If only one temporal column, auto-select it but still ask for granularity
402 if len(temporal_cols) == 1:
403 selected_column = temporal_cols[0]
404 console.print(
405 f"✅ [green]Auto-selected temporal column: {selected_column} (only temporal column found)[/green]"
406 )
407 else:
408 # Multiple temporal columns - let user choose
409 console.print("\n📅 [cyan]Available temporal columns:[/cyan]")
411 selection_table = create_temporal_selection_table()
413 for i, col in enumerate(temporal_cols, 1):
414 col_analysis = self.analysis_results["column_analysis"][col]
415 sample_vals = ", ".join(
416 str(v) for v in col_analysis["sample_values"][:2]
417 )
418 supported = (
419 ", ".join(granularity_support[col])
420 if granularity_support[col]
421 else "Unknown"
422 )
424 selection_table.add_row(
425 str(i), col, col_analysis["dtype"], sample_vals, supported
426 )
428 console.print(selection_table)
430 while True:
431 try:
432 choice = int_prompt_with_ctrl_c_reminder(
433 f"\n🤔 Select temporal column (1-{len(temporal_cols)})",
434 default=1,
435 )
436 if 1 <= choice <= len(temporal_cols):
437 selected_column = temporal_cols[choice - 1]
438 console.print(
439 f"✅ [green]Selected temporal column: {selected_column}[/green]"
440 )
441 break
442 else:
443 console.print(
444 f"❌ [red]Please enter a number between 1 and {len(temporal_cols)}[/red]"
445 )
446 except ValueError:
447 console.print("❌ [red]Please enter a valid number[/red]")
449 # Now select granularity for the chosen column
450 supported_granularities = granularity_support.get(selected_column, [])
452 if not supported_granularities:
453 console.print(
454 f"⚠️ [yellow]No specific granularity detected for {selected_column}. Defaulting to {recommended}.[/yellow]"
455 )
456 selected_granularity = recommended
457 elif len(supported_granularities) == 1:
458 selected_granularity = supported_granularities[0]
459 console.print(
460 f"✅ [green]Auto-selected granularity: {selected_granularity} (only supported granularity)[/green]"
461 )
462 else:
463 # Multiple granularities supported - let user choose
464 console.print(f"\n⏱️ [bold]Select Granularity for {selected_column}[/bold]")
465 console.print(
466 "💡 [dim]Choose the time granularity for your telemetry data:[/dim]"
467 )
469 # Show granularity options
470 granularity_columns = [
471 TableColumn("#", "dim"),
472 TableColumn("Granularity", "cyan"),
473 TableColumn("Description", "yellow"),
474 TableColumn("Use Case", "green")
475 ]
476 granularity_table = create_basic_table(None, granularity_columns)
478 granularity_info = {
479 "HOURLY": {
480 "description": "Hour-by-hour tracking",
481 "use_case": "High-frequency monitoring, real-time analytics",
482 },
483 "DAILY": {
484 "description": "Day-by-day tracking",
485 "use_case": "Daily reporting, cost aggregation",
486 },
487 }
489 for i, granularity in enumerate(supported_granularities, 1):
490 info = granularity_info.get(
491 granularity,
492 {
493 "description": "Custom granularity",
494 "use_case": "Specific use case",
495 },
496 )
497 granularity_table.add_row(
498 str(i), granularity, info["description"], info["use_case"]
499 )
501 console.print(granularity_table)
502 console.print(f"💡 [green]Recommended: {recommended}[/green]")
504 # Default to recommended if it's in supported options
505 default_choice = (
506 supported_granularities.index(recommended) + 1
507 if recommended in supported_granularities
508 else 1
509 )
511 while True:
512 try:
513 choice = int_prompt_with_ctrl_c_reminder(
514 f"\n🤔 Select granularity (1-{len(supported_granularities)})",
515 default=default_choice,
516 )
517 if 1 <= choice <= len(supported_granularities):
518 selected_granularity = supported_granularities[choice - 1]
519 console.print(
520 f"✅ [green]Selected granularity: {selected_granularity}[/green]"
521 )
522 break
523 else:
524 console.print(
525 f"❌ [red]Please enter a number between 1 and {len(supported_granularities)}[/red]"
526 )
527 except ValueError:
528 console.print("❌ [red]Please enter a valid number[/red]")
530 # Show final selection summary
531 console.print("\n📋 [bold]Temporal Selection Summary:[/bold]")
532 summary_table = create_temporal_summary_table()
533 summary_table.add_row("Temporal Column", selected_column)
534 summary_table.add_row("Granularity", selected_granularity)
535 summary_table.add_row(
536 "Impact",
537 f"Data will be grouped by {selected_granularity.lower()} intervals",
538 )
539 console.print(summary_table)
541 return {"column": selected_column, "granularity": selected_granularity}
543 def get_telemetry_type(self) -> str:
544 """Interactive prompt to determine telemetry type (or use config)."""
545 console.print("\n" + "=" * 80)
546 console.print("🎯 [bold cyan]CloudZero Telemetry Mapping Analysis[/bold cyan]")
547 console.print("=" * 80)
549 # Check if telemetry type is specified in config
550 if self.config.get("telemetry_type"):
551 telemetry_type = self.config["telemetry_type"].lower()
552 if telemetry_type in ["allocation", "unit"]:
553 console.print(
554 f"🤖 [cyan]Using configured telemetry type: {telemetry_type.upper()}[/cyan]"
555 )
556 self.analysis_results["telemetry_type"] = telemetry_type
557 return telemetry_type
558 else:
559 console.print(
560 f"⚠️ [yellow]Invalid telemetry type in config: {telemetry_type}. Using interactive mode.[/yellow]"
561 )
563 console.print("\n📋 [cyan]CloudZero supports two types of telemetry:[/cyan]")
564 console.print(
565 "1️⃣ [bold]Allocation Telemetry[/bold] - Track resource usage/consumption by business dimensions"
566 )
567 console.print(
568 " • Example: Database queries by customer, compute hours by team"
569 )
570 console.print(
571 " • Requires: element-name (what), value (amount), timestamp, filters (who/how)"
572 )
574 console.print(
575 "\n2️⃣ [bold]Unit Cost Telemetry[/bold] - Track costs associated with business metrics"
576 )
577 console.print(" • Example: Cost per API call, cost per user session")
578 console.print(
579 " • Requires: element-name (metric), value (amount), timestamp, associated-costs"
580 )
582 while True:
583 choice = prompt_with_ctrl_c_reminder(
584 "\n🤔 Which type of telemetry are you planning to create?",
585 choices=["allocation", "unit", "help"],
586 default="allocation",
587 )
589 if choice == "help":
590 console.print("\n📖 [cyan]More information:[/cyan]")
591 console.print(
592 "• Allocation: Use when tracking resource consumption (hours, requests, bytes)"
593 )
594 console.print(
595 "• Unit Cost: Use when tracking costs associated with business metrics"
596 )
597 console.print(
598 "• See: https://docs.cloudzero.com/reference/allocation-telemetry-api-1"
599 )
600 console.print(
601 "• See: https://docs.cloudzero.com/reference/unit-metric-telemetry-api-1"
602 )
603 continue
605 self.analysis_results["telemetry_type"] = choice
606 return choice
608 def present_column_candidates(self, telemetry_type: str):
609 """Present column candidates based on telemetry type and get user selections."""
610 console.print(
611 f"\n🔍 [cyan]Analyzing columns for {telemetry_type.upper()} telemetry...[/cyan]"
612 )
614 # Categorize columns
615 high_cardinality_candidates = []
616 numeric_cols = []
617 temporal_cols = []
618 filter_cols = []
620 for col, analysis in self.analysis_results["column_analysis"].items():
621 # All string columns for element-name, excluding datetime columns
622 if (
623 not analysis["patterns"]["is_datetime"]
624 and analysis["dtype"] == "String"
625 ):
626 high_cardinality_candidates.append((col, analysis))
628 if analysis["patterns"]["is_numeric"]:
629 numeric_cols.append(col)
631 if analysis["patterns"]["is_datetime"]:
632 temporal_cols.append(col)
634 # Only non-numeric, low cardinality columns for filters
635 if (
636 analysis["cardinality"] == "low"
637 and not analysis["patterns"]["is_numeric"]
638 and not analysis["patterns"]["is_datetime"]
639 ):
640 filter_cols.append(col)
642 # Sort all string candidates by cardinality ratio (descending) and take top 10
643 high_cardinality_candidates.sort(
644 key=lambda x: x[1]["cardinality_ratio"], reverse=True
645 )
646 high_cardinality_cols = [col for col, _ in high_cardinality_candidates[:10]]
648 # Get user selections
649 selections = {}
651 # Element name selection
652 selections["element_name"] = self._select_element_name(
653 high_cardinality_cols, telemetry_type
654 )
656 # Value selection
657 selections["value"] = self._select_value_column(numeric_cols, telemetry_type)
659 # Timestamp selection - use already selected from temporal analysis
660 temporal_analysis = self.analysis_results.get("temporal_analysis", {})
661 if temporal_analysis.get("selected_column"):
662 selections["timestamp"] = temporal_analysis["selected_column"]
663 console.print(
664 f"✅ [green]Using selected timestamp column: {temporal_analysis['selected_column']}[/green]"
665 )
666 else:
667 # Fallback to old method if no selection was made
668 selections["timestamp"] = self._select_timestamp_column(temporal_cols)
670 # Filter/associated costs selection
671 if telemetry_type == "allocation":
672 selections["filters"] = self._select_filter_columns(filter_cols)
673 else: # unit cost
674 selections["associated_costs"] = self._select_associated_costs(filter_cols)
676 self.analysis_results["selected_mapping"] = selections
677 return selections
679 def _select_element_name(self, candidates: List[str], telemetry_type: str) -> str:
680 """Interactive selection of element-name column(s) - single or composite."""
681 console.print("\n🏷️ [bold]Select ELEMENT-NAME column(s)[/bold]")
682 console.print(
683 f"This identifies what you're measuring in your {telemetry_type} telemetry."
684 )
685 console.print(
686 "💡 [dim]You can select a single column or create a composite element-name from multiple columns.[/dim]"
687 )
689 if not candidates:
690 console.print("❌ [red]No string columns found for element-name[/red]")
691 return self._manual_column_selection("element-name")
693 # Check if composite element-name is configured
694 config_key = "element_name"
695 if self.config.get("selections", {}).get(config_key):
696 configured_element = self.config["selections"][config_key]
697 if isinstance(configured_element, dict) and "columns" in configured_element:
698 # Composite element-name from config
699 columns = configured_element["columns"]
700 separator = configured_element.get("separator", "|")
701 if all(col in self.df.columns for col in columns):
702 console.print(
703 f"🤖 [cyan]Using configured composite element-name: {' + '.join(columns)} (separator: '{separator}')[/cyan]"
704 )
705 return self._create_composite_element_name(columns, separator)
706 elif (
707 isinstance(configured_element, str)
708 and configured_element in self.df.columns
709 ):
710 # Single column from config
711 console.print(
712 f"🤖 [cyan]Using configured element-name: {configured_element}[/cyan]"
713 )
714 return configured_element
716 # Interactive selection
717 console.print(
718 "\n📋 [cyan]Top 10 string columns by cardinality (ordered highest to lowest):[/cyan]"
719 )
720 console.print(
721 "💡 [yellow]Tip: Columns with 2-100 unique values are ideal for element-names (good balance of usability and performance)[/yellow]"
722 )
724 # Create candidate table using utility function
725 table = create_selection_table()
727 for i, col in enumerate(candidates, 1):
728 analysis = self.analysis_results["column_analysis"][col]
729 add_selection_row(table, i, col, analysis)
731 console.print(table)
733 while True:
734 choice = prompt_with_ctrl_c_reminder(
735 "\n🤔 Select element-name option",
736 choices=["single", "composite", "manual"],
737 default="single",
738 )
740 if choice == "single":
741 return self._select_single_element_name(candidates)
742 elif choice == "composite":
743 return self._select_composite_element_name(candidates)
744 else: # manual
745 return self._manual_column_selection("element-name")
747 def _select_single_element_name(self, candidates: List[str]) -> str:
748 """Select a single column for element-name."""
749 while True:
750 try:
751 choice = int_prompt_with_ctrl_c_reminder(
752 f"\n🤔 Select column number (1-{len(candidates)})", default=1
753 )
754 if 1 <= choice <= len(candidates):
755 selected_col = candidates[choice - 1]
757 # Validate that the selected column has no null values
758 col_analysis = self.analysis_results["column_analysis"][
759 selected_col
760 ]
761 if col_analysis["null_count"] > 0:
762 console.print(
763 f"⚠️ [yellow]Note: Column '{selected_col}' contains {col_analysis['null_count']:,} null values.[/yellow]"
764 )
765 console.print(
766 "💡 [cyan]Rows with null element-names will be excluded from final telemetry output.[/cyan]"
767 )
769 console.print(
770 f"✅ [green]Selected element-name: {selected_col}[/green]"
771 )
772 return selected_col
773 else:
774 console.print(
775 f"❌ [red]Please enter a number between 1 and {len(candidates)}[/red]"
776 )
777 except ValueError:
778 console.print("❌ [red]Please enter a valid number[/red]")
780 def _select_composite_element_name(self, candidates: List[str]) -> str:
781 """Select multiple columns to create a composite element-name."""
782 console.print("\n🔗 [bold]Creating composite element-name[/bold]")
783 console.print("💡 [dim]Select 2 or more columns to combine[/dim]")
785 selected_columns = []
786 available_candidates = candidates.copy()
788 while len(selected_columns) < 2 or confirm_with_ctrl_c_reminder(
789 f"\n➕ Add another column? (currently selected: {len(selected_columns)})"
790 ):
791 if not available_candidates:
792 console.print("❌ [red]No more columns available[/red]")
793 break
795 # Show available columns
796 console.print(
797 f"\n📋 [cyan]Available columns ({len(selected_columns)} already selected):[/cyan]"
798 )
799 table = Table(box=box.ROUNDED)
800 table.add_column("#", style="dim")
801 table.add_column("Column", style="cyan")
802 table.add_column("Type", style="yellow")
803 table.add_column("Unique Values", style="green")
804 table.add_column("Sample Values", style="magenta")
806 for i, col in enumerate(available_candidates, 1):
807 analysis = self.analysis_results["column_analysis"][col]
808 samples = ", ".join(str(v) for v in analysis["sample_values"][:5])
809 table.add_row(
810 str(i),
811 col,
812 analysis["dtype"],
813 f"{analysis['unique_count']:,}",
814 samples,
815 )
817 console.print(table)
819 try:
820 choice = int_prompt_with_ctrl_c_reminder(
821 f"🤔 Select column number (1-{len(available_candidates)})"
822 )
823 if 1 <= choice <= len(available_candidates):
824 selected_col = available_candidates[choice - 1]
825 selected_columns.append(selected_col)
826 available_candidates.remove(selected_col)
827 console.print(f"✅ [green]Added: {selected_col}[/green]")
828 console.print(
829 f"📝 [cyan]Selected columns: {' + '.join(selected_columns)}[/cyan]"
830 )
831 else:
832 console.print(
833 f"❌ [red]Please enter a number between 1 and {len(available_candidates)}[/red]"
834 )
835 continue
836 except ValueError:
837 console.print("❌ [red]Please enter a valid number[/red]")
838 continue
840 if len(selected_columns) >= 2:
841 break
843 if len(selected_columns) < 2:
844 console.print(
845 "❌ [red]Need at least 2 columns for composite element-name[/red]"
846 )
847 return self._select_single_element_name(candidates)
849 # Get column order and separator
850 return self._configure_composite_element_name(selected_columns)
852 def _configure_composite_element_name(self, selected_columns: List[str]) -> str:
853 """Configure column order and separator for composite element-name."""
854 console.print("\n⚙️ [bold]Configure composite element-name[/bold]")
855 console.print(
856 f"📝 [cyan]Selected columns: {', '.join(selected_columns)}[/cyan]"
857 )
859 # Column ordering (if more than 2 columns)
860 ordered_columns = selected_columns.copy()
861 if len(selected_columns) > 2:
862 console.print("\n🔢 [bold]Column ordering[/bold]")
863 console.print(
864 "💡 [dim]Current order will be used unless you want to reorder[/dim]"
865 )
867 if confirm_with_ctrl_c_reminder("🔄 Do you want to reorder the columns?"):
868 ordered_columns = self._reorder_columns(selected_columns)
870 # Separator selection
871 console.print("\n🔗 [bold]Choose separator[/bold]")
872 separator_options = ["_", "-", ".", ":", "|", " ", "custom"]
873 separator_choice = prompt_with_ctrl_c_reminder(
874 "Select separator", choices=separator_options, default="|"
875 )
877 if separator_choice == "custom":
878 separator = prompt_with_ctrl_c_reminder(
879 "Enter custom separator", default="|"
880 )
881 else:
882 separator = separator_choice
884 # Create composite element-name and preview
885 composite_name = self._create_composite_element_name(ordered_columns, separator)
887 # Show cardinality analysis for the composite element-name
888 self._show_composite_cardinality_analysis(
889 composite_name, ordered_columns, separator
890 )
892 # Check if any valid composite values exist
893 valid_rows = len(self.df.filter(pl.col(composite_name).is_not_null()))
895 if valid_rows == 0:
896 console.print("⚠️ [red]No valid composite element-names found! All rows contain null values.[/red]")
897 console.print("💡 [yellow]Consider selecting different columns or cleaning your data.[/yellow]")
898 return self._configure_composite_element_name(selected_columns) # Try again
900 if confirm_with_ctrl_c_reminder(
901 "\n✅ Accept this composite element-name?", default=True
902 ):
903 console.print(
904 f"✅ [green]Created composite element-name: {' + '.join(ordered_columns)} (separator: '{separator}')[/green]"
905 )
906 return composite_name
907 else:
908 console.print("🔄 [yellow]Let's try again...[/yellow]")
909 return self._configure_composite_element_name(selected_columns)
911 def _reorder_columns(self, columns: List[str]) -> List[str]:
912 """Allow user to reorder columns for composite element-name."""
913 console.print("\n🔢 [bold]Reorder columns[/bold]")
914 console.print("💡 [dim]Enter the column numbers in your desired order[/dim]")
916 # Show current order
917 table = Table(box=box.ROUNDED)
918 table.add_column("#", style="dim")
919 table.add_column("Column", style="cyan")
921 for i, col in enumerate(columns, 1):
922 table.add_row(str(i), col)
924 console.print(table)
926 while True:
927 try:
928 order_input = prompt_with_ctrl_c_reminder(
929 f"Enter column order (e.g., '1 3 2' for {len(columns)} columns)",
930 default=" ".join(str(i) for i in range(1, len(columns) + 1)),
931 )
933 order_indices = [int(x.strip()) for x in order_input.split()]
935 if len(order_indices) != len(columns):
936 console.print(
937 f"❌ [red]Please enter exactly {len(columns)} numbers[/red]"
938 )
939 continue
941 if not all(1 <= i <= len(columns) for i in order_indices):
942 console.print(
943 f"❌ [red]Numbers must be between 1 and {len(columns)}[/red]"
944 )
945 continue
947 if len(set(order_indices)) != len(columns):
948 console.print(
949 "❌ [red]Each column number must be used exactly once[/red]"
950 )
951 continue
953 ordered_columns = [columns[i - 1] for i in order_indices]
954 console.print(
955 f"✅ [green]New order: {' → '.join(ordered_columns)}[/green]"
956 )
957 return ordered_columns
959 except ValueError:
960 console.print(
961 "❌ [red]Please enter valid numbers separated by spaces[/red]"
962 )
964 def _create_composite_element_name(self, columns: List[str], separator: str) -> str:
965 """Create a composite element-name column in the dataframe."""
966 composite_col_name = f"composite_element_name_{'_'.join(columns)}"
968 # Validate that none of the source columns have null values
969 null_counts = {}
970 for col in columns:
971 null_count = self.df.select(col).to_series().null_count()
972 if null_count > 0:
973 null_counts[col] = null_count
975 if null_counts:
976 console.print(
977 "\nℹ️ [yellow]Note: Source columns contain null values:[/yellow]"
978 )
979 for col, count in null_counts.items():
980 console.print(f" • {col}: {count:,} null values")
981 console.print(
982 "💡 [cyan]Rows with null element-names will be excluded from final telemetry output.[/cyan]"
983 )
985 # Create the composite column by concatenating the selected columns
986 # Use .str.concat() with ignore_nulls=False to preserve null handling
987 composite_expr = pl.concat_str(
988 [pl.col(col).cast(pl.Utf8) for col in columns],
989 separator=separator,
990 ignore_nulls=False,
991 )
993 # Add the composite column to the dataframe
994 self.df = self.df.with_columns(composite_expr.alias(composite_col_name))
996 # Update analysis results for the new composite column
997 composite_series = self.df.select(composite_col_name).to_series()
998 null_count = composite_series.null_count()
999 total_count = len(composite_series)
1000 non_null_count = total_count - null_count
1002 if non_null_count > 0:
1003 non_null_data = composite_series.drop_nulls()
1004 unique_count_non_null = non_null_data.n_unique()
1005 cardinality_ratio = unique_count_non_null / non_null_count
1006 else:
1007 unique_count_non_null = 0
1008 cardinality_ratio = 0.0
1010 # Cardinality classification
1011 if cardinality_ratio > 0.9:
1012 cardinality = "very_high"
1013 elif cardinality_ratio > 0.5:
1014 cardinality = "high"
1015 elif cardinality_ratio > 0.1:
1016 cardinality = "medium"
1017 else:
1018 cardinality = "low"
1020 sample_values = (
1021 non_null_data.head(5).to_list() if len(non_null_data) > 0 else []
1022 )
1024 self.analysis_results["column_analysis"][composite_col_name] = {
1025 "dtype": "String",
1026 "unique_count": unique_count_non_null,
1027 "null_count": null_count,
1028 "total_count": total_count,
1029 "non_null_count": non_null_count,
1030 "cardinality": cardinality,
1031 "cardinality_ratio": cardinality_ratio,
1032 "sample_values": sample_values,
1033 "value_counts": {},
1034 "patterns": {
1035 "is_numeric": False,
1036 "is_datetime": False,
1037 "is_identifier": True, # Composite element-names are typically identifiers
1038 "datetime_format": None,
1039 },
1040 "temporal_analysis": {},
1041 "composite_info": {
1042 "is_composite": True,
1043 "source_columns": columns,
1044 "separator": separator,
1045 },
1046 }
1048 return composite_col_name
1050 def _show_composite_cardinality_analysis(
1051 self, composite_col_name: str, source_columns: List[str], separator: str
1052 ):
1053 """Display cardinality analysis for the newly created composite element-name."""
1054 console.print(
1055 "\n📊 [bold]Cardinality Analysis for Composite Element-Name[/bold]"
1056 )
1058 # Get analysis data for the composite column
1059 composite_analysis = self.analysis_results["column_analysis"][
1060 composite_col_name
1061 ]
1063 # Get individual column cardinalities for comparison
1064 individual_cardinalities = []
1065 for col in source_columns:
1066 col_analysis = self.analysis_results["column_analysis"][col]
1067 individual_cardinalities.append(
1068 {
1069 "column": col,
1070 "unique_count": col_analysis["unique_count"],
1071 "cardinality_ratio": col_analysis["cardinality_ratio"],
1072 "cardinality": col_analysis["cardinality"],
1073 }
1074 )
1076 # Create comparison table
1077 comparison_table = Table(box=box.ROUNDED)
1078 comparison_table.add_column("Column", style="cyan")
1079 comparison_table.add_column("Unique Values", style="green")
1080 comparison_table.add_column("Cardinality %", style="blue")
1081 comparison_table.add_column("Level", style="yellow")
1083 # Add individual columns
1084 for col_data in individual_cardinalities:
1085 comparison_table.add_row(
1086 col_data["column"],
1087 f"{col_data['unique_count']:,}",
1088 f"{col_data['cardinality_ratio'] * 100:.1f}%",
1089 col_data["cardinality"].title(),
1090 )
1092 # Add separator row
1093 comparison_table.add_row("─" * 20, "─" * 10, "─" * 10, "─" * 10, style="dim")
1095 # Add composite column
1096 comparison_table.add_row(
1097 f"[bold]Composite ({separator.join(source_columns)})[/bold]",
1098 f"[bold]{composite_analysis['unique_count']:,}[/bold]",
1099 f"[bold]{composite_analysis['cardinality_ratio'] * 100:.1f}%[/bold]",
1100 f"[bold]{composite_analysis['cardinality'].title()}[/bold]",
1101 style="green",
1102 )
1104 console.print(comparison_table)
1106 # Analysis insights
1107 total_individual_unique = sum(
1108 col["unique_count"] for col in individual_cardinalities
1109 )
1110 composite_unique = composite_analysis["unique_count"]
1112 console.print("\n🔍 [bold]Cardinality Insights:[/bold]")
1114 if composite_unique == total_individual_unique:
1115 console.print(
1116 "✨ [green]Perfect cardinality increase! Each combination is unique.[/green]"
1117 )
1118 elif composite_unique > max(
1119 col["unique_count"] for col in individual_cardinalities
1120 ):
1121 increase_pct = (
1122 (
1123 composite_unique
1124 / max(col["unique_count"] for col in individual_cardinalities)
1125 )
1126 - 1
1127 ) * 100
1128 console.print(
1129 f"📈 [cyan]Cardinality increased by {increase_pct:.1f}% compared to highest individual column[/cyan]"
1130 )
1131 else:
1132 console.print(
1133 "⚠️ [yellow]Composite cardinality is not higher than individual columns - possible data overlap[/yellow]"
1134 )
1136 # Show distribution if low cardinality
1137 if composite_analysis["cardinality"] in ["low", "medium"]:
1138 console.print("\n📋 [bold]Top Composite Values:[/bold]")
1140 # Add row/null summary
1141 total_rows = len(self.df)
1142 valid_rows = composite_analysis["non_null_count"]
1143 excluded_rows = total_rows - valid_rows
1145 console.print(
1146 f"📊 [dim]Total rows: {total_rows:,} | Valid for telemetry: {valid_rows:,} | Excluded (nulls): {excluded_rows:,}[/dim]"
1147 )
1149 composite_series = self.df.select(composite_col_name).to_series()
1150 value_counts = composite_series.drop_nulls().value_counts().head(10)
1152 value_table = Table(box=box.ROUNDED)
1153 value_table.add_column("Composite Value", style="cyan")
1154 value_table.add_column("Count", style="green")
1155 value_table.add_column("Percentage", style="blue")
1157 total_non_null = composite_analysis["non_null_count"]
1158 for row in value_counts.iter_rows():
1159 value, count = row
1160 percentage = (count / total_non_null) * 100
1161 value_table.add_row(str(value), f"{count:,}", f"{percentage:.1f}%")
1163 console.print(value_table)
1165 def _select_value_column(self, candidates: List[str], telemetry_type: str) -> str:
1166 """Interactive selection of value column."""
1167 console.print("\n📊 [bold]Select VALUE column[/bold]")
1169 if telemetry_type == "allocation":
1170 console.print(
1171 "This should be the quantity/amount being allocated (e.g., hours, requests, bytes)."
1172 )
1173 else:
1174 console.print(
1175 "This should be the metric value you're tracking costs for (e.g., API calls, users)."
1176 )
1178 if not candidates:
1179 console.print("❌ [red]No numeric columns found for value[/red]")
1180 return self._manual_column_selection("value")
1182 return self._interactive_column_selection(
1183 candidates, "value", "Numeric columns"
1184 )
1186 def _select_timestamp_column(self, candidates: List[str]) -> str:
1187 """Return the previously selected timestamp column from temporal analysis."""
1188 # Check if we already selected a temporal column during temporal analysis
1189 selected_column = self.analysis_results["temporal_analysis"].get(
1190 "selected_column"
1191 )
1192 selected_granularity = self.analysis_results["temporal_analysis"].get(
1193 "selected_granularity"
1194 )
1196 if selected_column and selected_column in candidates:
1197 console.print(
1198 f"\n⏰ [green]Using previously selected timestamp column: {selected_column} ({selected_granularity})[/green]"
1199 )
1200 return selected_column
1202 # Fallback to interactive selection if no previous selection
1203 console.print("\n⏰ [bold]Select TIMESTAMP column[/bold]")
1204 console.print("This should contain the date/time when the event occurred.")
1206 if not candidates:
1207 console.print("❌ [red]No temporal columns found[/red]")
1208 return self._manual_column_selection("timestamp")
1210 if len(candidates) == 1:
1211 col = candidates[0]
1212 console.print(
1213 f"✅ [green]Auto-selected: {col} (only temporal column found)[/green]"
1214 )
1215 return col
1217 return self._interactive_column_selection(
1218 candidates, "timestamp", "Temporal columns"
1219 )
1221 def _select_filter_columns(self, candidates: List[str]) -> List[str]:
1222 """Interactive selection of filter columns for allocation telemetry."""
1223 console.print("\n🔍 [bold]Select FILTER columns (optional)[/bold]")
1224 console.print(
1225 "These help categorize your allocations (e.g., customer, team, environment)."
1226 )
1228 if self.costformation_dimensions:
1229 console.print(
1230 f"💡 [cyan]Suggested from costformation.yaml: {', '.join(self.costformation_dimensions)}[/cyan]"
1231 )
1233 if not candidates:
1234 console.print("❌ [red]No low-cardinality columns found for filters[/red]")
1235 return []
1237 return self._multi_select_columns(
1238 candidates, "filters", "Low cardinality columns (good for grouping)"
1239 )
1241 def _select_associated_costs(self, candidates: List[str]) -> List[str]:
1242 """Interactive selection of associated costs for unit cost telemetry."""
1243 console.print("\n💰 [bold]Select ASSOCIATED COSTS columns (optional)[/bold]")
1244 console.print(
1245 "These should contain cost dimensions related to your unit metrics."
1246 )
1248 if self.costformation_dimensions:
1249 console.print(
1250 f"💡 [cyan]Suggested from costformation.yaml: {', '.join(self.costformation_dimensions)}[/cyan]"
1251 )
1253 if not candidates:
1254 console.print("❌ [red]No columns found for associated costs[/red]")
1255 return []
1257 return self._multi_select_columns(
1258 candidates, "associated_costs", "Potential cost-related columns"
1259 )
1261 def _interactive_column_selection(
1262 self, candidates: List[str], field_name: str, description: str
1263 ) -> str:
1264 """Present candidates and get user selection (or use config)."""
1265 # Check if selection is specified in config
1266 config_key = field_name.replace("-", "_")
1267 if self.config.get("selections", {}).get(config_key):
1268 configured_column = self.config["selections"][config_key]
1269 if configured_column in candidates:
1270 console.print(
1271 f"🤖 [cyan]Using configured {field_name}: {configured_column}[/cyan]"
1272 )
1273 return configured_column
1274 elif configured_column in self.df.columns:
1275 console.print(
1276 f"🤖 [cyan]Using configured {field_name}: {configured_column}[/cyan]"
1277 )
1278 return configured_column
1279 else:
1280 console.print(
1281 f"⚠️ [yellow]Configured {field_name} '{configured_column}' not found. Using interactive mode.[/yellow]"
1282 )
1284 console.print(f"\n📋 [cyan]{description}:[/cyan]")
1286 # Create candidate table
1287 table = Table(box=box.ROUNDED)
1288 table.add_column("#", style="dim")
1289 table.add_column("Column", style="cyan")
1290 table.add_column("Type", style="yellow")
1291 table.add_column("Unique Values", style="green")
1292 table.add_column("Null Count", style="red")
1294 # Add cardinality ratio for element-name selection to show ranking
1295 if field_name == "element-name":
1296 table.add_column("Cardinality %", style="blue")
1298 table.add_column("Sample Values", style="magenta")
1300 for i, col in enumerate(candidates, 1):
1301 analysis = self.analysis_results["column_analysis"][col]
1302 samples = ", ".join(str(v) for v in analysis["sample_values"][:5])
1304 row_data = [
1305 str(i),
1306 col,
1307 analysis["dtype"],
1308 f"{analysis['unique_count']:,}",
1309 f"{analysis['null_count']:,}",
1310 ]
1312 # Add cardinality percentage for element-name selection
1313 if field_name == "element-name":
1314 cardinality_pct = analysis["cardinality_ratio"] * 100
1315 row_data.append(f"{cardinality_pct:.1f}%")
1317 row_data.append(samples)
1318 table.add_row(*row_data)
1320 console.print(table)
1322 # Get selection
1323 while True:
1324 choice = Prompt.ask(
1325 f"Select {field_name} column",
1326 choices=[str(i) for i in range(1, len(candidates) + 1)]
1327 + ["manual", "skip"],
1328 )
1330 if choice == "skip":
1331 return None
1332 elif choice == "manual":
1333 return self._manual_column_selection(field_name)
1334 else:
1335 return candidates[int(choice) - 1]
1337 def _multi_select_columns(
1338 self, candidates: List[str], field_name: str, description: str
1339 ) -> List[str]:
1340 """Multi-selection of columns (or use config)."""
1341 # Check if selection is specified in config
1342 config_key = field_name.replace("-", "_")
1343 if self.config.get("selections", {}).get(config_key):
1344 configured_columns = self.config["selections"][config_key]
1345 if isinstance(configured_columns, str):
1346 configured_columns = [configured_columns]
1348 # Validate configured columns exist
1349 valid_columns = [
1350 col for col in configured_columns if col in self.df.columns
1351 ]
1352 if valid_columns:
1353 console.print(
1354 f"🤖 [cyan]Using configured {field_name}: {', '.join(valid_columns)}[/cyan]"
1355 )
1356 return valid_columns
1357 else:
1358 console.print(
1359 f"⚠️ [yellow]Configured {field_name} columns not found. Using interactive mode.[/yellow]"
1360 )
1362 console.print(f"\n📋 [cyan]{description}:[/cyan]")
1364 # Create candidate table using utility function
1365 table = create_multi_select_table()
1367 for i, col in enumerate(candidates, 1):
1368 analysis = self.analysis_results["column_analysis"][col]
1369 add_multi_select_row(table, i, col, analysis)
1371 console.print(table)
1373 console.print(
1374 "\n💡 [yellow]You can select multiple columns (comma-separated) or 'none' to skip[/yellow]"
1375 )
1376 console.print(
1377 "ℹ️ [dim]Note: Rows with null values in selected columns will be excluded from final telemetry output[/dim]"
1378 )
1380 while True:
1381 choice = Prompt.ask(f"Select {field_name} columns", default="none")
1383 if choice.lower() == "none":
1384 return []
1386 try:
1387 indices = [int(x.strip()) - 1 for x in choice.split(",")]
1388 selected = [candidates[i] for i in indices if 0 <= i < len(candidates)]
1390 if selected:
1391 print_success(f"Selected: {', '.join(selected)}")
1392 return selected
1393 else:
1394 console.print("❌ [red]Invalid selection. Please try again.[/red]")
1395 except ValueError:
1396 console.print(
1397 "❌ [red]Invalid format. Use numbers separated by commas (e.g., 1,3,5)[/red]"
1398 )
1400 def _manual_column_selection(self, field_name: str) -> str:
1401 """Manual column selection when no candidates found."""
1402 available_cols = list(self.df.columns)
1403 console.print(
1404 f"\n📋 [cyan]Available columns: {', '.join(available_cols)}[/cyan]"
1405 )
1407 while True:
1408 choice = Prompt.ask(f"Enter {field_name} column name (or 'skip')")
1410 if choice.lower() == "skip":
1411 return None
1412 elif choice in available_cols:
1413 return choice
1414 else:
1415 console.print(
1416 f"❌ [red]Column '{choice}' not found. Available: {', '.join(available_cols)}[/red]"
1417 )
1419 def generate_sample_records(
1420 self, selections: Dict, telemetry_type: str
1421 ) -> List[Dict]:
1422 """Generate sample telemetry records based on selections."""
1423 console.print(
1424 f"\n📝 [cyan]Generating sample {telemetry_type} telemetry records...[/cyan]"
1425 )
1427 samples = []
1429 # Create filter conditions to exclude rows with null values in critical fields
1430 critical_columns = [selections["element_name"], selections["timestamp"]]
1432 # Add value column if it's specified (not None)
1433 if selections.get("value"):
1434 critical_columns.append(selections["value"])
1436 # Create filter expression to exclude any rows with nulls in critical columns
1437 filter_expressions = []
1438 for col in critical_columns:
1439 filter_expressions.append(pl.col(col).is_not_null())
1441 # Combine all filter expressions with AND logic
1442 combined_filter = filter_expressions[0]
1443 for expr in filter_expressions[1:]:
1444 combined_filter = combined_filter & expr
1446 # Get valid rows (no nulls in critical fields) and take 5 samples
1447 valid_df = self.df.filter(combined_filter)
1449 if len(valid_df) == 0:
1450 console.print("⚠️ [red]No valid rows found for sample generation (all rows have null values in critical fields)[/red]")
1451 return []
1453 sample_df = valid_df.head(5)
1455 # Show summary of filtering
1456 total_rows = len(self.df)
1457 valid_rows = len(valid_df)
1458 excluded_rows = total_rows - valid_rows
1460 if excluded_rows > 0:
1461 console.print(
1462 f"📊 [dim]Filtered out {excluded_rows:,} rows with null values. Sampling from {valid_rows:,} valid rows.[/dim]"
1463 )
1465 for row in sample_df.iter_rows(named=True):
1466 if telemetry_type == "allocation":
1467 record = self._create_allocation_record(row, selections)
1468 else:
1469 record = self._create_unit_cost_record(row, selections)
1471 samples.append(record)
1473 self.analysis_results["sample_records"] = samples
1474 return samples
1476 def _create_allocation_record(self, row: Dict, selections: Dict) -> Dict:
1477 """Create allocation telemetry record from row data."""
1478 record = {
1479 "element-name": str(row.get(selections["element_name"], "UNKNOWN")),
1480 "value": float(row.get(selections["value"], 0))
1481 if selections["value"]
1482 else 0.0,
1483 "timestamp": self._format_timestamp(row.get(selections["timestamp"])),
1484 "granularity": self.analysis_results["temporal_analysis"].get(
1485 "selected_granularity",
1486 self.analysis_results["temporal_analysis"]["recommended_granularity"]
1487 ),
1488 }
1490 # Add filters (required field) - format as mapping from dimensions to lists of values
1491 filters = {}
1493 if selections.get("filters"):
1494 for filter_col in selections["filters"]:
1495 if filter_col in row and row[filter_col] is not None:
1496 # Convert to list format as required by CloudZero API
1497 filters[f"custom:{filter_col}"] = [str(row[filter_col])]
1499 # Add costformation dimensions if available and no user-selected filters
1500 if not filters and self.costformation_dimensions:
1501 for dim in self.costformation_dimensions[:3]:
1502 filters[f"custom:{dim}"] = ["PLACEHOLDER"]
1504 # Ensure filter is always present (required by CloudZero API)
1505 if filters:
1506 record["filters"] = filters
1507 else:
1508 # Use empty filter set as fallback (applies to all spend)
1509 record["filters"] = {}
1511 return record
1513 def _create_unit_cost_record(self, row: Dict, selections: Dict) -> Dict:
1514 """Create unit cost telemetry record from row data."""
1515 record = {
1516 "element-name": str(row.get(selections["element_name"], "UNKNOWN")),
1517 "value": float(row.get(selections["value"], 0))
1518 if selections["value"]
1519 else 0.0,
1520 "timestamp": self._format_timestamp(row.get(selections["timestamp"])),
1521 "granularity": self.analysis_results["temporal_analysis"].get(
1522 "selected_granularity",
1523 self.analysis_results["temporal_analysis"]["recommended_granularity"]
1524 ),
1525 }
1527 # Add associated costs
1528 if selections.get("associated_costs"):
1529 associated_costs = {}
1530 for cost_col in selections["associated_costs"]:
1531 if cost_col in row:
1532 associated_costs[cost_col] = (
1533 float(row[cost_col])
1534 if isinstance(row[cost_col], (int, float))
1535 else str(row[cost_col])
1536 )
1538 if associated_costs:
1539 record["associated-costs"] = associated_costs
1541 # Add costformation placeholder if no associated costs selected
1542 if not record.get("associated-costs") and self.costformation_dimensions:
1543 record["associated-costs"] = {
1544 dim: "PLACEHOLDER" for dim in self.costformation_dimensions[:3]
1545 }
1547 # Add filters (may be required for unit cost telemetry as well)
1548 # Use empty filter set as default (applies to all spend)
1549 record["filters"] = {}
1551 return record
1553 def _format_timestamp(self, timestamp_val) -> str:
1554 """Format timestamp value for telemetry record."""
1555 if timestamp_val is None:
1556 return datetime.now().isoformat()
1558 # If it's already a datetime, format it
1559 if hasattr(timestamp_val, "isoformat"):
1560 return timestamp_val.isoformat()
1562 # Try to parse string timestamps
1563 try:
1564 # Handle various string formats
1565 timestamp_str = str(timestamp_val)
1567 # Try common formats
1568 formats = [
1569 "%Y-%m-%d %H:%M:%S",
1570 "%Y-%m-%dT%H:%M:%S",
1571 "%Y-%m-%d",
1572 "%m/%d/%Y",
1573 "%Y/%m/%d",
1574 ]
1576 for fmt in formats:
1577 try:
1578 dt = datetime.strptime(timestamp_str, fmt)
1579 return dt.isoformat()
1580 except ValueError:
1581 continue
1583 # If no format matches, return as-is
1584 return timestamp_str
1586 except Exception:
1587 return datetime.now().isoformat()
1589 def display_results(
1590 self, selections: Dict, samples: List[Dict], telemetry_type: str
1591 ):
1592 """Display the final analysis results."""
1593 console.print("\n" + "=" * 80)
1594 console.print("📊 [bold cyan]CloudZero Telemetry Mapping Results[/bold cyan]")
1595 console.print("=" * 80)
1597 # Mapping summary
1598 console.print(f"\n🎯 [bold]Telemetry Type:[/bold] {telemetry_type.upper()}")
1599 selected_granularity = self.analysis_results['temporal_analysis'].get(
1600 'selected_granularity',
1601 self.analysis_results['temporal_analysis']['recommended_granularity']
1602 )
1603 console.print(
1604 f"📅 [bold]Selected Granularity:[/bold] {selected_granularity}"
1605 )
1607 # Selected columns table
1608 mapping_table = Table(title="Selected Column Mapping", box=box.ROUNDED)
1609 mapping_table.add_column("Field", style="cyan")
1610 mapping_table.add_column("Selected Column", style="green")
1611 mapping_table.add_column("Sample Value", style="yellow")
1613 sample_row = self.df.head(1).to_dicts()[0] if len(self.df) > 0 else {}
1615 for field, column in selections.items():
1616 if column:
1617 if isinstance(column, list):
1618 column_str = ", ".join(column)
1619 sample_val = ", ".join(
1620 str(sample_row.get(col, "N/A")) for col in column[:2]
1621 )
1622 else:
1623 column_str = column
1624 sample_val = str(sample_row.get(column, "N/A"))
1626 mapping_table.add_row(field.replace("_", "-"), column_str, sample_val)
1628 console.print("\n", mapping_table)
1630 # Sample records
1631 console.print(
1632 f"\n📝 [bold]Sample {telemetry_type.title()} Telemetry Records:[/bold]"
1633 )
1634 for i, record in enumerate(samples, 1):
1635 console.print(f"\n[dim]Record {i}:[/dim]")
1636 console.print(json.dumps(record, indent=2, default=str))
1638 def save_analysis_report(self):
1639 """Save comprehensive analysis report to JSON file."""
1640 output_file = (
1641 self.input_file.parent / f"customer_analysis_report_{self.input_file.stem}.json"
1642 )
1644 try:
1645 with open(output_file, "w") as f:
1646 json.dump(self.analysis_results, f, indent=2, default=str)
1648 console.print(
1649 f"\n💾 [green]Analysis report saved to: {output_file}[/green]"
1650 )
1651 console.print(
1652 "📄 [cyan]This report can be used for future telemetry creation.[/cyan]"
1653 )
1655 except Exception as e:
1656 console.print(f"❌ [red]Error saving analysis report: {e}[/red]")
1658 def run_interactive_analysis(self):
1659 """Run the complete interactive analysis workflow."""
1660 try:
1661 # Step 1: Determine telemetry type
1662 telemetry_type = self.get_telemetry_type()
1664 # Step 2: Analyze columns
1665 self.analyze_columns()
1667 # Step 3: Analyze temporal data
1668 self.analyze_temporal_columns()
1670 # Step 4: Get column selections
1671 selections = self.present_column_candidates(telemetry_type)
1673 # Step 5: Generate sample records
1674 samples = self.generate_sample_records(selections, telemetry_type)
1676 # Step 6: Display results
1677 self.display_results(selections, samples, telemetry_type)
1679 # Step 7: Save analysis report
1680 self.save_analysis_report()
1682 console.print("\n✨ [green]CloudZero telemetry analysis complete![/green]")
1684 except KeyboardInterrupt:
1685 console.print("\n\n⚠️ [yellow]Analysis interrupted by user[/yellow]")
1686 except Exception as e:
1687 console.print(f"\n❌ [red]Error during analysis: {e}[/red]")
1690def run_telemetry_analysis(
1691 df: pl.DataFrame,
1692 input_file: Path,
1693 costformation_file: Path = None,
1694 config_file: Path = None,
1695):
1696 """Main entry point for telemetry analysis."""
1697 analyzer = TelemetryAnalyzer(df, input_file, costformation_file, config_file)
1698 analyzer.run_interactive_analysis()