feat: Add ReverseGeocoder, PointInPolygon, and example notebooks (#1339)

Co-authored-by: Sudhindra Kovalam <[email protected]>

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/geospatial/AzMapsSearchSchemas.scala

@@ -87,7 +87,6 @@ case class ClassificationName (
   // Name property
   name: Option[String])
 
-
 case class DataSources (
   geometry: Option[Geometry])
 
@@ -138,6 +137,8 @@ case class LongRunningOperationResult (
   operationId: Option[String],
   // The status state of the request.
   status: Option[String],
+  // resource location
+  resourceLocation: Option[String],
   // The created timestamp.
   created: Option[Date],
   error: Option[ErrorDetail],
@@ -166,7 +167,6 @@ case class OperatingHoursTimeRange (
  // if it was closed before the range.
  endTime: Option[OperatingHoursTime])
 
-
 case class PointOfInterest (
  // Name of the POI property
  name: Option[String],
@@ -186,7 +186,6 @@ case class PointOfInterestCategorySet (
   // Category ID
   id: Option[Integer])
 
-
 object ReverseSearchAddressResult extends SparkBindings[ReverseSearchAddressResult]
 case class ReverseSearchAddressResult (
   // The error object.

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/geospatial/AzMapsSpatialSchemas.scala

@@ -0,0 +1,36 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.geospatial
+
+import com.microsoft.azure.synapse.ml.core.schema.SparkBindings
+
+case class PointInPolygonSummary (
+ sourcePoint: Option[LatLongPairAbbreviated],
+ // A unique data id (udid) for the uploaded content.
+ // Udid is not applicable for POST spatial operations(set to null)
+ udid: Option[String],
+ // Processing information
+ information: Option[String]
+)
+
+case class PointInPolygonResult(
+  pointInPolygons: Option[Boolean],
+  intersectingGeometries: Option[Seq[String]]
+)
+
+object PointInPolygonProcessResult extends SparkBindings[PointInPolygonProcessResult]
+case class PointInPolygonProcessResult (
+  summary: Option[PointInPolygonSummary],
+  // Point In Polygon Result Object
+  result: Option[PointInPolygonResult]
+)
+
+case class MapDataMetadataResult(
+  udid: Option[String],
+  location: Option[String],
+  created: Option[String],
+  updated: Option[String],
+  sizeInBytes: Option[String],
+  uploadStatus: Option[String]
+)

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/geospatial/AzureMapsHelpers.scala → cognitive/src/main/scala/com/microsoft/azure/synapse/ml/geospatial/AzureMapsTraits.scala

@@ -4,24 +4,75 @@
 package com.microsoft.azure.synapse.ml.geospatial
 
 import com.microsoft.azure.synapse.ml.build.BuildInfo
-import com.microsoft.azure.synapse.ml.cognitive.{HasAsyncReply, HasServiceParams, HasSubscriptionKey, URLEncodingUtils}
-import com.microsoft.azure.synapse.ml.io.http.{CustomInputParser, HTTPInputParser, HasURL}
-import com.microsoft.azure.synapse.ml.io.http._
+import com.microsoft.azure.synapse.ml.codegen.Wrappable
+import com.microsoft.azure.synapse.ml.cognitive.{HasAsyncReply, HasServiceParams, HasUrlPath}
 import com.microsoft.azure.synapse.ml.io.http.HandlingUtils._
-import org.apache.http.client.methods.{HttpGet, HttpPost, HttpRequestBase}
-import org.apache.http.entity.{AbstractHttpEntity, StringEntity}
+import com.microsoft.azure.synapse.ml.io.http.{HasURL, _}
+import org.apache.http.client.methods.HttpGet
 import org.apache.http.impl.client.CloseableHttpClient
 import org.apache.spark.ml.param._
-import org.apache.spark.sql.Row
-import org.apache.spark.sql.types.StructType
+import spray.json.DefaultJsonProtocol.{DoubleJsonFormat, StringJsonFormat, seqFormat}
 
-import java.net.{URI, URLEncoder}
+import java.net.URI
 import java.util.concurrent.TimeoutException
 import scala.concurrent.blocking
 import scala.language.{existentials, postfixOps}
-import spray.json.DefaultJsonProtocol.{StringJsonFormat, seqFormat}
 
-trait HasAddressInput extends HasServiceParams with HasSubscriptionKey with HasURL {
+trait HasSetGeography extends Wrappable with HasURL with HasUrlPath {
+  override def pyAdditionalMethods: String = super.pyAdditionalMethods + {
+    """
+      |def setGeography(self, value):
+      |    self._java_obj = self._java_obj.setGeography(value)
+      |    return self
+      |""".stripMargin
+  }
+
+  def setGeography(v: String): this.type = {
+    setUrl(s"https://$v.atlas.microsoft.com/" + urlPath)
+  }
+}
+
+trait HasUserDataIdInput extends HasServiceParams {
+  val userDataIdentifier = new ServiceParam[String](
+    this, "userDataIdentifier", "the identifier for the user uploaded data")
+
+  def getUserDataIdentifier: String = getScalarParam(userDataIdentifier)
+
+  def setUserDataIdentifier(v: String): this.type = setScalarParam(userDataIdentifier, v)
+
+  def getUserDataIdentifierCol: String = getVectorParam(userDataIdentifier)
+
+  def setUserDataIdentifierCol(v: String): this.type = setVectorParam(userDataIdentifier, v)
+}
+
+trait HasLatLonPairInput extends HasServiceParams {
+  val latitude = new ServiceParam[Seq[Double]](
+    this, "latitude", "the latitude of location")
+  val longitude = new ServiceParam[Seq[Double]](
+    this, "longitude", "the longitude of location")
+
+  def getLatitude: Seq[Double] = getScalarParam(latitude)
+
+  def setLatitude(v: Seq[Double]): this.type = setScalarParam(latitude, v)
+
+  def setLatitude(v: Double): this.type = setScalarParam(latitude, Seq(v))
+
+  def getLatitudeCol: String = getVectorParam(latitude)
+
+  def setLatitudeCol(v: String): this.type = setVectorParam(latitude, v)
+
+  def getLongitude: Seq[Double] = getScalarParam(longitude)
+
+  def setLongitude(v: Seq[Double]): this.type = setScalarParam(longitude, v)
+
+  def setLongitude(v: Double): this.type = setScalarParam(longitude, Seq(v))
+
+  def getLongitudeCol: String = getVectorParam(longitude)
+
+  def setLongitudeCol(v: String): this.type = setVectorParam(longitude, v)
+}
+
+trait HasAddressInput extends HasServiceParams {
   val address = new ServiceParam[Seq[String]](
     this, "address", "the address to geocode")
 
@@ -34,34 +85,6 @@ trait HasAddressInput extends HasServiceParams with HasSubscriptionKey with HasU
   def getAddressCol: String = getVectorParam(address)
 
   def setAddressCol(v: String): this.type = setVectorParam(address, v)
-
-  protected def prepareEntity: Row => Option[AbstractHttpEntity]
-
-  protected def contentType: Row => String = { _ => "application/json" }
-
-  protected def inputFunc(schema: StructType): Row => Option[HttpRequestBase] = {
-    { row: Row =>
-      if (shouldSkip(row)) {
-        None
-      } else {
-
-        val queryParams = "?" + URLEncodingUtils.format(Map("api-version" -> "1.0",
-          "subscription-key" -> getSubscriptionKey))
-        val post = new HttpPost(new URI(getUrl + queryParams))
-        post.setHeader("Content-Type", "application/json")
-        post.setHeader("User-Agent", s"synapseml/${BuildInfo.version}${HeaderValues.PlatformInfo}")
-        val encodedAddresses = getValue(row, address).map(x => URLEncoder.encode(x, "UTF-8")).toList
-        val payloadItems = encodedAddresses.map(x => s"""{ "query": "?query=$x&limit=1" }""").mkString(",")
-        val payload = s"""{ "batchItems": [ $payloadItems ] }"""
-        post.setEntity(new StringEntity(payload))
-        Some(post)
-      }
-    }
-  }
-
-  protected def getInternalInputParser(schema: StructType): HTTPInputParser = {
-    new CustomInputParser().setNullableUDF(inputFunc(schema))
-  }
 }
 
 trait MapsAsyncReply extends HasAsyncReply {
@@ -109,4 +132,3 @@ trait MapsAsyncReply extends HasAsyncReply {
     }
   }
 }
-

cognitive/src/main/scala/com/microsoft/azure/synapse/ml/geospatial/CheckPointInPolygon.scala

@@ -0,0 +1,59 @@
+// Copyright (C) Microsoft Corporation. All rights reserved.
+// Licensed under the MIT License. See LICENSE in project root for information.
+
+package com.microsoft.azure.synapse.ml.geospatial
+
+import com.microsoft.azure.synapse.ml.build.BuildInfo
+import com.microsoft.azure.synapse.ml.cognitive._
+import com.microsoft.azure.synapse.ml.io.http.{CustomInputParser, HTTPInputParser, HeaderValues}
+import com.microsoft.azure.synapse.ml.logging.BasicLogging
+import org.apache.http.client.methods.{HttpGet, HttpRequestBase}
+import org.apache.spark.ml.ComplexParamsReadable
+import org.apache.spark.ml.util.Identifiable
+import org.apache.spark.sql.Row
+import org.apache.spark.sql.types.{DataType, StructType}
+
+import java.net.URI
+
+object CheckPointInPolygon extends ComplexParamsReadable[CheckPointInPolygon]
+
+class CheckPointInPolygon(override val uid: String)
+  extends CognitiveServicesBase(uid)
+    with HasInternalJsonOutputParser with BasicLogging with HasServiceParams
+    with HasSubscriptionKey with HasSetGeography with HasLatLonPairInput with HasUserDataIdInput {
+
+  protected def inputFunc: Row => Option[HttpRequestBase] = {
+    { row: Row =>
+      if (shouldSkip(row)) {
+        None
+      } else {
+        val udid = getValue(row, userDataIdentifier).mkString
+        val lat = String.valueOf(getValue(row, latitude))
+        val lon = String.valueOf(getValue(row, longitude))
+
+        val queryParams = "?" + URLEncodingUtils.format(Map("api-version" -> "1.0",
+          "subscription-key" -> getSubscriptionKey,
+          "udid" -> udid,
+          "lat" -> lat,
+          "lon" -> lon))
+        val get = new HttpGet()
+        get.setURI(new URI(getUrl + queryParams))
+        get.setHeader("User-Agent", s"synapseml/${BuildInfo.version}${HeaderValues.PlatformInfo}")
+        Some(get)
+      }
+    }
+  }
+
+  protected def getInternalInputParser(schema: StructType): HTTPInputParser = {
+    new CustomInputParser().setNullableUDF(inputFunc)
+  }
+
+  def this() = this(Identifiable.randomUID("CheckPointInPolygon"))
+
+  setDefault(
+    url -> "https://atlas.microsoft.com/")
+
+  override protected def responseDataType: DataType = PointInPolygonProcessResult.schema
+
+  override def urlPath: String = "spatial/pointInPolygon/json"
+}