Get min/max timestamps from data. Returns (min_timestamp, max_timestamp) or (None, None) if no data.
(self, csv_type: str = 'samples')
| 134 | return sorted(self.datadir.glob(pattern)) |
| 135 | |
| 136 | def get_time_range(self, csv_type: str = 'samples') -> Tuple[Optional[datetime], Optional[datetime]]: |
| 137 | """ |
| 138 | Get min/max timestamps from data. |
| 139 | Returns (min_timestamp, max_timestamp) or (None, None) if no data. |
| 140 | """ |
| 141 | conn = self.connect() |
| 142 | |
| 143 | # Get the CSV pattern for the type |
| 144 | patterns = { |
| 145 | 'samples': 'xcapture_samples_*.csv', |
| 146 | 'syscend': 'xcapture_syscend_*.csv', |
| 147 | 'iorqend': 'xcapture_iorqend_*.csv', |
| 148 | 'kstacks': 'xcapture_kstacks_*.csv', |
| 149 | 'ustacks': 'xcapture_ustacks_*.csv' |
| 150 | } |
| 151 | |
| 152 | pattern = patterns.get(csv_type, 'xcapture_samples_*.csv') |
| 153 | csv_path = str(self.datadir / pattern) |
| 154 | |
| 155 | try: |
| 156 | # Determine timestamp column based on CSV type |
| 157 | timestamp_col = 'TIMESTAMP' if csv_type == 'samples' else 'SYSC_ENTER_TIME' |
| 158 | |
| 159 | query = f""" |
| 160 | SELECT |
| 161 | MIN({timestamp_col}) as min_time, |
| 162 | MAX({timestamp_col}) as max_time |
| 163 | FROM read_csv_auto('{csv_path}') |
| 164 | WHERE {timestamp_col} IS NOT NULL |
| 165 | """ |
| 166 | |
| 167 | result = conn.execute(query).fetchone() |
| 168 | if result and result[0] and result[1]: |
| 169 | return (result[0], result[1]) |
| 170 | except Exception: |
| 171 | pass |
| 172 | |
| 173 | return (None, None) |
| 174 | |
| 175 | def validate_columns(self, columns: List[str], csv_type: str = None) -> List[str]: |
| 176 | """ |