def _generate_sensor_logs()

in sagemaker/source/dataset/dataset_generator.py [0:0]

• 18 lines of code
• 4 McCabe index (conditional complexity)

    def _generate_sensor_logs(self, fleet_info_df):
        data = []
        for idx, stats in self.statistics_df.iterrows():
            threshold = SensorSeries.threshold(stats)

            for c in range(self.size_per_type):
                vehicle_id = self.size_per_type * idx + c
                timestamp = self.start_time

                sensor_series = iter(SensorSeries(stats))

                samples = []
                for t in range(self.num_sensor_readings):
                    voltage, current, resistance = next(sensor_series)
                    target = int(abs(voltage-current*resistance) > threshold)
                    samples.append(target)
                    timestamp_str = timestamp.strftime("%Y-%m-%dT%H:%M:%S.%f")
                    data.append([vehicle_id, target, timestamp_str, voltage, current])

                    timestamp += self.time_delta
                
        fleet_sensor_logs_df = pd.DataFrame(data=data, columns=['vehicle_id', 'target', 'timestamp', 'voltage', 'current'])
        return fleet_sensor_logs_df