def get_data_at_time()

in aiops/MicroAgents/utils/process.py [0:0]

• 12 lines of code
• 11 McCabe index (conditional complexity)

def get_data_at_time(df, start_time=None, end_time=None, time_window={'minutes':1}, time_col='m_timestamp', date_format= '%Y-%m-%dT%H:%M:%S.%fZ'):

    if start_time is None and end_time is None:
        raise ValueError("start_time and end_time cannot be both None")
    
    if start_time is not None and isinstance(start_time, str):
        start_time = datetime.strptime(start_time, date_format)
    if end_time is not None and isinstance(end_time, str):
        end_time = datetime.strptime(end_time, date_format)
    
    if time_window is not None and end_time is not None:
        start_time = end_time - timedelta(**time_window)
    if time_window is not None and start_time is not None:
        end_time = start_time + timedelta(**time_window)
        
    return df.filter((pl.col(time_col) >= start_time) & (pl.col(time_col) < end_time))