def get()

in next_steps/kinesis_stream_connector/l4m_detector/src/synth_live_data_csv.py [0:0]

• 15 lines of code
• 5 McCabe index (conditional complexity)

    def get( self, t ):    
        if random.random() < anomaly_possibility:
            self.anomaly = Anomaly()
        
        value = self.daily_pattern.get(t)
        
        value += self.random_factor.get()

        is_anomaly = bool(self.anomaly)
        if self.anomaly:
            value += self.anomaly.get()
            if self.anomaly.proceed_time():
                self.anomaly = None
        
        metric_values = []
        for i, metric in enumerate(metrics):
            value = introduce_metric_from_upstream[i](value)
            metric_values.append(value)
        
        return metric_values, is_anomaly