feat: Call noise correction python binary from kotlin.

experimental/dp_consistency/src/main/kotlin/reporting/BUILD.bazel

@@ -0,0 +1,29 @@
+load("@wfa_rules_kotlin_jvm//kotlin:defs.bzl", "kt_jvm_binary", "kt_jvm_library")
+
+package(default_visibility = ["//visibility:public"])
+
+kt_jvm_binary(
+    name = "report_post_processing",
+    srcs = ["ReportPostProcessing.kt"],
+    main_class = "experimental.dp_consistency.src.main.kotlin.reporting.ReportPostProcessing",
+    resources = [
+        "//experimental/dp_consistency/src/main/python/tools:post_process_origin_report_pyzip",
+    ],
+    deps = [
+        ":report_converter",
+        "//experimental/dp_consistency/src/main/proto/reporting:report_summary_kt_jvm_proto",
+        "//src/main/proto/wfa/measurement/reporting/v2alpha:report_kt_jvm_proto",
+        "@wfa_common_jvm//src/main/kotlin/org/wfanet/measurement/common",
+    ],
+)
+
+kt_jvm_library(
+    name = "report_converter",
+    srcs = ["ReportConverter.kt"],
+    deps = [
+        "//experimental/dp_consistency/src/main/proto/reporting:report_summary_kt_jvm_proto",
+        "//src/main/proto/wfa/measurement/reporting/v2alpha:report_kt_jvm_proto",
+        "@maven//:com_google_protobuf_protobuf_java_util",
+        "@wfa_common_jvm//imports/java/com/google/gson",
+    ],
+)

experimental/dp_consistency/src/main/kotlin/reporting/ReportConverter.kt

