Add Fitbit spo2 route and converter

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

@@ -106,6 +106,11 @@ public class FitbitRestSourceConnectorConfig extends RestSourceConnectorConfig {
   private static final String FITBIT_INTRADAY_HEART_RATE_VARIABILITY_TOPIC_DISPLAY = "Intraday heart rate variability topic";
   private static final String FITBIT_INTRADAY_HEART_RATE_VARIABILITY_TOPIC_DEFAULT = "connect_fitbit_intraday_heart_rate_variability";
 
+  private static final String FITBIT_INTRADAY_SPO2_TOPIC_CONFIG = "fitbit.intraday.spo2.topic";
+  private static final String FITBIT_INTRADAY_SPO2_TOPIC_DOC = "Topic for Fitbit intraday intraday_spo2";
+  private static final String FITBIT_INTRADAY_SPO2_TOPIC_DISPLAY = "Intraday spo2 topic";
+  private static final String FITBIT_INTRADAY_SPO2_TOPIC_DEFAULT = "connect_fitbit_intraday_spo2";
+
   private static final String FITBIT_BREATHING_RATE_TOPIC_CONFIG = "fitbit.breathing.rate.topic";
   private static final String FITBIT_BREATHING_RATE_TOPIC_DOC = "Topic for Fitbit breathing rate";
   private static final String FITBIT_BREATHING_RATE_TOPIC_DISPLAY = "Breathing rate topic";
@@ -343,6 +348,17 @@ public String toString() {
             Width.SHORT,
             FITBIT_INTRADAY_HEART_RATE_VARIABILITY_TOPIC_DISPLAY)
 
+        .define(FITBIT_INTRADAY_SPO2_TOPIC_CONFIG,
+            Type.STRING,
+            FITBIT_INTRADAY_SPO2_TOPIC_DEFAULT,
+            nonControlChar,
+            Importance.LOW,
+            FITBIT_INTRADAY_SPO2_TOPIC_DOC,
+            group,
+            ++orderInGroup,
+            Width.SHORT,
+            FITBIT_INTRADAY_SPO2_TOPIC_DISPLAY)
+
         .define(FITBIT_BREATHING_RATE_TOPIC_CONFIG,
             Type.STRING,
             FITBIT_BREATHING_RATE_TOPIC_DEFAULT,
@@ -511,6 +527,10 @@ public String getFitbitIntradayHeartRateVariabilityTopic() {
     return getString(FITBIT_INTRADAY_HEART_RATE_VARIABILITY_TOPIC_CONFIG);
   }
 
+  public String getFitbitIntradaySpo2Topic() {
+    return getString(FITBIT_INTRADAY_SPO2_TOPIC_CONFIG);
+  }
+
   public String getFitbitBreathingRateTopic() {
     return getString(FITBIT_BREATHING_RATE_TOPIC_CONFIG);
   }

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

@@ -0,0 +1,79 @@
+/*
+ * Copyright 2018 The Hyve
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.radarbase.connect.rest.fitbit.converter;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.confluent.connect.avro.AvroData;
+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.FitbitIntradaySpo2;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.time.Instant;
+import java.time.LocalDateTime;
+import java.time.ZonedDateTime;
+import java.util.stream.Stream;
+
+import static org.radarbase.connect.rest.util.ThrowingFunction.tryOrNull;
+
+public class FitbitIntradaySpo2AvroConverter extends FitbitAvroConverter {
+    private static final Logger logger = LoggerFactory.getLogger(FitbitIntradaySpo2AvroConverter.class);
+    private String spo2Topic;
+
+    public FitbitIntradaySpo2AvroConverter(AvroData avroData) {
+        super(avroData);
+    }
+
+    @Override
+    public void initialize(RestSourceConnectorConfig config) {
+        spo2Topic = ((FitbitRestSourceConnectorConfig) config).getFitbitIntradaySpo2Topic();
+        logger.info("Using intraday spo2 topic {}", spo2Topic);
+    }
+
+    @Override
+    protected Stream<TopicData> processRecords(FitbitRestRequest request, JsonNode root, double timeReceived) {
+        JsonNode spo2 = root;
+        if (spo2 == null || !spo2.isArray()) {
+            logger.warn("No Spo2 is provided for {}: {}", request, root);
+            return Stream.empty();
+        }
+        ZonedDateTime startDate = request.getDateRange().end();
+
+        return iterableToStream(spo2)
+                .filter(m -> m != null && m.isObject())
+                .map(m -> m.get("minutes"))
+                .filter(minutes -> minutes != null && minutes.isArray())
+                .flatMap(FitbitAvroConverter::iterableToStream)
+                .map(tryOrNull(minuteData -> parseSpo2(minuteData, startDate, timeReceived),
+                        (a, ex) -> logger.warn("Failed to convert spo2 from request {}, {}", request, a, ex)));
+    }
+
+    private TopicData parseSpo2(JsonNode minuteData, ZonedDateTime startDate, double timeReceived) {
+      Instant time = startDate.with(LocalDateTime.parse(minuteData.get("minute").asText())).toInstant();
+      Float value = (float) minuteData.get("value").asDouble();
+      if (value == null) {
+        return null;
+      }
+      FitbitIntradaySpo2 fitbitSpo2 = new FitbitIntradaySpo2(time.toEpochMilli() / 1000d,
+              timeReceived,
+              (float) value);
+      return new TopicData(time, spo2Topic, fitbitSpo2);
+    }
+}

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2018 The Hyve
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ */
+
+package org.radarbase.connect.rest.fitbit.route;
+
+import io.confluent.connect.avro.AvroData;
+import org.radarbase.connect.rest.fitbit.converter.FitbitIntradaySpo2AvroConverter;
+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;
+
+import java.util.stream.Stream;
+
+import static java.time.temporal.ChronoUnit.SECONDS;
+import java.time.Duration;
+
+
+public class FitbitIntradaySpo2Route extends FitbitPollingRoute {
+  private final FitbitIntradaySpo2AvroConverter converter;
+
+  public FitbitIntradaySpo2Route(FitbitRequestGenerator generator,
+      UserRepository userRepository, AvroData avroData) {
+    super(generator, userRepository, "intraday_spo2");
+    this.converter = new FitbitIntradaySpo2AvroConverter(avroData);
+  }
+
+  @Override
+  protected String getUrlFormat(String baseUrl) {
+    return baseUrl + "/1/user/%s/spo2/date/%s/%s/all.json";
+  }
+
+  protected Stream<FitbitRestRequest> createRequests(User user) {
+    return startDateGenerator(getOffset(user).plus(ONE_SECOND).truncatedTo(SECONDS))
+        .map(dateRange -> newRequest(user, dateRange,
+            user.getExternalUserId(), DATE_FORMAT.format(dateRange.start()), DATE_FORMAT.format(dateRange.end())));
+  }
+
+  /** Limit range to 30 days as documented here: https://dev.fitbit.com/build/reference/web-api/intraday/get-spo2-intraday-by-interval/ */
+  @Override
+  Duration getDateRangeInterval() {
+    return THIRTY_DAYS;
+  }
+
+  @Override
+  public FitbitIntradaySpo2AvroConverter converter() {
+    return converter;
+  }
+}