<dependency>
    <groupId>com.here.schema.rib</groupId>
    <artifactId>road-attributes_v2_java</artifactId>
</dependency>
<dependency>
    <groupId>com.here.schema.rib</groupId>
    <artifactId>topology-geometry_v2_java</artifactId>
</dependency>

/*
 * Copyright (c) 2018-2023 HERE Europe B.V.
 *
 * 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.
 */

import akka.actor.ActorSystem;
import akka.actor.CoordinatedShutdown;
import akka.japi.Pair;
import com.here.hrn.HRN;
import com.here.platform.data.client.engine.javadsl.DataEngine;
import com.here.platform.data.client.engine.javadsl.ReadEngine;
import com.here.platform.data.client.engine.javadsl.WriteEngine;
import com.here.platform.data.client.javadsl.*;
import com.here.platform.data.client.model.VersionDependency;
import com.here.platform.pipeline.PipelineContext;
import com.here.schema.geometry.v2.GeometryOuterClass.LineString;
import com.here.schema.rib.v2.Anchor;
import com.here.schema.rib.v2.Common;
import com.here.schema.rib.v2.RoadAttributes;
import com.here.schema.rib.v2.RoadAttributesPartitionOuterClass.RoadAttributesPartition;
import com.here.schema.rib.v2.TopologyGeometry;
import com.here.schema.rib.v2.TopologyGeometryPartitionOuterClass.TopologyGeometryPartition;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import com.typesafe.config.ConfigValueFactory;
import java.util.*;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class RoadAttrWalkthruJava {

  public static void main(String[] args)
      throws InterruptedException, ExecutionException, TimeoutException {

    // Initialize the Akka Actor System used by the Data Client Library.
    Config config = ConfigFactory.empty();
    config =
        config.withValue(
            "here.platform.data-client.endpoint-locator.discovery-service-env",
            ConfigValueFactory.fromAnyRef("custom"));
    config =
        config.withValue(
            "here.platform.data-client.endpoint-locator.discovery-service-url",
            // Please use https://api-lookup.data.api.platform.hereolp.cn URL for China Environment
            // We should define a custom URL, specific to China Environment, for a discovery service
            // endpoint that allows discovering various Data Service APIs like publish, metadata,
            // query, etc.
            ConfigValueFactory.fromAnyRef("https://api-lookup.data.api.platform.here.com"));

    ActorSystem actorSystem = ActorSystem.create("RoadAttrWalkthruJavaExampleApp", config);

    try {
      PipelineContext pipelineContext = new PipelineContext();

      HRN hereMapContentHrn = pipelineContext.config().getInputCatalogs().get("hereMapContent");
      long hereMapContentVersion =
          pipelineContext.getJob().get().getInputCatalogs().get("hereMapContent").version();

      // Choosing a Chatham Islands partition in New Zealand to keep things simple.
      // Please use partition 23551658 in Tibet for China Environment.
      // Feel free to replace this with one or more other partitions of interest.
      List<String> roadAttrPartitionNames = Collections.singletonList("17302687");

      HRN outputHrn = pipelineContext.config().getOutputCatalog();

      // Initialize the I/O for the catalog.
      QueryApi queryApi = DataClient.get(actorSystem).queryApi(hereMapContentHrn);
      ReadEngine readEngine = DataEngine.get(actorSystem).readEngine(hereMapContentHrn);
      WriteEngine writeEngine = DataEngine.get(actorSystem).writeEngine(outputHrn);
      PublishApi publishApi = DataClient.get(actorSystem).publishApi(outputHrn);

      // Retrieve the metadata for the road-attribute partition(s).
      List<Partition> roadAttrMetadata =
          queryApi
              .getPartitionsById(
                  hereMapContentVersion,
                  "road-attributes",
                  roadAttrPartitionNames,
                  Collections.emptySet())
              .toCompletableFuture()
              .get(60, TimeUnit.SECONDS);

      roadAttrMetadata.forEach(
          metadata -> {
            // For users using platform.here.com:
            // https://here-tech.skawa.fun/documentation/data-client-library/dev_guide/client/get-data.html
            // https://here-tech.skawa.fun/documentation/here-map-content/topics/road-layer.html
            // For users using platform.hereolp.cn:
            // https://here-tech.skawa.fun/cn/documentation/data-client-library/dev_guide/client/get-data.html
            // https://repo.platform.hereolp.cn/artifactory/open-location-platform-docs/Data_Specifications/HERE_Map_Content/HERE_Map_Content_OLP_China_v2019-05_Data_Specification.pdf

            // Retrieve the payload for the road-attribute partition and decode its contents.
            RoadAttributesPartition roadAttributesPartition =
                readEngine
                    .get(metadata, RoadAttributesPartition::parseFrom)
                    .toCompletableFuture()
                    .join();

            // For users using platform.here.com:
            // https://here-tech.skawa.fun/documentation/here-map-content/topics_api/com.here.schema.rib.v2.physicalattribute.html
            // For users using platform.hereolp.cn:
            // https://repo.platform.hereolp.cn/artifactory/open-location-platform-docs/Data_Specifications/HERE_Map_Content/HERE_Map_Content_OLP_China_v2019-05_Data_Specification.pdf

            // Separate the physical attribute values combinations into those which have paved=true,
            // and those which have paved=false.
            Map<Boolean, List<RoadAttributes.PhysicalAttribute>> attrsByPaved =
                roadAttributesPartition
                    .getPhysicalList()
                    .stream()
                    .collect(Collectors.partitioningBy(RoadAttributes.PhysicalAttribute::getPaved));

            List<RoadAttributes.PhysicalAttribute> paved = attrsByPaved.get(true);
            List<RoadAttributes.PhysicalAttribute> unpaved = attrsByPaved.get(false);

            // For users using platform.here.com:
            // https://here-tech.skawa.fun/documentation/here-map-content/topics/anchor.html
            // For users using platform.hereolp.cn:
            // https://repo.platform.hereolp.cn/artifactory/open-location-platform-docs/Data_Specifications/HERE_Map_Content/HERE_Map_Content_OLP_China_v2019-05_Data_Specification.pdf

            // Get the segment anchor indices for the paved and unpaved segments.
            List<Integer> pavedSegmentAnchorIndices =
                paved
                    .stream()
                    .flatMap(attr -> attr.getSegmentAnchorIndexList().stream())
                    .collect(Collectors.toList());
            List<Integer> unpavedSegmentAnchorIndices =
                unpaved
                    .stream()
                    .flatMap(attr -> attr.getSegmentAnchorIndexList().stream())
                    .collect(Collectors.toList());

            // Follow the segment anchor references to get the name of the partition(s)
            // which contain the actual segment geometry within the topology-geometry layer.
            Set<String> pavedTopologyPartitions =
                getTopologyPartitionNames(roadAttributesPartition, pavedSegmentAnchorIndices);
            Set<String> unpavedTopologyPartitions =
                getTopologyPartitionNames(roadAttributesPartition, unpavedSegmentAnchorIndices);

            // Retrieve the metadata for the referenced topology-geometry partitions.
            Set<String> allTopoPartitions =
                Stream.concat(pavedTopologyPartitions.stream(), unpavedTopologyPartitions.stream())
                    .collect(Collectors.toSet());
            List<Partition> topoMetadata =
                queryApi
                    .getPartitionsById(
                        hereMapContentVersion,
                        "topology-geometry",
                        new ArrayList<>(allTopoPartitions),
                        Collections.emptySet())
                    .toCompletableFuture()
                    .join();

            // Retrieve the payloads and decode the content of the referenced topology-geometry
            // partitions.
            Map<String, TopologyGeometryPartition> topoPartitionsByName =
                topoMetadata
                    .stream()
                    .map(
                        topoMeta ->
                            new Pair<>(
                                topoMeta.getPartition(),
                                readEngine
                                    .get(topoMeta, TopologyGeometryPartition::parseFrom)
                                    .toCompletableFuture()
                                    .join()))
                    .collect(Collectors.toMap(Pair::first, Pair::second));

            // Follow the segment anchor references to the decoded topology-geometry content to get
            // the geometry
            // for each paved and unpaved segment.
            Map<String, List<LineString>> pavedGeo =
                getGeo(topoPartitionsByName, roadAttributesPartition, pavedSegmentAnchorIndices);
            Map<String, List<LineString>> unpavedGeo =
                getGeo(topoPartitionsByName, roadAttributesPartition, unpavedSegmentAnchorIndices);

            // Write out the decoded geometry as GeoJSON.
            publishGeoJson(
                pavedGeo,
                unpavedGeo,
                writeEngine,
                publishApi,
                queryApi,
                hereMapContentHrn,
                hereMapContentVersion);
          });

    } finally {
      shutdownActorSystem(actorSystem);
    }
  }

  private static void shutdownActorSystem(ActorSystem actorSystem) {
    CoordinatedShutdown.get(actorSystem)
        .runAll(CoordinatedShutdown.unknownReason())
        .toCompletableFuture()
        .join();
  }

  // Gets the segment references for a given segment anchor index in a road attributes partition.
  private static Stream<Common.Reference> getSegmentRefs(
      RoadAttributesPartition roadAttributesPartition, Integer segmentAnchorIdx) {
    return roadAttributesPartition
        .getSegmentAnchor(segmentAnchorIdx)
        .getOrientedSegmentRefList()
        .stream()
        .map(Anchor.SegmentAnchor.OrientedSegmentReference::getSegmentRef);
  }

  // Follow the segment anchor references to get the name of the partition(s)
  // which contain the actual segment geometry within the topology-geometry layer.
  // We need to do this because it is not guaranteed that every segment anchor refers to a segment
  // that is in the topology-geometry partition that has the same name as the road-attributes
  // partition which contains the segment anchor.
  private static Set<String> getTopologyPartitionNames(
      RoadAttributesPartition roadAttributesPartition, List<Integer> segmentAnchorIndices) {
    return segmentAnchorIndices
        .stream()
        .flatMap(idx -> getSegmentRefs(roadAttributesPartition, idx))
        .map(Common.Reference::getPartitionName)
        .collect(Collectors.toSet());
  }

  // Follow the segment anchor references to the decoded topology-geometry content to get the
  // geometry
  // for each paved and unpaved segment.
  private static Map<String, List<LineString>> getGeo(
      Map<String, TopologyGeometryPartition> topoPartitionsByName,
      RoadAttributesPartition roadAttributesPartition,
      List<Integer> segmentAnchorIndices) {

    return segmentAnchorIndices
        .stream()

        // Follow the segment anchor reference to get the corresponding topology-geometry partition.
        .flatMap(idx -> getSegmentRefs(roadAttributesPartition, idx))
        .flatMap(
            ref ->
                topoPartitionsByName
                    .get(ref.getPartitionName())
                    .getSegmentList()
                    .stream()

                    // For each segment anchor reference, get the segment from
                    // topology-geometry which matches its identifier.
                    .filter(seg -> seg.getIdentifier().equals(ref.getIdentifier()))
                    .map(TopologyGeometry.Segment::getGeometry)
                    .map(lineString -> new Pair<>(ref.getPartitionName(), lineString)))

        // Group the LineStrings under their respective topology partitions.
        .collect(Collectors.groupingBy(Pair::first))
        .entrySet()
        .stream()
        .map(
            pair ->
                new Pair<>(
                    pair.getKey(),
                    pair.getValue().stream().map(Pair::second).collect(Collectors.toList())))
        .collect(Collectors.toMap(Pair::first, Pair::second));
  }

  // Formats paved and unpaved LineStrings as GeoJSON.
  private static String makeGeoJsonFeature(boolean paved, LineString lineString) {
    StringJoiner coordJoiner = new StringJoiner(",", "[", "]");
    List<String> pointsAsString =
        lineString
            .getPointList()
            .stream()
            .map(point -> "[" + point.getLongitude() + ", " + point.getLatitude() + "]")
            .collect(Collectors.toList());
    pointsAsString.forEach(coordJoiner::add);

    // Give paved and unpaved segments tooltips indicating paved or unpaved, and color
    // them green and red, respectively.
    String tooltip = paved ? "paved" : "unpaved";
    String color = paved ? "\"#02F93E\"" : "\"#FF3333\"";

    return "{ \"type\": \"Feature\", \"geometry\": "
        + "{ \"type\": \"LineString\", \"coordinates\": "
        + coordJoiner.toString()
        + "}"
        + ", \"properties\": { \"tooltip\": \""
        + tooltip
        + "\""
        + ", \"width\" : 3.0"
        + ", \"style\": { \"color\": "
        + color
        + "}"
        + "} }";
  }

  // Write out the decoded geometry as GeoJSON.
  private static void publishGeoJson(
      Map<String, List<LineString>> pavedGeo,
      Map<String, List<LineString>> unpavedGeo,
      WriteEngine writeEngine,
      PublishApi publishApi,
      QueryApi queryApi,
      HRN inputHrn,
      Long inputVersion) {
    Set<String> allGeoKeys =
        Stream.concat(pavedGeo.keySet().stream(), unpavedGeo.keySet().stream())
            .collect(Collectors.toSet());
    String layerId = "roadsegments";
    List<CommitPartition> commitPartitions =
        allGeoKeys
            .stream()
            .map(
                key -> {
                  List<String> pavedGeoJson =
                      pavedGeo
                          .getOrDefault(key, Collections.emptyList())
                          .stream()
                          .map(lineString -> makeGeoJsonFeature(true, lineString))
                          .collect(Collectors.toList());
                  List<String> unpavedGeoJson =
                      unpavedGeo
                          .getOrDefault(key, Collections.emptyList())
                          .stream()
                          .map(lineString -> makeGeoJsonFeature(false, lineString))
                          .collect(Collectors.toList());

                  List<String> allGeoJson =
                      Stream.concat(pavedGeoJson.stream(), unpavedGeoJson.stream())
                          .collect(Collectors.toList());

                  String geoJsonString =
                      allGeoJson
                          .stream()
                          .collect(
                              Collectors.joining(
                                  ",", "{ \"type\": \"FeatureCollection\", \"features\": [", "]}"));

                  NewPartition newPartition =
                      new NewPartition.Builder()
                          .withPartition(key)
                          .withLayer(layerId)
                          .withData(geoJsonString.getBytes())
                          .build();

                  return writeEngine.put(newPartition).toCompletableFuture().join();
                })
            .collect(Collectors.toList());

    // Write the HERE Map Content catalog's dependencies as indirect dependencies of
    // our output catalog, and the HERE Map Content catalog itself as a direct
    // dependency. This is good practice so that consumers of this catalog can
    // perform dependency analysis if needed, for example, when scheduling downstream
    // pipelines, and performing incremental processing.
    List<VersionDependency> dependencies =
        queryApi
            .getVersion(inputVersion)
            .toCompletableFuture()
            .join()
            .getDependencies()
            .stream()
            .map(dep -> new VersionDependency(dep.hrn(), dep.version(), false))
            .collect(Collectors.toList());
    dependencies.add(new VersionDependency(inputHrn, inputVersion, true));

    OptionalLong baseVersion = publishApi.getBaseVersion().toCompletableFuture().join();
    publishApi
        .publishBatch2(
            baseVersion,
            Optional.of(Collections.singletonList(layerId)),
            dependencies,
            commitPartitions.iterator())
        .toCompletableFuture()
        .join();
  }
}

mvn compile exec:java -Dexec.mainClass=RoadAttrWalkthruJava \
  -Dpipeline-config.file=pipeline-config.conf \
  -Dpipeline-job.file=pipeline-job.conf \
  -Dhere.platform.data-client.request-signer.credentials.here-account.here-token-scope={{YOUR_PROJECT_HRN}}