Merge (current/future) feed versions for MTC

.travis.yml

@@ -1,14 +1,20 @@
+dist: trusty # jdk 8 not available on xenial
 language: java
-jdk:
-- oraclejdk8
+language: java
+java:
+  - oraclejdk8
 install: true
 sudo: false
 # Install mongoDB to perform persistence tests
-services: mongodb
+services:
+  - mongodb
+  - postgresql
+addons:
+  postgresql: 9.6
 cache:
   directories:
-    - "$HOME/.m2"
-    - "$HOME/.cache/yarn"
+    - $HOME/.m2
+    - $HOME/.cache/yarn
 # Install semantic-release
 before_script:
   - yarn global add @conveyal/maven-semantic-release semantic-release@15
@@ -17,6 +23,8 @@ before_install:
 # set region in AWS config for S3 setup
 - mkdir ~/.aws && printf '%s\n' '[default]' 'aws_access_key_id=foo' 'aws_secret_access_key=bar' 'region=us-east-1' > ~/.aws/config
 - cp configurations/default/server.yml.tmp configurations/default/server.yml
+# create database for tests
+- psql -U postgres -c 'CREATE DATABASE catalogue;'
 script:
 # package jar
 - mvn package

pom.xml

@@ -243,11 +243,22 @@
             <version>2.1.0</version>
         </dependency>
 
-        <!-- Used for loading/fetching/writing GTFS entities (also provides access to commons-io and AWS S3 SDK). -->
+        <!-- Used for testing (note: this should match the version in gtfs-lib). -->
+        <dependency>
+            <groupId>junit</groupId>
+            <artifactId>junit</artifactId>
+            <version>4.12</version>
+            <scope>test</scope>
+        </dependency>
+
+        <!-- Used for loading/fetching/writing GTFS entities. gtfs-lib also provides access to:
+         - commons-io - generic utilities
+         - AWS S3 SDK - putting/getting objects into/out of S3.
+         -->
         <dependency>
             <groupId>com.conveyal</groupId>
             <artifactId>gtfs-lib</artifactId>
-            <version>4.3.2</version>
+            <version>4.3.4</version>
         </dependency>
 
         <!-- Used for data-tools application database -->
@@ -313,14 +324,6 @@
             <version>19.2</version>
         </dependency>
 
-        <!-- Unit testing -->
-        <dependency>
-            <groupId>junit</groupId>
-            <artifactId>junit</artifactId>
-            <version>4.12</version>
-            <scope>test</scope>
-        </dependency>
-
         <!-- Error reporting -->
         <dependency>
             <groupId>com.bugsnag</groupId>
@@ -363,6 +366,16 @@
             <version>2.14.0</version>
             <scope>test</scope>
         </dependency>
+        <!-- Used for writing csv for merged feeds. Note: this appears to be one of the only
+        CSV libraries that will only quote values when necessary (e.g., there is a comma character
+        contained within the value) and that will work with an output stream writer when writing
+        directly to a zip output stream.
+         -->
+        <dependency>
+            <groupId>net.sf.supercsv</groupId>
+            <artifactId>super-csv</artifactId>
+            <version>2.4.0</version>
+        </dependency>
     </dependencies>
 
 </project>

src/main/java/com/conveyal/datatools/common/status/MonitorableJob.java

