Merge pull request #121 from RADAR-base/fix/breathing-skinTemperature

Fix breathing and skin temperature convertors

Author: peyman-mohtashami

buildSrc/src/main/kotlin/Versions.kt

@@ -22,7 +22,7 @@ object Versions {
     const val okhttp = "4.11.0"
 
     const val firebaseAdmin = "9.1.0"
-    const val radarSchemas = "0.8.7-SNAPSHOT"
+    const val radarSchemas = "0.8.6"
     const val ktor = "2.3.5"
 
     const val junit = "5.9.3"

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

@@ -26,9 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZonedDateTime;
+import java.time.*;
 import java.util.stream.Stream;
 
 import static org.radarbase.connect.rest.util.ThrowingFunction.tryOrNull;
@@ -58,23 +56,22 @@ protected Stream<TopicData> processRecords(FitbitRestRequest request, JsonNode r
 
         return iterableToStream(br)
             .filter(m -> m != null && m.isObject())
-            .flatMap(FitbitAvroConverter::iterableToStream)
             .map(tryOrNull(m -> parseBr(m, startDate, timeReceived),
                 (a, ex) -> logger.warn("Failed to convert breathing rate from request {}, {}", request, a, ex)));
     }
 
     private TopicData parseBr(JsonNode data, ZonedDateTime startDate, double timeReceived) {
-      Instant time = startDate.with(LocalDateTime.parse(data.get("dateTime").asText())).toInstant();
+      Instant time = LocalDate.parse(data.get("dateTime").asText()).atStartOfDay(ZoneOffset.UTC).toInstant();
       JsonNode value = data.get("value");
       if (value == null || !value.isObject()) {
         return null;
       }
       FitbitBreathingRate fitbitBr = new FitbitBreathingRate(time.toEpochMilli() / 1000d,
               timeReceived,
-              (float) value.get("deepSleepBrSummary").get("breathingRate").asDouble(),
-              (float) value.get("remSleepBrSummary").get("breathingRate").asDouble(),
-              (float) value.get("fullSleepBrSummary").get("breathingRate").asDouble(),
-              (float) value.get("lightSleepBrSummary").get("breathingRate").asDouble());
+              (float) value.get("deepSleepSummary").get("breathingRate").asDouble(),
+              (float) value.get("remSleepSummary").get("breathingRate").asDouble(),
+              (float) value.get("fullSleepSummary").get("breathingRate").asDouble(),
+              (float) value.get("lightSleepSummary").get("breathingRate").asDouble());
       return new TopicData(time, breathingRateTopic, fitbitBr);
     }
 }

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

@@ -26,9 +26,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.time.Instant;
-import java.time.LocalDateTime;
-import java.time.ZonedDateTime;
+import java.time.*;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.stream.Stream;
@@ -68,24 +66,23 @@ protected Stream<TopicData> processRecords(FitbitRestRequest request, JsonNode r
 
         return iterableToStream(tempSkin)
                 .filter(m -> m != null && m.isObject())
-                .flatMap(FitbitAvroConverter::iterableToStream)
                 .map(tryOrNull(m -> parseTempSkin(m, startDate, timeReceived),
                         (a, ex) -> logger.warn("Failed to convert skin temperature from request {}, {}", request, a, ex)));
     }
 
     private TopicData parseTempSkin(JsonNode data, ZonedDateTime startDate, double timeReceived) {
-      Instant time = startDate.with(LocalDateTime.parse(data.get("dateTime").asText())).toInstant();
+      Instant time = LocalDate.parse(data.get("dateTime").asText()).atStartOfDay(ZoneOffset.UTC).toInstant();
       JsonNode value = data.get("value");
       if (value == null || !value.isObject()) {
         return null;
       }
-      String logType = data.get("level").asText();
-      FitbitSkinTemperature fitbitHrv = new FitbitSkinTemperature(
+      String logType = data.get("logType").asText();
+      FitbitSkinTemperature fitbitSkinTemperature = new FitbitSkinTemperature(
           time.toEpochMilli() / 1000d,
           timeReceived,
           (float) value.get("nightlyRelative").asDouble(),
           LOG_TYPE_MAP.getOrDefault(logType, FitbitSkinTemperatureLogType.UNKNOWN)
       );
-      return new TopicData(time, skinTemperatureTopic, fitbitHrv);
+      return new TopicData(time, skinTemperatureTopic, fitbitSkinTemperature);
     }
 }

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

@@ -40,7 +40,7 @@ public FitbitSkinTemperatureRoute(FitbitRequestGenerator generator,
 
   @Override
   protected String getUrlFormat(String baseUrl) {
-    return baseUrl + "/1/user/%s/temp/skin/date/%s/%s/.json";
+    return baseUrl + "/1/user/%s/temp/skin/date/%s/%s.json";
   }
 
   protected Stream<FitbitRestRequest> createRequests(User user) {
@@ -59,4 +59,4 @@ Duration getDateRangeInterval() {
   public FitbitSkinTemperatureAvroConverter converter() {
     return converter;
   }
-}
+}