[WiP] Huawei API integration

.github/workflows/main.yml

@@ -22,6 +22,11 @@ env:
     'build_file': 'kafka-connect-oura-source/Dockerfile',
     'authors': 'Pim van Nierop <pim@thehyve.nl>, Pauline Conde <pauline.conde@kcl.ac.uk>',
     'description': 'RADAR-base Oura connector application'
+    },{
+    'name': 'kafka-connect-huawei-source',
+    'build_file': 'kafka-connect-huawei-source/Dockerfile',
+    'authors': 'Yatharth Ranjan <yatharth.ranjan@kcl.ac.uk>, Yatharth Ranjan <yatharthranjan@onsentia.com>',
+    'description': 'RADAR-base Huawei Health Kit connector application'
     }]
 
 jobs:

.github/workflows/release.yml

@@ -16,8 +16,13 @@ env:
     },{
     'name': 'kafka-connect-oura-source',
     'build_file': 'kafka-connect-oura-source/Dockerfile',
-    'authors': 'Pauline Conde <pauline.conde@kcl.ac.uk>, Yatharth Ranjan <yatharth.ranjan@kcl.ac.uk>',
+    'authors': 'Pauline Conde <pauline.conde@kcl.ac.uk>, Yatharth Ranjan <yatharth.ranjan@kcl.ac.uk>, Yatharth Ranjan <yatharthranjan@onsentia.com>',
     'description': 'RADAR-base Oura connector application'
+    },{
+    'name': 'kafka-connect-huawei-source',
+    'build_file': 'kafka-connect-huawei-source/Dockerfile',
+    'authors': 'Yatharth Ranjan <yatharth.ranjan@kcl.ac.uk>, Yatharth Ranjan <yatharthranjan@onsentia.com>',
+    'description': 'RADAR-base Huawei Health Kit connector application'
     }]
 
 jobs:

README.md

@@ -1,15 +1,16 @@
 # Kafka Connect REST Source and Fitbit Source
 
 This project contains a Kafka Connect source connector for a general REST API, for