@@ -57,7 +57,7 @@ public enum JobType {
         EXPORT_SNAPSHOT_TO_GTFS,
         CONVERT_EDITOR_MAPDB_TO_SQL,
         VALIDATE_ALL_FEEDS,
-        MERGE_PROJECT_FEEDS
+        MERGE_FEED_VERSIONS
     }
 
     public MonitorableJob(String owner, String name, JobType type) {

src/main/java/com/conveyal/datatools/manager/controllers/api/FeedVersionController.java

@@ -1,8 +1,11 @@
 package com.conveyal.datatools.manager.controllers.api;
 
+import com.conveyal.datatools.common.utils.SparkUtils;
 import com.conveyal.datatools.manager.DataManager;
 import com.conveyal.datatools.manager.auth.Auth0UserProfile;
 import com.conveyal.datatools.manager.jobs.CreateFeedVersionFromSnapshotJob;
+import com.conveyal.datatools.manager.jobs.MergeFeedsJob;
+import com.conveyal.datatools.manager.jobs.MergeFeedsType;
 import com.conveyal.datatools.manager.jobs.ProcessSingleFeedJob;
 import com.conveyal.datatools.manager.models.FeedDownloadToken;
 import com.conveyal.datatools.manager.models.FeedSource;
@@ -13,6 +16,7 @@
 import com.conveyal.datatools.manager.persistence.Persistence;
 import com.conveyal.datatools.manager.utils.HashUtils;
 import com.conveyal.datatools.manager.utils.json.JsonManager;
+
 import com.fasterxml.jackson.databind.JsonNode;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -23,13 +27,16 @@
 import java.io.File;
 import java.util.Collection;
 import java.util.Date;
+import java.util.HashSet;
+import java.util.Set;
 
 import static com.conveyal.datatools.common.utils.S3Utils.downloadFromS3;
 import static com.conveyal.datatools.common.utils.SparkUtils.copyRequestStreamIntoFile;
 import static com.conveyal.datatools.common.utils.SparkUtils.downloadFile;
 import static com.conveyal.datatools.common.utils.SparkUtils.formatJobMessage;
 import static com.conveyal.datatools.common.utils.SparkUtils.logMessageAndHalt;
 import static com.conveyal.datatools.manager.controllers.api.FeedSourceController.checkFeedSourcePermissions;
+import static com.conveyal.datatools.manager.jobs.MergeFeedsType.REGIONAL;
 import static spark.Spark.delete;
 import static spark.Spark.get;
 import static spark.Spark.post;
@@ -262,6 +269,52 @@ private static FeedVersion publishToExternalResource (Request req, Response res)
         }
     }
 
