Merge pull request #3736 from LBNL-UCB-STI/do/#3232-activitysim-skims…

…-omx-format

Activitysim skims omx format

build.gradle

@@ -212,6 +212,7 @@ dependencies {
 
     implementation("com.uber:h3:3.7.2")
     implementation("com.github.LBNL-UCB-STI:jsprit-wrapper:v0.5.1")
+    implementation("com.github.LBNL-UCB-STI:omx-java:v2.0.1")
 
     testImplementation group: 'junit', name: 'junit', version: '4.8'
     testImplementation group: 'org.mockito', name: 'mockito-inline', version: '2.27.0'

src/main/resources/beam-template.conf

@@ -715,6 +715,8 @@ beam.router.skim = {
   activity-sim-skimmer {
     name = "activity-sim-skimmer"
     fileBaseName = "String | activitySimODSkims"
+    #values: csv | omx
+    fileOutputFormat = "csv"
   }
   drive-time-skimmer {
     name = "drive-time-skimmer"
@@ -1167,7 +1169,7 @@ beam.urbansim.backgroundODSkimsCreator {
   routerType = "string | r5"
   # possible values: taz, h3
   skimsGeoType = "string | h3"
-  # possible values: od, activitySim
+  # possible values: od, activitySim, activitySimOmx
   skimsKind = "string | od"
   numberOfH3Indexes = "int | 1000"
 }

src/main/scala/beam/router/skim/ActivitySimMetric.scala

@@ -0,0 +1,34 @@
+package beam.router.skim
+
+import enumeratum.{Enum, EnumEntry}
+
+import scala.collection.immutable
+
+/**
+  * @author Dmitry Openkov
+  */
+sealed abstract class ActivitySimMetric extends EnumEntry
+
+object ActivitySimMetric extends Enum[ActivitySimMetric] {
+  val values: immutable.IndexedSeq[ActivitySimMetric] = findValues
+
+  case object TOTIVT extends ActivitySimMetric
+  case object FERRYIVT extends ActivitySimMetric
+  case object FAR extends ActivitySimMetric
+  case object XWAIT extends ActivitySimMetric
+  case object KEYIVT extends ActivitySimMetric
+  case object DTIM extends ActivitySimMetric
+  case object IWAIT extends ActivitySimMetric
+  case object BOARDS extends ActivitySimMetric
+  case object DDIST extends ActivitySimMetric
+  case object WAUX extends ActivitySimMetric
+  case object BTOLL extends ActivitySimMetric
+  case object VTOLL extends ActivitySimMetric
+  case object TIME extends ActivitySimMetric
+  case object DIST extends ActivitySimMetric
+  case object WEGR extends ActivitySimMetric
+  case object WACC extends ActivitySimMetric
+  case object IVT extends ActivitySimMetric
+  case object TRIPS extends ActivitySimMetric
+  case object FAILURES extends ActivitySimMetric
+}

src/main/scala/beam/router/skim/ActivitySimPathType.scala

@@ -2,6 +2,7 @@ package beam.router.skim
 
 import beam.router.Modes.BeamMode
 import beam.router.model.{EmbodiedBeamLeg, EmbodiedBeamTrip}
+import beam.router.skim.ActivitySimMetric._
 import org.matsim.api.core.v01.population.Activity
 
 sealed trait ActivitySimPathType
@@ -68,7 +69,7 @@ object ActivitySimPathType {
 
     // so far not used:
     //    WLK_EXP_WLK, = express bus
-    //    WLK_TRN_WLK  = train ??
+    //    WLK_TRN_WLK  = walk transit (general)
 
     val (longestWalkTransitLeg, _) = tryGetLongestLegId(trip, isWalkTransit)
     longestWalkTransitLeg.map(leg => leg.beamLeg.mode) match {
@@ -214,6 +215,14 @@ object ActivitySimPathType {
     }
   }
 
+  val walkTransitPathTypes: Seq[ActivitySimPathType] = Seq(
+    WLK_COM_WLK,
+    WLK_HVY_WLK,
+    WLK_EXP_WLK,
+    WLK_LOC_WLK,
+    WLK_LRF_WLK
+  )
+
   val allPathTypes: Seq[ActivitySimPathType] = Seq(
     DRV_COM_WLK,
     DRV_HVY_WLK,
@@ -238,11 +247,13 @@ object ActivitySimPathType {
     WLK_LOC_WLK,
     WLK_LRF_DRV,
     WLK_LRF_WLK,
-    // ignored because we did not understand what kind of vehicles are TRN yet
-    //    WLK_TRN_WLK
+    // UPDATE: TRN is a catch-all for all walk-transit trips
+    WLK_TRN_WLK,
     WALK
   )
 
+  def isWalkTransit(pathType: ActivitySimPathType): Boolean = walkTransitPathTypes.contains(pathType)
+
   val allPathTypesMap: Map[String, ActivitySimPathType] =
     allPathTypes.map(x => x.toString -> x).toMap
 

src/main/scala/beam/router/skim/ActivitySimSkimmer.scala

@@ -1,6 +1,8 @@
 package beam.router.skim
 
+import beam.router.skim.ActivitySimPathType.{isWalkTransit, WLK_TRN_WLK}
 import beam.router.skim.core.{AbstractSkimmer, AbstractSkimmerInternal, AbstractSkimmerKey, AbstractSkimmerReadOnly}
+import beam.router.skim.urbansim.ActivitySimOmxWriter
 import beam.sim.BeamScenario
 import beam.sim.config.BeamConfig
 import beam.utils.ProfilingUtils
@@ -11,6 +13,7 @@ import org.matsim.core.controler.MatsimServices
 import org.matsim.core.controler.events.IterationEndsEvent
 
 import java.io.BufferedWriter
+import scala.collection.immutable.SortedSet
 import scala.util.Failure
 import scala.util.control.NonFatal
 
@@ -29,8 +32,9 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
 
   override def writeToDisk(event: IterationEndsEvent): Unit =
     if (config.writeSkimsInterval > 0 && event.getIteration % config.writeSkimsInterval == 0) {
+      val extension = if (config.activity_sim_skimmer.fileOutputFormat.equalsIgnoreCase("csv")) "csv.gz" else "omx"
       val filePath = event.getServices.getControlerIO
-        .getIterationFilename(event.getServices.getIterationNumber, skimFileBaseName + "_current.csv.gz")
+        .getIterationFilename(event.getServices.getIterationNumber, s"${skimFileBaseName}_current.$extension")
       writePresentedSkims(filePath)
     }
 
@@ -197,6 +201,17 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
     ProfilingUtils.timed("Writing skims that are created during simulation ", x => logger.info(x)) {
       val excerptData = currentSkim
         .asInstanceOf[Map[ActivitySimSkimmerKey, ActivitySimSkimmerInternal]]
+        .flatMap {
+          case (key, value) if isWalkTransit(key.pathType) =>
+            // The WLK_TRN_WLK skim is a catch-all for all transit trips, so rather than storing extra in the skims we
+            // just duplicate the appropriate values when aggregating them up
+            Map[ActivitySimSkimmerKey, ActivitySimSkimmerInternal](
+              key                              -> value,
+              key.copy(pathType = WLK_TRN_WLK) -> value
+            )
+          case (key, value) =>
+            Map[ActivitySimSkimmerKey, ActivitySimSkimmerInternal](key -> value)
+        }
         .groupBy { case (key, _) =>
           val asTimeBin = ActivitySimTimeBin.toTimeBin(key.hour)
           ActivitySimKey(asTimeBin, key.pathType, key.origin, key.destination)
@@ -205,10 +220,16 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
           weightedData(key.timeBin.entryName, key.origin, key.destination, key.pathType, skimMap.values.toList)
         }
 
-      val csvWriter = new CsvWriter(filePath, ExcerptData.csvHeaderSeq)
-      csvWriter.writeAllAndClose(excerptData.map(_.toCsvSeq)) match {
+      val writeResult = if (config.activity_sim_skimmer.fileOutputFormat.trim.equalsIgnoreCase("csv")) {
+        val csvWriter = new CsvWriter(filePath, ExcerptData.csvHeaderSeq)
+        csvWriter.writeAllAndClose(excerptData.map(_.toCsvSeq))
+      } else {
+        val geoUnits = SortedSet[String](beamScenario.tazTreeMap.getTAZs.map(_.tazId.toString).toSeq: _*)
+        ActivitySimOmxWriter.writeToOmx(filePath, excerptData.iterator, geoUnits)
+      }
+      writeResult match {
         case Failure(exception) =>
-          logger.error(s"Cannot write to $filePath", exception)
+          logger.error(s"Cannot write skims to {}", filePath, exception)
         case _ =>
       }
     }
@@ -252,14 +273,23 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
     individualSkims: List[ActivitySimSkimmerInternal]
   ) = {
     val weights = individualSkims.map(sk => sk.observations)
-    val sumWeights = if (weights.sum == 0) 1 else weights.sum
-
-    def getWeightedSkimsValue(getValue: ActivitySimSkimmerInternal => Double): Double =
-      individualSkims.map(getValue).zip(weights).map(tup => tup._1 * tup._2).sum / sumWeights
+    val sumWeights = weights.sum
+
+    def getWeightedSkimsValue(getValue: ActivitySimSkimmerInternal => Double): Double = {
+      if (weights.sum == 0) { 0 }
+      else {
+        individualSkims
+          .map(getValue)
+          .zip(weights)
+          .map { case (value, weight) => value * weight }
+          .filter(!_.isNaN)
+          .foldLeft(0d)(_ + _) / sumWeights
+      }
+    }
 
     val weightedDistance = getWeightedSkimsValue(_.distanceInMeters)
     val weightedTotalTime = getWeightedSkimsValue(_.travelTimeInMinutes)
-    val weightedCost = getWeightedSkimsValue(_.cost)
+    val weightedCostInDollars = getWeightedSkimsValue(_.cost)
     val weightedWalkAccessTime = getWeightedSkimsValue(_.walkAccessInMinutes)
     val weightedWalkEgressTime = getWeightedSkimsValue(_.walkEgressInMinutes)
     val weightedWalkAuxiliaryTime = getWeightedSkimsValue(_.walkAuxiliaryInMinutes)
@@ -282,7 +312,7 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
       destinationId = destinationId,
       weightedTotalTime = weightedTotalTime,
       weightedTotalInVehicleTime = weightedTotalInVehicleTime,
-      weightedTotalCost = weightedCost,
+      weightedTotalFareInCents = weightedCostInDollars * 100,
       weightedDistance = weightedDistance,
       weightedWalkAccess = weightedWalkAccessTime,
       weightedWalkAuxiliary = weightedWalkAuxiliaryTime,
@@ -294,7 +324,7 @@ class ActivitySimSkimmer @Inject() (matsimServices: MatsimServices, beamScenario
       weightedKeyInVehicleTimeInMinutes = weightedKeyInVehicleTime,
       weightedFerryInVehicleTimeInMinutes = weightedFerryTime,
       weightedTransitBoardingsCount = weightedTransitBoardingsCount,
-      weightedCost = weightedCost,
+      weightedCost = weightedCostInDollars,
       failedTrips = failedTrips,
       completedTrips = completedTrips,
       debugText = debugText
@@ -387,7 +417,7 @@ object ActivitySimSkimmer extends LazyLogging {
     destinationId: String,
     weightedTotalTime: Double,
     weightedTotalInVehicleTime: Double,
-    weightedTotalCost: Double,
+    weightedTotalFareInCents: Double,
     weightedDistance: Double,
     weightedWalkAccess: Double,
     weightedWalkAuxiliary: Double,
@@ -405,13 +435,56 @@ object ActivitySimSkimmer extends LazyLogging {
     debugText: String = ""
   ) {
 
+    def getValue(metric: ActivitySimMetric): Double = {
+      metric match {
+        case ActivitySimMetric.TOTIVT   => weightedTotalInVehicleTime
+        case ActivitySimMetric.IVT      => weightedTotalInVehicleTime
+        case ActivitySimMetric.FERRYIVT => weightedFerryInVehicleTimeInMinutes
+        case ActivitySimMetric.FAR      => weightedTotalFareInCents
+        case ActivitySimMetric.KEYIVT   => weightedKeyInVehicleTimeInMinutes
+        case ActivitySimMetric.DTIM     => weightedDriveTimeInMinutes
+        case ActivitySimMetric.BOARDS   => weightedTransitBoardingsCount
+        case ActivitySimMetric.DDIST    => weightedDriveDistanceInMeters
+        case ActivitySimMetric.WAUX     => weightedWalkAuxiliary
+        case ActivitySimMetric.TIME     => weightedTotalTime
+        case ActivitySimMetric.DIST     => weightedDistance
+        case ActivitySimMetric.WEGR     => weightedWalkEgress
+        case ActivitySimMetric.WACC     => weightedWalkAccess
+        case ActivitySimMetric.IWAIT    => weightedWaitInitial
+        case ActivitySimMetric.XWAIT    => weightedWaitTransfer
+        case ActivitySimMetric.TRIPS    => completedTrips
+        case ActivitySimMetric.FAILURES => failedTrips
+        case _                          => Double.NaN
+      }
+    }
+
     def toCsvString: String = productIterator.mkString("", ",", "\n")
 
     def toCsvSeq: Seq[Any] = productIterator.toSeq
   }
 
   object ExcerptData {
 
+    val supportedActivitySimMetric: Set[ActivitySimMetric] = Set(
+      ActivitySimMetric.TOTIVT,
+      ActivitySimMetric.IVT,
+      ActivitySimMetric.FERRYIVT,
+      ActivitySimMetric.FAR,
+      ActivitySimMetric.KEYIVT,
+      ActivitySimMetric.DTIM,
+      ActivitySimMetric.BOARDS,
+      ActivitySimMetric.DDIST,
+      ActivitySimMetric.WAUX,
+      ActivitySimMetric.TIME,
+      ActivitySimMetric.DIST,
+      ActivitySimMetric.WEGR,
+      ActivitySimMetric.WACC,
+      ActivitySimMetric.IWAIT,
+      ActivitySimMetric.XWAIT,
+      ActivitySimMetric.TRIPS,
+      ActivitySimMetric.FAILURES
+    )
+
     val csvHeaderSeq: Seq[String] = Seq(
       "timePeriod",
       "pathType",