-specific Fitbit and Oura devices. The documentation of the Kafka Connect REST source still needs to
-be done.
+specific Fitbit, Oura, and Huawei Health Kit devices. The documentation of the Kafka Connect REST
+source still needs to be done.
 
 <!-- TOC -->
 
 * [Kafka Connect REST Source and Fitbit Source](#kafka-connect-rest-source-and-fitbit-source)
   * [Fitbit source connector](#fitbit-source-connector)
     * [Installation](#installation)
     * [Usage](#usage)
+  * [Huawei Health Kit source connector](#huawei-health-kit-source-connector)
   * [Sentry monitoring](#sentry-monitoring)
   * [Contributing](#contributing)
 
@@ -208,9 +209,66 @@ sequenceDiagram
   connector ->> connector: Update offset times
 ```
 
+## Huawei Health Kit source connector
+
+The `kafka-connect-huawei-source` module polls the
+[Huawei Health Kit Data API](https://developer.huawei.com/consumer/en/doc/HMSCore-References/rest-overview-0000001254420693)
+for the data types documented in the
+[`radar-huawei-connector` schema specification](https://github.com/RADAR-base/RADAR-Schemas/blob/huawei_schemas/specifications/connector/radar-huawei-connector-1.0.0.yml)
+(RADAR-Schemas, `huawei_schemas` branch) — activity records, continuous/instantaneous sample
+statistics (steps, distance, calories, heart rate, SpO2, blood pressure, breathing rate, ECG,
+sleep stages, and more), health records (ambulatory blood pressure, heart rate alerts,
+hyperthermia, low SpO2 alerts, menstrual cycle, sleep), and daily summaries. It follows the same
+`rest.source.*`, `huawei.api.client`/`huawei.api.secret`, and `huawei.user.repository.*`
+configuration conventions as the Fitbit and Oura connectors above, plus one
+`huawei.<data-type>.enabled` / `huawei.<data-type>.topic` pair per Huawei data type — see
+`org.radarbase.huawei.route.HuaweiRouteFactory` for the full list of `<data-type>` keys and their
+default topic names, and `docker/source-huawei.properties.template` for a minimal example.
+
+This connector requires a
+[published `radar-schemas-commons` build containing the `huawei_schemas` branch](https://github.com/RADAR-base/RADAR-Schemas/tree/huawei_schemas)
+(currently `0.9.0-SNAPSHOT`) to be resolvable from one of the repositories declared in
+`huawei-library/build.gradle` / `kafka-connect-huawei-source/build.gradle.kts`.
+
+### Testing locally
+
+The easiest way to try out the Huawei connector without standing up a
+`rest-source-authorizer` webservice is the file-based
+`org.radarbase.connect.rest.huawei.user.HuaweiYamlUserRepository`, which reads one YAML file per
+user from a local directory, mirroring the Fitbit `YamlUserRepository` above.
+
+1. [Register a Huawei Health Kit OAuth 2.0 app](https://developer.huawei.com/consumer/en/doc/HMSCore-Guides/config-agc-0000001050170137)
+   and obtain an access token and refresh token for one test user by hand, using Huawei's
+   [OAuth 2.0 authorization code flow](https://developer.huawei.com/consumer/en/doc/HMSCore-Guides/authorization-code-0000001053629189).
+2. Copy `docker/huawei-user.yml.template` to a file in `docker/users/` (e.g. `docker/users/test.yml`)
+   and fill in the `externalUserId`, `oauth2.accessToken`, and `oauth2.refreshToken` fields.
+3. Copy `docker/source-huawei.properties.template` to `docker/source-huawei.properties`, set
+   `huawei.api.client` / `huawei.api.secret` to your Huawei app's client ID and secret, and change
+   `huawei.user.repository.class` to `org.radarbase.connect.rest.huawei.user.HuaweiYamlUserRepository`
+   (the `huawei.user.repository.url`/`.client.id`/`.client.secret`/`.oauth2.token.url` properties
+   are only used by the webservice-based repository and can be left as-is or removed).
+4. Run the full stack with `docker-compose up -d --build` and inspect the connector's progress with
+   `docker-compose logs -f radar-huawei-connector`.
+5. To inspect the data coming out of a specific route, run, for example:
+
+   ```shell
+   docker-compose exec schema-registry-1 kafka-avro-console-consumer \
+     --bootstrap-server kafka-1:9092,kafka-2:9092,kafka-3:9092 \
+     --from-beginning \
+     --topic connect_huawei_activity_record
+   ```
+
+   (replace the topic with any `huawei.<data-type>.topic` default from
+   `org.radarbase.huawei.route.HuaweiRouteFactory`).
+
+For a full RADAR-base deployment, use the webservice-based
+`org.radarbase.connect.rest.huawei.user.HuaweiServiceUserRepository` (the default) against a
+`rest-source-authorizer` instance instead, following the same pattern as the Fitbit connector's
+ManagementPortal setup above.
+
 ## Sentry monitoring
 
-To enable Sentry monitoring for the generic REST, Fitbit, or Oura source connector service:
+To enable Sentry monitoring for the generic REST, Fitbit, Oura, or Huawei source connector service:
 
 1. Set a `SENTRY_DSN` environment variable that points to the desired Sentry DSN.
 2. (Optional) Set the `SENTRY_LOG_LEVEL` environment variable to control the minimum log level of

docker-compose.yml

@@ -4,6 +4,7 @@ version: "2.4"
 volumes:
   fitbit-logs: {}
   oura-logs: {}
+  huawei-logs: {}
 
 services:
   #---------------------------------------------------------------------------#
@@ -231,3 +232,49 @@ services:
       # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000'
       # SENTRY_ATTACHSTACKTRACE: true
       # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest
+
+  #---------------------------------------------------------------------------#
+  # RADAR Huawei connector                                                    #
+  #---------------------------------------------------------------------------#
+  radar-huawei-connector:
+    build:
+      context: .
+      dockerfile: ./kafka-connect-huawei-source/Dockerfile
+    image: radarbase/radar-connect-huawei-source
+    restart: on-failure
+    volumes:
+      - ./docker/source-huawei.properties:/etc/kafka-connect/source-huawei.properties
+      - ./docker/users:/var/lib/kafka-connect-huawei-source/users
+      - huawei-logs:/var/lib/kafka-connect-huawei-source/logs
+    depends_on:
+      - zookeeper-1
+      - zookeeper-2
+      - zookeeper-3
+      - kafka-1
+      - kafka-2
+      - kafka-3
+      - schema-registry-1
+    environment:
+      CONNECT_BOOTSTRAP_SERVERS: PLAINTEXT://kafka-1:9092,PLAINTEXT://kafka-2:9092,PLAINTEXT://kafka-3:9092
+      CONNECT_REST_PORT: 8083
+      CONNECT_GROUP_ID: "default"
+      CONNECT_CONFIG_STORAGE_TOPIC: "default.config"
+      CONNECT_OFFSET_STORAGE_TOPIC: "default.offsets"
+      CONNECT_STATUS_STORAGE_TOPIC: "default.status"
+      CONNECT_KEY_CONVERTER: "io.confluent.connect.avro.AvroConverter"
+      CONNECT_VALUE_CONVERTER: "io.confluent.connect.avro.AvroConverter"
+      CONNECT_KEY_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry-1:8081"
+      CONNECT_VALUE_CONVERTER_SCHEMA_REGISTRY_URL: "http://schema-registry-1:8081"
+      CONNECT_INTERNAL_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_INTERNAL_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
+      CONNECT_OFFSET_STORAGE_FILE_FILENAME: "/var/lib/kafka-connect-huawei-source/logs/connect.offsets"
+      CONNECT_REST_ADVERTISED_HOST_NAME: "radar-huawei-connector"
+      CONNECT_ZOOKEEPER_CONNECT: zookeeper-1:2181,zookeeper-2:2181,zookeeper-3:2181
+      CONNECTOR_PROPERTY_FILE_PREFIX: "source-huawei"
+      KAFKA_HEAP_OPTS: "-Xms256m -Xmx768m"
+      KAFKA_BROKERS: 3
+      CONNECT_LOG4J_LOGGERS: "org.reflections=ERROR"
+      # SENTRY_LOG_LEVEL: 'ERROR'
+      # SENTRY_DSN: 'https://000000000000.ingest.de.sentry.io/000000000000'
+      # SENTRY_ATTACHSTACKTRACE: true
+      # SENTRY_STACKTRACE_APP_PACKAGES: io.confluent.connect,org.radarbase.connect.rest

docker/huawei-user.yml.template

@@ -0,0 +1,24 @@
+---
+# Unique user key
+id: test
+# Project ID to be used in org.radarcns.kafka.ObservationKey record keys
+projectId: radar-test
+# User ID to be used in org.radarcns.kafka.ObservationKey record keys
+userId: test
+# Source ID to be used in org.radarcns.kafka.ObservationKey record keys
+sourceId: huawei-watch
+# Date from when to collect data.
+startDate: 2018-08-06T00:00:00Z
+# Date until when to collect data.
+endDate: 2099-01-01T00:00:00Z
+# Huawei user ID as returned by the Huawei OAuth 2.0 authentication procedure
+externalUserId: ?
+oauth2:
+  # Huawei Health Kit OAuth 2.0 access token as returned by the Huawei authentication procedure
+  accessToken: ?
+  # Huawei Health Kit OAuth 2.0 refresh token as returned by the Huawei authentication procedure
+  refreshToken: ?
+  # Optional expiry time of the access token. If absent, it will be estimated to one hour
+  # when the source connector starts. When an authentication error occurs, a new access token will
+  # be fetched regardless of the value in this field.
+  #expiresAt: 2018-08-06T00:00:00Z

docker/source-huawei.properties.template

@@ -0,0 +1,18 @@
+name=radar-huawei-source
+connector.class=org.radarbase.connect.rest.huawei.HuaweiSourceConnector
+tasks.max=4
+rest.source.base.url=https://health-api.cloud.huawei.com/healthkit/v1
+rest.source.poll.interval.ms=5000
+huawei.api.client=?
+huawei.api.secret=?
+# For local testing without a rest-source-authorizer webservice, use the file-based
+# repository instead, backed by per-user YAML files under huawei.user.dir - see
+# docker/huawei-user.yml.template and the "Testing locally" section in README.md.
+#huawei.user.repository.class=org.radarbase.connect.rest.huawei.user.HuaweiYamlUserRepository
+#huawei.user.dir=/var/lib/kafka-connect-huawei-source/users
+
+huawei.user.repository.class=org.radarbase.connect.rest.huawei.user.HuaweiServiceUserRepository
+huawei.user.repository.url=http://localhost:8080/
+huawei.user.repository.client.id=radar_huawei_connector
+huawei.user.repository.client.secret=
+huawei.user.repository.oauth2.token.url=

gradle/libs.versions.toml

@@ -16,6 +16,9 @@ sentryOpenTelemetryAgent = "8.36.0"
 okhttp = "4.12.0"
 firebaseAdmin = "9.8.0"
 radarSchemas = "0.8.16"
+# Huawei connector schemas are not yet released; pin to the published snapshot until a stable
+# radar-schemas-commons release containing the huawei_schemas branch is cut.
+radarSchemasHuawei = "0.9.0-SNAPSHOT"
 # @pin Upgrade to 3.x.x requires kotlin v2 minimum
 ktor = "2.3.13"
 wiremock = "3.0.1"
@@ -28,6 +31,7 @@ lz4 = "1.10.1"
 lz4 = { module = "at.yawk.lz4:lz4-java", version.ref = "lz4" }
 radar-commons-kotlin = { module = "org.radarbase:radar-commons-kotlin", version.ref = "radarCommons" }
 radar-schemas-commons = { module = "org.radarbase:radar-schemas-commons", version.ref = "radarSchemas" }
+radar-schemas-commons-huawei = { module = "org.radarbase:radar-schemas-commons", version.ref = "radarSchemasHuawei" }
 kafka-connect-api = "org.apache.kafka:connect-api:7.8.1-ce"
 kafka-connect-avro-converter = { module = "io.confluent:kafka-connect-avro-converter", version.ref = "confluent" }
 okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" }
@@ -55,6 +59,7 @@ mockito-core = { module = "org.mockito:mockito-core", version.ref = "mockito" }
 wiremock = { module = "com.github.tomakehurst:wiremock", version.ref = "wiremock" }
 kotlin-test = { module = "org.jetbrains.kotlin:kotlin-test", version.ref = "kotlin" }
 kotlin-test-junit = { module = "org.jetbrains.kotlin:kotlin-test-junit", version.ref = "kotlin" }
+kotlin-test-junit5 = { module = "org.jetbrains.kotlin:kotlin-test-junit5", version.ref = "kotlin" }
 
 [plugins]
 radar-root-project = { id = "org.radarbase.radar-root-project", version.ref = "radarCommons" }

huawei-library/build.gradle

@@ -0,0 +1,70 @@
+
+group = 'org.radarbase'
+version = '0.0.1'
+
+apply plugin: 'maven-publish'
+
+repositories {
+    // Use jcenter for resolving dependencies.
+    // You can declare any Maven/Ivy/file repository here.
+    mavenCentral()
+
+    // Prefer a locally-published snapshot (e.g. built by hand from the RADAR-Schemas
+    // huawei_schemas branch via `./gradlew :radar-schemas-commons:publishToMavenLocal`) before
+    // falling back to remote snapshot hosts.
+    mavenLocal()
+
+    // radar-schemas-commons huawei_schemas is only published as a snapshot; declare the
+    // candidate snapshot hosts here so the huawei-library build can resolve it regardless of
+    // which one the RADAR-Schemas release pipeline currently targets.
+    maven {
+        url = uri("https://central.sonatype.com/repository/maven-snapshots/")
+    }
+    maven {
+        url = uri("https://s01.oss.sonatype.org/content/repositories/snapshots/")
+    }
+    maven {
+        url = uri("https://maven.pkg.github.com/RADAR-base/RADAR-Schemas")
+        credentials {
+            username = project.findProperty("public.gpr.user") ?: System.getenv("GPR_USER")
+            password = project.findProperty("public.gpr.token") ?: System.getenv("GPR_TOKEN")
+        }
+    }
+}
+
+dependencies {
+    // Use the Kotlin JDK 8 standard library.
+    implementation libs.kotlin.stdlib
+
+    implementation libs.okhttp
+
+    implementation libs.radar.schemas.commons.huawei
+
+    implementation libs.jackson.annotations
+
+    implementation libs.jackson.databind
+
+    implementation libs.avro
+
+    implementation libs.jackson.datatype.jsr310
+
+    // Use the Kotlin test library.
+    testImplementation libs.kotlin.test
+
+    // Use the Kotlin JUnit 5 integration (matches the JUnit Platform the radar-kotlin
+    // Gradle plugin configures the `test` task with).
+    testImplementation libs.kotlin.test.junit5
+}
+
+project.afterEvaluate {
+    publishing {
+        publications {
+            library(MavenPublication) {
+                setGroupId "$group"
+                setArtifactId "huawei-library"
+                version "$version"
+                from components.java
+            }
+        }
+    }
+}

huawei-library/src/main/kotlin/org/radarbase/huawei/converter/FieldValues.kt

@@ -0,0 +1,82 @@
+/*
+ * Copyright 2026 Onsentia
+ *
+ * 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.huawei.converter
+
+import com.fasterxml.jackson.databind.JsonNode
+
+/**
+ * Typed accessor for a single Huawei Health Kit sample point's field values.
+ *
+ * The Health Kit Data API (`sampleSet:polymerize`) represents each field of a sample point using
+ * the same typed-value wrapper as the on-device HiHealth SDK's `Field`/`Value` model: a list of
+ * objects shaped like `{"fieldName": "steps_delta", "integerValue": 123}` (or `floatValue`,
+ * `longValue`, `stringValue` depending on the field's declared type). This class also tolerates a
+ * flattened `{"fieldName": value, ...}` object, in case a particular endpoint or API version
+ * returns the simplified shape, so a single parser can be reused across all sample-set based
+ * routes.
+ *
+ * Field name constants follow Huawei's public `Field` identifiers (e.g. `steps_delta`, `calories`,
+ * `avg`, `max`, `min`), as documented for the on-device and REST Health Kit APIs.
+ *
+ * @author yatharthranjan
+ */
+class FieldValues private constructor(private val values: Map<String, JsonNode>) {
+
+    fun getInt(field: String): Int? = values[field]?.let { if (it.isNull) null else it.asInt() }
+
+    fun getLong(field: String): Long? = values[field]?.let { if (it.isNull) null else it.asLong() }
+
+    fun getDouble(field: String): Double? = values[field]?.let {
+        if (it.isNull) null else it.asDouble()
+    }
+
+    fun getFloat(field: String): Float? = getDouble(field)?.toFloat()
+
+    fun getString(field: String): String? = values[field]?.let {
+        if (it.isNull) null else it.asText()
+    }
+
+    companion object {
+        private const val FIELD_NAME_KEY = "fieldName"
+        private val VALUE_KEYS =
+            listOf("integerValue", "floatValue", "longValue", "stringValue", "value")
+
+        fun from(node: JsonNode?): FieldValues {
+            if (node == null || node.isMissingNode || node.isNull) {
+                return FieldValues(emptyMap())
+            }
+            if (node.isArray) {
+                val map = LinkedHashMap<String, JsonNode>()
+                node.forEach { entry ->
+                    val name = entry.get(FIELD_NAME_KEY)?.asText() ?: return@forEach
+                    val value = VALUE_KEYS.firstNotNullOfOrNull { key -> entry.get(key) }
+                    if (value != null) {
+                        map[name] = value
+                    }
+                }
+                return FieldValues(map)
+            }
+            if (node.isObject) {
+                val map = LinkedHashMap<String, JsonNode>()
+                node.properties().forEach { (name, value) -> map[name] = value }
+                return FieldValues(map)
+            }
+            return FieldValues(emptyMap())
+        }
+    }
+}