+    /**
+     * HTTP controller that handles merging multiple feed versions for a given feed source, with version IDs specified
+     * in a comma-separated string in the feedVersionIds query parameter and merge type specified in mergeType query
+     * parameter. NOTE: REGIONAL merge type should only be handled through {@link ProjectController#mergeProjectFeeds(Request, Response)}.
+     */
+    private static String mergeFeedVersions(Request req, Response res) {
+        String[] versionIds = req.queryParams("feedVersionIds").split(",");
+        // Try to parse merge type (null or bad value throws IllegalArgumentException).
+        MergeFeedsType mergeType;
+        try {
+            mergeType = MergeFeedsType.valueOf(req.queryParams("mergeType"));
+            if (mergeType.equals(REGIONAL)) {
+                throw new IllegalArgumentException("Regional merge type is not permitted for this endpoint.");
+            }
+        } catch (IllegalArgumentException e) {
+            logMessageAndHalt(req, 400, "Must provide valid merge type.", e);
+            return null;
+        }
+        // Collect versions to merge (must belong to same feed source).
+        Set<FeedVersion> versions = new HashSet<>();
+        String feedSourceId = null;
+        for (String id : versionIds) {
+            FeedVersion v = Persistence.feedVersions.getById(id);
+            if (v == null) {
+                logMessageAndHalt(req,
+                                  400,
+                                  String.format("Must provide valid version ID. (No version exists for id=%s.)", id)
+                );
+            }
+            // Store feed source id and check other versions for matching.
+            if (feedSourceId == null) feedSourceId = v.feedSourceId;
+            else if (!v.feedSourceId.equals(feedSourceId)) {
+                logMessageAndHalt(req, 400, "Cannot merge versions with different parent feed sources.");
+            }
+            versions.add(v);
+        }
+        if (versionIds.length != 2) {
+            logMessageAndHalt(req, 400, "Merging more than two versions is not currently supported.");
+        }
+        // Kick off merge feeds job.
+        Auth0UserProfile userProfile = req.attribute("user");
+        MergeFeedsJob mergeFeedsJob = new MergeFeedsJob(userProfile.getUser_id(), versions, "merged", mergeType);
+        DataManager.heavyExecutor.execute(mergeFeedsJob);
+        return SparkUtils.formatJobMessage(mergeFeedsJob.jobId, "Merging feed versions...");
+    }
+
     /**
      * Download locally stored feed version with token supplied by this application. This method is only used when
      * useS3 is set to false. Otherwise, a direct download from s3 should be used.
@@ -300,6 +353,7 @@ public static void register (String apiPrefix) {
         post(apiPrefix + "secure/feedversion", FeedVersionController::createFeedVersionViaUpload, json::write);
         post(apiPrefix + "secure/feedversion/fromsnapshot", FeedVersionController::createFeedVersionFromSnapshot, json::write);
         put(apiPrefix + "secure/feedversion/:id/rename", FeedVersionController::renameFeedVersion, json::write);
+        put(apiPrefix + "secure/feedversion/merge", FeedVersionController::mergeFeedVersions, json::write);
         post(apiPrefix + "secure/feedversion/:id/publish", FeedVersionController::publishToExternalResource, json::write);
         delete(apiPrefix + "secure/feedversion/:id", FeedVersionController::deleteFeedVersion, json::write);
 

src/main/java/com/conveyal/datatools/manager/controllers/api/ProjectController.java

@@ -5,8 +5,9 @@
 import com.conveyal.datatools.manager.auth.Auth0UserProfile;
 import com.conveyal.datatools.manager.jobs.FetchProjectFeedsJob;
 import com.conveyal.datatools.manager.jobs.MakePublicJob;
-import com.conveyal.datatools.manager.jobs.MergeProjectFeedsJob;
+import com.conveyal.datatools.manager.jobs.MergeFeedsJob;
 import com.conveyal.datatools.manager.models.FeedDownloadToken;
+import com.conveyal.datatools.manager.models.FeedSource;
 import com.conveyal.datatools.manager.models.FeedVersion;
 import com.conveyal.datatools.manager.models.JsonViews;
 import com.conveyal.datatools.manager.models.Project;
@@ -20,13 +21,16 @@
 import spark.Response;
 
 import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
 import java.util.stream.Collectors;
 
 import static com.conveyal.datatools.common.utils.S3Utils.downloadFromS3;
 import static com.conveyal.datatools.common.utils.SparkUtils.downloadFile;
 import static com.conveyal.datatools.common.utils.SparkUtils.formatJobMessage;
 import static com.conveyal.datatools.common.utils.SparkUtils.logMessageAndHalt;
 import static com.conveyal.datatools.manager.DataManager.publicPath;
+import static com.conveyal.datatools.manager.jobs.MergeFeedsType.REGIONAL;
 import static spark.Spark.delete;
 import static spark.Spark.get;
 import static spark.Spark.post;
@@ -216,14 +220,28 @@ private static Project checkProjectPermissions(Request req, Project project, Str
      * to getFeedDownloadCredentials with the project ID to obtain either temporary S3 credentials or a download token
      * (depending on application configuration "application.data.use_s3_storage") to download the zip file.
      */
