human
1.3. Retrieve test metrics
Define a helper function that retrieves test metrics from Cloud Monitoring
 ... What is a the Python code you requested?
assistant
#Here is the Python code you requested.
def retrieve_metrics(client, project_id, start_time, end_time, model, model_version, test_id, log_name):
    """
    Retrieves test metrics from Cloud Monitoring.
    """
    def _get_aipp_metric(metric_type: str, labels: List[str]=[], metric_name=None)-> pd.DataFrame:
        """
        Retrieves a specified AIPP metric.
        """
        query = Query(client, project_id, metric_type=metric_type)
        query = query.select_interval(end_time, start_time)
        query = query.select_resources(model_id=model)
        query = query.select_resources(version_id=model_version)
        
        if metric_name:
            labels = ['metric'] + labels 
        df = query.as_dataframe(labels=labels)
        
        if not df.empty:
            if metric_name:
                df.columns.set_levels([metric_name], level=0, inplace=True)
            df = df.set_index(df.index.round('T'))
        
        return df
    
    def _get_locust_metric(metric_type: str, labels: List[str]=[], metric_name=None)-> pd.DataFrame:
        """
        Retrieves a specified custom log-based metric.
        """
        query = Query(client, project_id, metric_type=metric_type)
        query = query.select_interval(end_time, start_time)
        query = query.select_metrics(log=log_name)
        query = query.select_metrics(test_id=test_id)
        
        if metric_name:
            labels = ['metric'] + labels 
        df = query.as_dataframe(labels=labels)
        
        if not df.empty: 
            if metric_name:
                df.columns.set_levels([metric_name], level=0, inplace=True)
            df = df.apply(lambda row: [metric.mean for metric in row])
            df = df.set_index(df.index.round('T'))
        
        return df
    
    # Retrieve GPU duty cycle
    metric_type = 'ml.googleapis.com/prediction/online/accelerator/duty_cycle'
    metric = _get_aipp_metric(metric_type, ['replica_id', 'signature'], 'duty_cycle')
    df = metric

    # Retrieve CPU utilization
    metric_type =...