Merge pull request #48 from RADAR-base/release-0.2.4

Release 0.2.4

Author: yatharthranjan

README.md

@@ -74,6 +74,8 @@ your Fitbit App client ID and client secret. The following tables shows the poss
 <td>fitbit.time.zone.topic</td></td><td>Topic for Fitbit profile time zone</td></td><td>string</td></td><td>connect_fitbit_time_zone</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 <tr>
 <td>fitbit.activity.log.topic</td></td><td>Topic for Fitbit activity log.</td></td><td>string</td></td><td>connect_fitbit_activity_log</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
+<tr>
+<td>fitbit.intraday.calories.topic</td></td><td>Topic for Fitbit intraday calories</td></td><td>string</td></td><td>connect_fitbit_intraday_calories</td></td><td>non-empty string without control characters</td></td><td>low</td></td></tr>
 </tbody></table>
 
 Now you can run a full Kafka stack using

build.gradle

@@ -11,7 +11,7 @@ subprojects {
     apply plugin: 'java-library'
 
     group = 'org.radarbase'
-    version = '0.2.3'
+    version = '0.2.4'
 
     sourceCompatibility = 1.8
     targetCompatibility = 1.8

kafka-connect-fitbit-source/build.gradle

@@ -1,7 +1,8 @@
 dependencies {
     api project(':kafka-connect-rest-source')
     api group: 'io.confluent', name: 'kafka-connect-avro-converter', version: confluentVersion
-    api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.5.1'
+    api group: 'org.radarcns', name: 'radar-schemas-commons', version: '0.5.3'
+
 
     implementation group: 'com.fasterxml.jackson.dataformat', name: 'jackson-dataformat-yaml', version: jacksonVersion
     implementation group: 'com.fasterxml.jackson.datatype', name: 'jackson-datatype-jsr310', version: jacksonVersion

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/FitbitRestSourceConnectorConfig.java

@@ -118,6 +118,11 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DEFAULT = "connect_fitbit_activity_log";
   private static final String FITBIT_ACTIVITY_LOG_TOPIC_DISPLAY = "Activity log topic";
 
+  private static final String FITBIT_INTRADAY_CALORIES_TOPIC_CONFIG = "fitbit.intraday.calories.topic";
+  private static final String FITBIT_INTRADAY_CALORIES_TOPIC_DOC = "Topic for Fitbit intraday calories";
+  private static final String FITBIT_INTRADAY_CALORIES_TOPIC_DISPLAY = "Intraday calories topic";
+  private static final String FITBIT_INTRADAY_CALORIES_TOPIC_DEFAULT = "connect_fitbit_intraday_calories";
+
   private final UserRepository userRepository;
   private final Headers clientCredentials;
 
@@ -302,6 +307,17 @@ public String toString() {
             ++orderInGroup,
             Width.SHORT,
             FITBIT_ACTIVITY_LOG_TOPIC_DISPLAY)
+
+        .define(FITBIT_INTRADAY_CALORIES_TOPIC_CONFIG,
+            Type.STRING,
+            FITBIT_INTRADAY_CALORIES_TOPIC_DEFAULT,
+            nonControlChar,
+            Importance.LOW,
+            FITBIT_INTRADAY_CALORIES_TOPIC_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_INTRADAY_CALORIES_TOPIC_DISPLAY)
         ;
   }
 
@@ -378,4 +394,8 @@ public Duration getPollIntervalPerUser() {
   public Duration getTooManyRequestsCooldownInterval() {
     return Duration.ofHours(1);
   }
+
+  public String getFitbitIntradayCaloriesTopic() {
+    return getString(FITBIT_INTRADAY_CALORIES_TOPIC_CONFIG);
+  }
 }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/converter/FitbitIntradayCaloriesAvroConverter.java

@@ -0,0 +1,79 @@
+package org.radarbase.connect.rest.fitbit.converter;
+
+import static org.radarbase.connect.rest.util.ThrowingFunction.tryOrNull;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.confluent.connect.avro.AvroData;
+import java.time.Instant;
+import java.time.LocalTime;
+import java.time.ZonedDateTime;
+import java.util.stream.Stream;
+import org.radarbase.connect.rest.RestSourceConnectorConfig;
+import org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig;
+import org.radarbase.connect.rest.fitbit.request.FitbitRestRequest;
+import org.radarcns.connector.fitbit.FitbitIntradayCalories;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FitbitIntradayCaloriesAvroConverter extends FitbitAvroConverter {
+
+  private static final Logger logger =
+      LoggerFactory.getLogger(FitbitIntradayCaloriesAvroConverter.class);
+
+  private String caloriesTopic;
+
+  public FitbitIntradayCaloriesAvroConverter(AvroData avroData) {
+    super(avroData);
+  }
+
+  @Override
+  protected Stream<TopicData> processRecords(
+      FitbitRestRequest request, JsonNode root, double timeReceived) {
+    JsonNode intraday = root.get("activities-calories-intraday");
+    if (intraday == null) {
+      return Stream.empty();
+    }
+
+    JsonNode dataset = intraday.get("dataset");
+    if (dataset == null) {
+      return Stream.empty();
+    }
+
+    int interval = getRecordInterval(intraday, 60);
+
+    // Used as the date to convert the local times in the dataset to absolute times.
+    ZonedDateTime startDate = request.getDateRange().end();
+
+    return iterableToStream(dataset)
+        .map(
+            tryOrNull(
+                activity -> {
+                  Instant time =
+                      startDate.with(LocalTime.parse(activity.get("time").asText())).toInstant();
+
+                  FitbitIntradayCalories calories =
+                      new FitbitIntradayCalories(
+                          time.toEpochMilli() / 1000d,
+                          timeReceived,
+                          interval,
+                          activity.get("value").asDouble(),
+                          activity.get("level").asInt(),
+                          activity.get("mets").asDouble());
+
+                  return new TopicData(time, caloriesTopic, calories);
+                },
+                (a, ex) ->
+                    logger.warn(
+                        "Failed to convert calories from request {} of user {}, {}",
+                        request.getRequest().url(),
+                        request.getUser(),
+                        a,
+                        ex)));
+  }
+
+  @Override
+  public void initialize(RestSourceConnectorConfig config) {
+    caloriesTopic = ((FitbitRestSourceConnectorConfig) config).getFitbitIntradayCaloriesTopic();
+    logger.info("Using calories topic {}", caloriesTopic);
+  }
+}

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/request/FitbitRequestGenerator.java

@@ -35,6 +35,7 @@
 import org.radarbase.connect.rest.RestSourceConnectorConfig;
 import org.radarbase.connect.rest.fitbit.FitbitRestSourceConnectorConfig;
 import org.radarbase.connect.rest.fitbit.route.FitbitActivityLogRoute;
+import org.radarbase.connect.rest.fitbit.route.FitbitIntradayCaloriesRoute;
 import org.radarbase.connect.rest.fitbit.route.FitbitIntradayHeartRateRoute;
 import org.radarbase.connect.rest.fitbit.route.FitbitIntradayStepsRoute;
 import org.radarbase.connect.rest.fitbit.route.FitbitSleepRoute;
@@ -90,6 +91,7 @@ private List<RequestRoute> getRoutes(FitbitRestSourceConnectorConfig config) {
     if (config.hasIntradayAccess()) {
       localRoutes.add(new FitbitIntradayStepsRoute(this, userRepository, avroData));
       localRoutes.add(new FitbitIntradayHeartRateRoute(this, userRepository, avroData));
+      localRoutes.add(new FitbitIntradayCaloriesRoute(this, userRepository, avroData));
     }
     return localRoutes;
   }

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitIntradayCaloriesRoute.java

@@ -0,0 +1,47 @@
+package org.radarbase.connect.rest.fitbit.route;
+
+import static java.time.temporal.ChronoUnit.MINUTES;
+
+import io.confluent.connect.avro.AvroData;
+import java.util.stream.Stream;
+import org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter;
+import org.radarbase.connect.rest.fitbit.converter.FitbitIntradayCaloriesAvroConverter;
+import org.radarbase.connect.rest.fitbit.request.FitbitRequestGenerator;
+import org.radarbase.connect.rest.fitbit.request.FitbitRestRequest;
+import org.radarbase.connect.rest.fitbit.user.User;
+import org.radarbase.connect.rest.fitbit.user.UserRepository;
+
+public class FitbitIntradayCaloriesRoute extends FitbitPollingRoute {
+
+  private final FitbitIntradayCaloriesAvroConverter caloriesAvroConverter;
+
+  public FitbitIntradayCaloriesRoute(
+      FitbitRequestGenerator generator, UserRepository userRepository, AvroData avroData) {
+    super(generator, userRepository, "intraday_calories");
+    caloriesAvroConverter = new FitbitIntradayCaloriesAvroConverter(avroData);
+  }
+
+  @Override
+  protected Stream<FitbitRestRequest> createRequests(User user) {
+    return startDateGenerator(this.getOffset(user).plus(ONE_MINUTE).truncatedTo(MINUTES))
+        .map(
+            dateRange ->
+                newRequest(
+                    user,
+                    dateRange,
+                    user.getExternalUserId(),
+                    DATE_FORMAT.format(dateRange.start()),
+                    TIME_FORMAT.format(dateRange.start()),
+                    TIME_FORMAT.format(dateRange.end())));
+  }
+
+  @Override
+  protected String getUrlFormat(String baseUrl) {
+    return baseUrl + "/1/user/%s/activities/calories/date/%s/1d/1min/time/%s/%s.json?timezone=UTC";
+  }
+
+  @Override
+  public PayloadToSourceRecordConverter converter() {
+    return caloriesAvroConverter;
+  }
+}

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitIntradayStepsRoute.java

@@ -29,8 +29,6 @@
 import org.radarbase.connect.rest.fitbit.user.UserRepository;
 
 public class FitbitIntradayStepsRoute extends FitbitPollingRoute {
-  private static final TemporalAmount ONE_MINUTE = MINUTES.getDuration();
-
   private final FitbitIntradayStepsAvroConverter converter;
 
   public FitbitIntradayStepsRoute(FitbitRequestGenerator generator,

kafka-connect-fitbit-source/src/main/java/org/radarbase/connect/rest/fitbit/route/FitbitPollingRoute.java

@@ -19,6 +19,7 @@
 
 import static java.time.ZoneOffset.UTC;
 import static java.time.temporal.ChronoUnit.DAYS;
+import static java.time.temporal.ChronoUnit.MINUTES;
 import static java.time.temporal.ChronoUnit.NANOS;
 import static java.time.temporal.ChronoUnit.SECONDS;
 import static org.radarbase.connect.rest.converter.PayloadToSourceRecordConverter.MIN_INSTANT;
@@ -95,6 +96,7 @@ public abstract class FitbitPollingRoute implements PollingRequestRoute {
   protected static final Duration ONE_DAY = DAYS.getDuration();
   protected static final Duration ONE_NANO = NANOS.getDuration();
   protected static final TemporalAmount ONE_SECOND = SECONDS.getDuration();
+  protected static final TemporalAmount ONE_MINUTE = MINUTES.getDuration();
 
   private static final Logger logger = LoggerFactory.getLogger(FitbitSleepRoute.class);