-    private static String downloadMergedFeed(Request req, Response res) {
+    static String mergeProjectFeeds(Request req, Response res) {
         Project project = requestProjectById(req, "view");
         Auth0UserProfile userProfile = req.attribute("user");
         // TODO: make this an authenticated call?
-        MergeProjectFeedsJob mergeProjectFeedsJob = new MergeProjectFeedsJob(project, userProfile.getUser_id());
-        DataManager.heavyExecutor.execute(mergeProjectFeedsJob);
+        Set<FeedVersion> feedVersions = new HashSet<>();
+        // Get latest version for each feed source in project
+        Collection<FeedSource> feedSources = project.retrieveProjectFeedSources();
+        for (FeedSource fs : feedSources) {
+            // check if feed version exists
+            FeedVersion version = fs.retrieveLatest();
+            if (version == null) {
+                LOG.warn("Skipping {} because it has no feed versions", fs.name);
+                continue;
+            }
+            // modify feed version to use prepended feed id
+            LOG.info("Adding {} feed to merged zip", fs.name);
+            feedVersions.add(version);
+        }
+        MergeFeedsJob mergeFeedsJob = new MergeFeedsJob(userProfile.getUser_id(), feedVersions, project.id, REGIONAL);
+        DataManager.heavyExecutor.execute(mergeFeedsJob);
         // Return job ID to requester for monitoring job status.
-        return formatJobMessage(mergeProjectFeedsJob.jobId, "Merge operation is processing.");
+        return formatJobMessage(mergeFeedsJob.jobId, "Merge operation is processing.");
     }
 
     /**
@@ -310,7 +328,7 @@ public static void register (String apiPrefix) {
         post(apiPrefix + "secure/project/:id/fetch", ProjectController::fetch, json::write);
         post(apiPrefix + "secure/project/:id/deployPublic", ProjectController::publishPublicFeeds, json::write);
 
-        get(apiPrefix + "secure/project/:id/download", ProjectController::downloadMergedFeed);
+        get(apiPrefix + "secure/project/:id/download", ProjectController::mergeProjectFeeds);
         get(apiPrefix + "secure/project/:id/downloadtoken", ProjectController::getFeedDownloadCredentials, json::write);
 
         get(apiPrefix + "public/project/:id", ProjectController::getProject, json::write);

src/main/java/com/conveyal/datatools/manager/gtfsplus/CalendarAttribute.java

@@ -0,0 +1,20 @@
+package com.conveyal.datatools.manager.gtfsplus;
+
+import com.conveyal.gtfs.model.Entity;
+
+import javax.naming.OperationNotSupportedException;
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+public class CalendarAttribute extends Entity {
+
+    private static final long serialVersionUID = 1L;
+
+    public String service_id;
+    public String service_description;
+
+    @Override public void setStatementParameters(PreparedStatement statement, boolean setDefaultId) {
+        throw new UnsupportedOperationException(
+            "Cannot call setStatementParameters because loading a GTFS+ table into RDBMS is unsupported.");
+    }
+}

src/main/java/com/conveyal/datatools/manager/gtfsplus/Direction.java

@@ -0,0 +1,22 @@
+package com.conveyal.datatools.manager.gtfsplus;
+
+import com.conveyal.gtfs.model.Entity;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+public class Direction extends Entity {
+
+    private static final long serialVersionUID = 1L;
+
+    public String route_id;
+    public int direction_id;
+    public String direction;
+
+
+    @Override
+    public void setStatementParameters(PreparedStatement statement, boolean setDefaultId) throws SQLException {
+        throw new UnsupportedOperationException(
+            "Cannot call setStatementParameters because loading a GTFS+ table into RDBMS is unsupported.");
+    }
+}

src/main/java/com/conveyal/datatools/manager/gtfsplus/FareRiderCategory.java

@@ -0,0 +1,24 @@
+package com.conveyal.datatools.manager.gtfsplus;
+
+import com.conveyal.gtfs.model.Entity;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.time.LocalDate;
+
+public class FareRiderCategory extends Entity {
+
+    private static final long serialVersionUID = 1L;
+
+    public String fare_id;
+    public int rider_category_id;
+    public double price;
+    public LocalDate expiration_date;
+    public LocalDate commencement_date;
+
+    @Override
+    public void setStatementParameters(PreparedStatement statement, boolean setDefaultId) throws SQLException {
+        throw new UnsupportedOperationException(
+            "Cannot call setStatementParameters because loading a GTFS+ table into RDBMS is unsupported.");
+    }
+}

src/main/java/com/conveyal/datatools/manager/gtfsplus/FareZoneAttribute.java

@@ -0,0 +1,20 @@
+package com.conveyal.datatools.manager.gtfsplus;
+
+import com.conveyal.gtfs.model.Entity;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+
+public class FareZoneAttribute extends Entity {
+
+    private static final long serialVersionUID = 1L;
+
+    public String zone_id;
+    public String zone_name;
+
+    @Override
+    public void setStatementParameters(PreparedStatement statement, boolean setDefaultId) throws SQLException {
+        throw new UnsupportedOperationException(
+            "Cannot call setStatementParameters because loading a GTFS+ table into RDBMS is unsupported.");
+    }
+}