@@ -0,0 +1,159 @@
+// Copyright 2024 The Cross-Media Measurement Authors
+//
+// 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 experimental.dp_consistency.src.main.kotlin.reporting
+
+import com.google.protobuf.InvalidProtocolBufferException
+import com.google.protobuf.util.JsonFormat
+import experimental.dp_consistency.src.main.proto.reporting.MeasurementDetailKt
+import experimental.dp_consistency.src.main.proto.reporting.ReportSummary
+import experimental.dp_consistency.src.main.proto.reporting.measurementDetail
+import experimental.dp_consistency.src.main.proto.reporting.reportSummary
+import org.wfanet.measurement.reporting.v2alpha.Metric
+import org.wfanet.measurement.reporting.v2alpha.Report
+import org.wfanet.measurement.reporting.v2alpha.report
+
+data class ReportingSetSummary(val measurementPolicy: String, val dataProviders: List<String>)
+
+data class SetOperationSummary(val isCumulative: Boolean, val setOperation: String)
+
+object ReportConversion {
+  fun getReportFromJsonString(reportAsJsonString: String): Report {
+    val report =
+      try {
+        val protoBuilder = Report.newBuilder()
+        JsonFormat.parser().ignoringUnknownFields().merge(reportAsJsonString, protoBuilder)
+        protoBuilder.build()
+      } catch (e: InvalidProtocolBufferException) {
+        Report.getDefaultInstance()
+      }
+    return report
+  }
+
+  fun convertJsontoReportSummaries(reportAsJsonString: String): List<ReportSummary> {
+    val report =
+      try {
+        val protoBuilder = Report.newBuilder()
+        JsonFormat.parser().ignoringUnknownFields().merge(reportAsJsonString, protoBuilder)
+        protoBuilder.build()
+      } catch (e: InvalidProtocolBufferException) {
+        Report.getDefaultInstance()
+      }
+
+    return report.toReportSummaries()
+  }
+}
+
+fun getMeasurementPolicy(tag: String): String {
+  when {
+    "measurement_policy=AMI" in tag -> return "ami"
+    "measurement_policy=MRC" in tag -> return "mrc"
+    "measurement_policy=CUSTOM" in tag -> return "custom"
+    else -> error("Measurement policy not specified in the tag.")
+  }
+}
+
+fun getSetOperation(tag: String): String {
+  val parts = tag.split(", ")
+  val setOperationPart = parts.find { it.startsWith("set_operation=") }
+  return setOperationPart?.let { it.substringAfter("set_operation=") }
+    ?: error("Set operation must be specified.")
+}
+
+fun isCumulative(tag: String): Boolean {
+  return tag.contains("cumulative=true")
+}
+
+fun getTargets(tag: String): List<String> {
+  val parts = tag.split(", ")
+  val targetPart = parts.find { it.startsWith("target=") }
+  return targetPart?.let { it.substringAfter("target=").split(",") }
+    ?: error("There must be at least one target.")
+}
+
+fun Report.toReportSummaries(): List<ReportSummary> {
+  require(this.state == Report.State.SUCCEEDED) { "Unsucceeded report is not supported." }
+
+  val measurementPoliciesByReportingSet =
+    this.reportingMetricEntriesList.associate { entry ->
+      val reportingSet = entry.key
+      val tag = this.tags.getValue(reportingSet)
+      reportingSet to ReportingSetSummary(getMeasurementPolicy(tag), getTargets(tag))
+    }
+
+  val metricCalculationSpecs =
+    this.reportingMetricEntriesList.flatMapTo(mutableSetOf()) {
+      it.value.metricCalculationSpecsList
+    }
+
+  val setOperationByMetricCalculationSpec =
+    metricCalculationSpecs.associate { spec ->
+      val tag = this.tags.getValue(spec)
+      spec to SetOperationSummary(isCumulative(tag), getSetOperation(tag))
+    }
+
+  val filterGroupByMetricCalculationSpec =
+    metricCalculationSpecs.associate { spec ->
+      val tag = this.tags.getValue(spec)
+      spec to tag.split(", ").find { it.startsWith("common_filter=") }
+    }
+
+  val filterGroups = filterGroupByMetricCalculationSpec.values.toSet()
+
+  // Groups results by (reporting set x metric calculation spec).
+  val measurementSets =
+    this.metricCalculationResultsList.groupBy { Pair(it.metricCalculationSpec, it.reportingSet) }
+
+  val reportSummaries = mutableListOf<ReportSummary>()
+  for (filter in filterGroups) {
+    val reportSummary = reportSummary {
+      measurementSets.forEach { (key, value) ->
+        if (filterGroupByMetricCalculationSpec.getValue(key.first) == filter) {
+          measurementDetails += measurementDetail {
+            measurementPolicy = measurementPoliciesByReportingSet[key.second]!!.measurementPolicy
+            dataProviders += measurementPoliciesByReportingSet[key.second]!!.dataProviders
+            isCumulative = setOperationByMetricCalculationSpec[key.first]!!.isCumulative
+            setOperation = setOperationByMetricCalculationSpec[key.first]!!.setOperation
+            var measurementList =
+              value
+                .flatMap { it.resultAttributesList }
+                .sortedBy { it.timeInterval.endTime.seconds }
+                .filter { it.metricResult.hasReach() || it.metricResult.hasReachAndFrequency() }
+                .map { resultAttribute ->
+                  require(resultAttribute.state == Metric.State.SUCCEEDED) {
+                    "Unsucceeded measurement result is not supported."
+                  }
+                  MeasurementDetailKt.measurementResult {
+                    if (resultAttribute.metricResult.hasReach()) {
+                      reach = resultAttribute.metricResult.reach.value
+                      standardDeviation =
+                        resultAttribute.metricResult.reach.univariateStatistics.standardDeviation
+                    } else {
+                      reach = resultAttribute.metricResult.reachAndFrequency.reach.value
+                      standardDeviation =
+                        resultAttribute.metricResult.reachAndFrequency.reach.univariateStatistics
+                          .standardDeviation
+                    }
+                    metric = resultAttribute.metric
+                  }
+                }
+            measurementResults.addAll(measurementList)
+          }
+        }
+      }
+    }
+    reportSummaries.add(reportSummary)
+  }
+  return reportSummaries
+}