Sync up with master: Part 1

build.gradle

@@ -99,7 +99,7 @@ allprojects {
 dependencies {
 
     //beam-utilities
-    compile group: 'com.github.LBNL-UCB-STI', name: 'beam-utilities', version: 'v0.1'
+    compile group: 'com.github.LBNL-UCB-STI', name: 'beam-utilities', version: 'v0.2.1'
     ////////////////////////////
     // Java dependencies
     ////////////////////////////

src/main/java/beam/router/r5/R5MnetBuilder.java

@@ -1,6 +1,7 @@
 package beam.router.r5;
 
 import beam.sim.config.BeamConfig;
+import beam.utils.osm.WayFixer$;
 import com.conveyal.osmlib.OSM;
 import com.conveyal.osmlib.Way;
 import com.conveyal.r5.streets.EdgeStore;
@@ -56,18 +57,21 @@ public R5MnetBuilder(TransportNetwork r5Net, BeamConfig beamConfig) {
 
     public void buildMNet() {
         // Load the OSM file for retrieving the number of lanes, which is not stored in the R5 network
-//        OSM osm = new OSM(osmFile);
         Map<Long, Way> ways = new OSM(osmFile).ways;
+        WayFixer$.MODULE$.fix(ways);
+
         EdgeStore.Edge cursor = r5Network.streetLayer.edgeStore.getCursor();  // Iterator of edges in R5 network
         OsmToMATSim OTM = new OsmToMATSim(mNetwork, true);
+
+        int numberOfFixes = 0;
+        HashMap<String, Integer> highwayTypeToCounts = new HashMap<>();
+
         while (cursor.advance()) {
 //            log.debug("Edge Index:{}. Cursor {}.", cursor.getEdgeIndex(), cursor);
             // TODO - eventually, we should pass each R5 link to OsmToMATSim and do the two-way handling there.
             // Check if we have already seen this OSM way. Skip if we have.
             Integer edgeIndex = cursor.getEdgeIndex();
-
             Long osmID = cursor.getOSMID();  // id of edge in the OSM db
-
             Way way = ways.get(osmID);
 
             Set<Integer> deezNodes = new HashSet<>(2);
@@ -93,16 +97,28 @@ public void buildMNet() {
             // Grab existing nodes from mNetwork if they already exist, else make new ones and add to mNetwork
             Node fromNode = getOrMakeNode(fromCoord);
             Node toNode = getOrMakeNode(toCoord);
+            Link link = null;
             if (way == null) {
                 // Made up numbers, this is a PT to road network connector or something
-                Link link = buildLink(edgeIndex, flagStrings, length, fromNode, toNode);
+                link = buildLink(edgeIndex, flagStrings, length, fromNode, toNode);
                 mNetwork.addLink(link);
                 log.debug("Created special link: {}", link);
             } else {
-                Link link = OTM.createLink(way, osmID, edgeIndex, fromNode, toNode, length, (HashSet<String>)flagStrings);
+                link = OTM.createLink(way, osmID, edgeIndex, fromNode, toNode, length, (HashSet<String>)flagStrings);
                 mNetwork.addLink(link);
                 log.debug("Created regular link: {}", link);
             }
+            if (fromNode.getId() == toNode.getId()) {
+                cursor.setLengthMm(1);
+                cursor.setSpeed((short)2905); // 65 miles per hour
+                link.setLength(0.001);
+                link.setCapacity(10000);
+                link.setFreespeed(29.0576);   // 65 miles per hour
+                numberOfFixes += 1;
+            }
+        }
+        if (numberOfFixes > 0) {
+            log.warn("Fixed {} links which were having the same `fromNode` and `toNode`", numberOfFixes);
         }
     }
 

src/main/resources/logback.xml

@@ -87,7 +87,7 @@
 
     </Logger>
 
-    <Logger name="beam.router.BeamRouter" level="ERROR" additivity="false">
+    <Logger name="beam.router.BeamRouter" level="INFO" additivity="false">
         <appender-ref ref="STDOUT"/>
     </Logger>
 

src/main/scala/beam/agentsim/agents/PersonAgent.scala

@@ -3,8 +3,8 @@ package beam.agentsim.agents
 import akka.actor.FSM.Failure
 import akka.actor.{ActorRef, FSM, Props, Stash}
 import beam.agentsim.Resource._
-import beam.agentsim.agents.PersonAgent._
 import beam.agentsim.agents.BeamAgent._
+import beam.agentsim.agents.PersonAgent._
 import beam.agentsim.agents.household.HouseholdActor.{CancelCAVTrip, ReleaseVehicle}
 import beam.agentsim.agents.modalbehaviors.ChoosesMode.ChoosesModeData
 import beam.agentsim.agents.modalbehaviors.DrivesVehicle._
@@ -17,19 +17,18 @@ import beam.agentsim.agents.ridehail.RideHailManager.TravelProposal
 import beam.agentsim.agents.ridehail._
 import beam.agentsim.agents.vehicles.BeamVehicle.FuelConsumed
 import beam.agentsim.agents.vehicles.VehicleCategory.Bike
-import beam.agentsim.agents.vehicles.VehicleProtocol.RemovePassengerFromTrip
 import beam.agentsim.agents.vehicles._
 import beam.agentsim.events._
 import beam.agentsim.events.resources.ReservationError
 import beam.agentsim.infrastructure.ParkingManager.ParkingInquiryResponse
 import beam.agentsim.scheduler.BeamAgentScheduler.{CompletionNotice, IllegalTriggerGoToError, ScheduleTrigger}
 import beam.agentsim.scheduler.Trigger
 import beam.agentsim.scheduler.Trigger.TriggerWithId
-import beam.router.{BeamSkimmer, RouteHistory}
 import beam.router.Modes.BeamMode
-import beam.router.Modes.BeamMode.{CAR, CAV, RIDE_HAIL_POOLED, WALK, WALK_TRANSIT}
+import beam.router.Modes.BeamMode.{CAR, CAV, WALK, WALK_TRANSIT}
 import beam.router.model.{EmbodiedBeamLeg, EmbodiedBeamTrip}
 import beam.router.osm.TollCalculator
+import beam.router.{BeamSkimmer, RouteHistory, TravelTimeObserved}
 import beam.sim.BeamServices
 import beam.sim.population.AttributesOfIndividual
 import com.conveyal.r5.transit.TransportNetwork
@@ -218,6 +217,8 @@ class PersonAgent(
     with ChoosesParking
     with Stash {
 
+  val travelTimeObserved: TravelTimeObserved = beamServices.injector.getInstance(classOf[TravelTimeObserved])
+
   val body = new BeamVehicle(
     bodyVehicleIdFromPersonID(id),
     BeamVehicleType.powerTrainForHumanBody,
@@ -775,6 +776,12 @@ class PersonAgent(
             curFuelConsumed.primaryFuel + curFuelConsumed.secondaryFuel,
             beamServices
           )
+          travelTimeObserved.observeTrip(
+            correctedTrip,
+            generalizedTime,
+            generalizedCost,
+            curFuelConsumed.primaryFuel + curFuelConsumed.secondaryFuel
+          )
           resetFuelConsumed()
 
           eventsManager.processEvent(

src/main/scala/beam/agentsim/agents/modalbehaviors/DrivesVehicle.scala

@@ -146,8 +146,12 @@ trait DrivesVehicle[T <: DrivingData] extends BeamAgent[T] with HasServices with
       val isLastLeg = data.currentLegPassengerScheduleIndex + 1 == data.passengerSchedule.schedule.size
       val fuelConsumed = currentBeamVehicle.useFuel(currentLeg, beamServices)
 
-      val nbPassengers = data.passengerSchedule.schedule(currentLeg).riders.size
+      var nbPassengers = data.passengerSchedule.schedule(currentLeg).riders.size
       if (nbPassengers > 0) {
+        if (currentLeg.mode.isTransit) {
+          nbPassengers =
+            (nbPassengers / beamServices.beamConfig.beam.agentsim.tuning.transitCapacity.getOrElse(1.0)).toInt
+        }
         data.passengerSchedule.schedule(currentLeg).riders foreach { rider =>
           updateFuelConsumedByTrip(rider.personId, fuelConsumed, nbPassengers)
         }

src/main/scala/beam/router/BeamRouter.scala

@@ -32,6 +32,7 @@ import beam.router.gtfs.FareCalculator
 import beam.router.model._
 import beam.router.osm.TollCalculator
 import beam.router.r5.R5RoutingWorker
+import beam.router.r5.R5RoutingWorker.MinSpeedUsage
 import beam.sim.BeamServices
 import beam.sim.population.AttributesOfIndividual
 import beam.utils.IdGeneratorImpl
@@ -240,6 +241,14 @@ class BeamRouter(
     case ClearRoutedWorkerTracker(workIdToClear) =>
       //TODO: Maybe do this for all tracker removals?
       removeOutstandingWorkBy(workIdToClear)
+    case IterationFinished(iteration) =>
+      remoteNodes.foreach(workerAddress => workerFrom(workerAddress) ! IterationFinished(iteration))
+      localNodes.foreach(_ ! IterationFinished(iteration))
+    case MinSpeedUsage(iteration, count) =>
+      log.info(
+        s"Worker[${sender()}]. Iteration $iteration had $count cases when min speed ${services.beamConfig.beam.physsim.quick_fix_minCarSpeedInMetersPerSecond} was used."
+      )
+
     case work =>
       val originalSender = context.sender
       if (!isWorkAvailable) { //No existing work
@@ -454,6 +463,7 @@ object BeamRouter {
 
   case class TryToSerialize(obj: Object)
   case class UpdateTravelTimeRemote(linkIdToTravelTimePerHour: java.util.Map[String, Array[Double]])
+  case class IterationFinished(iteration: Int)
 
   /**
     * It is use to represent a request object

src/main/scala/beam/router/BeamSkimmer.scala

@@ -1,6 +1,5 @@
 package beam.router
 
-import java.io._
 import java.util.concurrent.TimeUnit
 
 import beam.agentsim.agents.choice.mode.DrivingCost
@@ -203,16 +202,18 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
     val mode = trip.tripClassifier
     val correctedTrip = mode match {
       case WALK =>
-        trip.beamLegs()
+        trip
       case _ =>
-        trip.beamLegs().drop(1).dropRight(1)
+        val legs = trip.legs.drop(1).dropRight(1)
+        EmbodiedBeamTrip(legs)
     }
-    val origLeg = correctedTrip.head
+    val beamLegs = correctedTrip.beamLegs()
+    val origLeg = beamLegs.head
     val origCoord = beamServices.geo.wgs2Utm(origLeg.travelPath.startPoint.loc)
     val origTaz = beamServices.tazTreeMap
       .getTAZ(origCoord.getX, origCoord.getY)
       .tazId
-    val destLeg = correctedTrip.last
+    val destLeg = beamLegs.last
     val destCoord = beamServices.geo.wgs2Utm(destLeg.travelPath.endPoint.loc)
     val destTaz = beamServices.tazTreeMap
       .getTAZ(destCoord.getX, destCoord.getY)
@@ -221,11 +222,11 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
     val key = (timeBin, mode, origTaz, destTaz)
     val payload =
       SkimInternal(
-        trip.totalTravelTimeInSecs.toDouble,
+        correctedTrip.totalTravelTimeInSecs.toDouble,
         generalizedTimeInHours * 3600,
         generalizedCost,
-        trip.beamLegs().map(_.travelPath.distanceInM).sum,
-        trip.costEstimate,
+        beamLegs.map(_.travelPath.distanceInM).sum,
+        correctedTrip.costEstimate,
         1,
         energyConsumption
       )
@@ -258,8 +259,11 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
     ProfilingUtils.timed(s"writeObservedSkims on iteration ${event.getIteration}", x => logger.info(x)) {
       writeObservedSkims(event)
     }
-    ProfilingUtils.timed(s"writeCarSkimsForPeakNonPeakPeriods on iteration ${event.getIteration}", x => logger.info(x)) {
-      writeCarSkimsForPeakNonPeakPeriods(event)
+    ProfilingUtils.timed(
+      s"writeAllModeSkimsForPeakNonPeakPeriods on iteration ${event.getIteration}",
+      x => logger.info(x)
+    ) {
+      writeAllModeSkimsForPeakNonPeakPeriods(event)
     }
     // Writing full skims are very large, but code is preserved here in case we want to enable it.
     // TODO make this a configurable output "writeFullSkimsInterval" with default of 0
@@ -269,16 +273,15 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
     skims = new TrieMap()
   }
 
-  def averageAndWriteSkims(
+  def getExcerptData(
     timePeriodString: String,
     hoursIncluded: List[Int],
     origin: TAZ,
     destination: TAZ,
-    mode: BeamMode.CAR.type,
+    mode: BeamMode,
     get: BeamServices,
-    dummyId: Id[BeamVehicleType],
-    writer: BufferedWriter
-  ): Unit = {
+    dummyId: Id[BeamVehicleType]
+  ): ExcerptData = {
     val individualSkims = hoursIncluded.map { timeBin =>
       getSkimValue(timeBin * 3600, mode, origin.tazId, destination.tazId)
         .map(_.toSkimExternal)
@@ -318,16 +321,26 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
       .sum / sumWeights
     val weightedEnergy = individualSkims.map(_.energy).zip(weights).map(tup => tup._1 * tup._2).sum / sumWeights
 
-    writer.write(
-      s"$timePeriodString,$mode,${origin.tazId},${destination.tazId},${weightedTime},${weightedGeneralizedTime},${weightedCost},${weightedGeneralizedCost},${weightedDistance},${sumWeights},$weightedEnergy\n"
+    ExcerptData(
+      timePeriodString = timePeriodString,
+      mode = mode,
+      originTazId = origin.tazId,
+      destinationTazId = destination.tazId,
+      weightedTime = weightedTime,
+      weightedGeneralizedTime = weightedGeneralizedTime,
+      weightedCost = weightedCost,
+      weightedGeneralizedCost = weightedGeneralizedCost,
+      weightedDistance = weightedDistance,
+      sumWeights = sumWeights,
+      weightedEnergy = weightedEnergy
     )
   }
 
-  def writeCarSkimsForPeakNonPeakPeriods(event: IterationEndsEvent): Unit = {
+  def writeAllModeSkimsForPeakNonPeakPeriods(event: IterationEndsEvent): Unit = {
     val morningPeakHours = (7 to 8).toList
     val afternoonPeakHours = (15 to 16).toList
     val nonPeakHours = (0 to 6).toList ++ (9 to 14).toList ++ (17 to 23).toList
-    val modes = List(CAR)
+    val modes = BeamMode.allModes
     val fileHeader =
       "period,mode,origTaz,destTaz,travelTimeInS,generalizedTimeInS,cost,generalizedCost,distanceInM,numObservations,energy"
     val filePath = event.getServices.getControlerIO.getIterationFilename(
@@ -339,44 +352,49 @@ class BeamSkimmer @Inject()(val beamConfig: BeamConfig, val beamServices: BeamSe
     writer.write(fileHeader)
     writer.write("\n")
 
-    beamServices.tazTreeMap.getTAZs
-      .foreach { origin =>
-        beamServices.tazTreeMap.getTAZs.foreach { destination =>
-          modes.foreach { mode =>
-            averageAndWriteSkims(
+    val weightedSkims = ProfilingUtils.timed("Get weightedSkims for modes", x => logger.info(x)) {
+      modes.toParArray.flatMap { mode =>
+        beamServices.tazTreeMap.getTAZs.flatMap { origin =>
+          beamServices.tazTreeMap.getTAZs.flatMap { destination =>
+            val am = getExcerptData(
               "AM",
               morningPeakHours,
               origin,
               destination,
               mode,
               beamServices,
-              dummyId,
-              writer
+              dummyId
             )
-            averageAndWriteSkims(
+            val pm = getExcerptData(
               "PM",
               afternoonPeakHours,
               origin,
               destination,
               mode,
               beamServices,
-              dummyId,
-              writer
+              dummyId
             )
-            averageAndWriteSkims(
+            val offPeak = getExcerptData(
               "OffPeak",
               nonPeakHours,
               origin,
               destination,
               mode,
               beamServices,
-              dummyId,
-              writer
+              dummyId
             )
+            List(am, pm, offPeak)
           }
         }
       }
+    }
+    logger.info(s"weightedSkims size: ${weightedSkims.size}")
 
+    weightedSkims.foreach { ws =>
+      writer.write(
+        s"${ws.timePeriodString},${ws.mode},${ws.originTazId},${ws.destinationTazId},${ws.weightedTime},${ws.weightedGeneralizedTime},${ws.weightedCost},${ws.weightedGeneralizedCost},${ws.weightedDistance},${ws.sumWeights},${ws.weightedEnergy}\n"
+      )
+    }
     writer.close()
   }
 
@@ -518,6 +536,20 @@ object BeamSkimmer extends LazyLogging {
     energy: Double
   )
 
+  case class ExcerptData(
+    timePeriodString: String,
+    mode: BeamMode,
+    originTazId: Id[TAZ],
+    destinationTazId: Id[TAZ],
+    weightedTime: Double,
+    weightedGeneralizedTime: Double,
+    weightedCost: Double,
+    weightedGeneralizedCost: Double,
+    weightedDistance: Double,
+    sumWeights: Double,
+    weightedEnergy: Double
+  )
+
   private def readCsvFile(filePath: String): TrieMap[(Int, BeamMode, Id[TAZ], Id[TAZ]), SkimInternal] = {
     val mapReader = new CsvMapReader(FileUtils.readerFromFile(filePath), CsvPreference.STANDARD_PREFERENCE)
     val res = TrieMap[(Int, BeamMode, Id[TAZ], Id[TAZ]), SkimInternal]()