Add server-side support for storing trip and place additions

bin/debug/load_timeline_for_day_and_user.py

@@ -41,6 +41,6 @@
             print("About to save %s" % entry)
         munged_entries.append(entry)
 
-    (tsdb_count, ucdb_count) = estcs.insert_entries(override_uuid, munged_entries)
+    (tsdb_count, ucdb_count) = estcs.insert_entries(override_uuid, munged_entries, continue_on_error=False)
     print("Finished loading %d entries into the usercache and %d entries into the timeseries" %
         (ucdb_count, tsdb_count))

emission/analysis/userinput/matcher.py

@@ -9,6 +9,7 @@
 import emission.storage.decorations.trip_queries as esdt
 import emission.storage.pipeline_queries as epq
 import emission.core.wrapper.entry as ecwe
+import emission.core.wrapper.tripuserinput as ecwtui
 
 obj_to_dict_key = lambda key: key.split("/")[1]
 
@@ -27,29 +28,52 @@ def match_incoming_user_inputs(user_id):
 
 def match_incoming_inputs(user_id, timerange):
     ts = esta.TimeSeries.get_time_series(user_id)
-    input_key_list = eac.get_config()["userinput.keylist"]
+    single_most_recent_match_key_list = eac.get_config()["userinput.keylist"]
+    multi_non_deleted_key_list = ["manual/trip_addition_input"]
+    input_key_list = single_most_recent_match_key_list + multi_non_deleted_key_list
     toMatchInputs = [ecwe.Entry(e) for e in ts.find_entries(input_key_list, time_query=timerange)]
-    logging.debug("Matching %d inputs to trips" % len(toMatchInputs))
+    logging.debug("Matching %d single inputs to trips" % len(toMatchInputs))
     lastInputProcessed = None
     if len(toMatchInputs) == 0:
         logging.debug("len(toMatchInputs) == 0, early return")
         return None
     for ui in toMatchInputs:
         confirmed_trip = esdt.get_trip_for_user_input_obj(ts, ui)
         if confirmed_trip is not None:
-            input_name = obj_to_dict_key(ui.metadata.key)
-            if input_name == "trip_user_input":
-                confirmed_trip["data"]["user_input"][input_name] = ui
+            if ui.metadata.key in single_most_recent_match_key_list:
+                handle_single_most_recent_match(confirmed_trip, ui)
+            elif ui.metadata.key in multi_non_deleted_key_list:
+                handle_multi_non_deleted_match(confirmed_trip, ui)
             else:
-                confirmed_trip["data"]["user_input"][input_name] = ui.data.label
+                assert False, "Found weird key {ui.metadata.key} that was not in the search list"
             import emission.storage.timeseries.builtin_timeseries as estbt
             estbt.BuiltinTimeSeries.update(confirmed_trip)
         else:
-            logging.warn("No match found for user input %s, moving forward anyway" % ui)
+            logging.warn("No match found for single user input %s, moving forward anyway" % ui)
         lastInputProcessed = ui
 
     return lastInputProcessed
 
+def handle_single_most_recent_match(confirmed_trip, ui):
+    input_name = obj_to_dict_key(ui.metadata.key)
+    if input_name == "trip_user_input":
+        confirmed_trip["data"]["user_input"][input_name] = ui
+    else:
+        confirmed_trip["data"]["user_input"][input_name] = ui.data.label
+
+def handle_multi_non_deleted_match(confirmed_trip, ui):
+    logging.debug(f"handling user input {ui} for {confirmed_trip}")
+    if "trip_addition" not in confirmed_trip["data"] or \
+        confirmed_trip["data"]["trip_addition"] is None:
+        confirmed_trip["data"]["trip_addition"] = []
+    if "status" not in ui.data or ui.data.status == ecwtui.InputStatus.ACTIVE:
+        confirmed_trip["data"]["trip_addition"].append(ui)
+    elif ui.data.status == ecwtui.InputStatus.DELETED:
+        after_del_list = [ta for ta in confirmed_trip["data"]["trip_addition"] if ta["match_id"] != ui["match_id"]]
+        confirmed_trip["data"]["trip_addition"] = after_del_list
+    else:
+        # TODO: Decide whether to error or to warn here
+        logging.warn("Invalid status found in user input %s, moving forward anyway" % ui)
 
 def create_confirmed_objects(user_id):
     time_query = epq.get_time_range_for_confirmed_object_creation(user_id)
@@ -82,6 +106,8 @@ def create_confirmed_trips(user_id, timerange):
         confirmed_trip_dict["data"]["expected_trip"] = tct.get_id()
         confirmed_trip_dict["data"]["user_input"] = \
             get_user_input_dict(ts, tct, input_key_list)
+        confirmed_trip_dict["data"]["trip_addition"] = \
+            esdt.get_additions_for_trip_object(ts, tct)
         confirmed_trip_entry = ecwe.Entry(confirmed_trip_dict)
         # save the entry
         ts.insert(confirmed_trip_entry)

emission/core/wrapper/confirmedtrip.py

@@ -23,7 +23,8 @@ class Confirmedtrip(ecwt.Trip):
                   "inferred_primary_mode": ecwb.WrapperBase.Access.WORM,
 # the user input will have all `manual/*` entries
 # let's make that be somewhat flexible instead of hardcoding into the data model
-                  "user_input": ecwb.WrapperBase.Access.WORM
+                  "user_input": ecwb.WrapperBase.Access.WORM,
+                  "trip_addition": ecwb.WrapperBase.Access.WORM
                   })
 
     def _populateDependencies(self):

emission/core/wrapper/entry.py

@@ -79,6 +79,11 @@ def _getData2Wrapper():
             "manual/replaced_mode": "userlabel",
             # user input for the trip; in one entry instead of being split up
             "manual/trip_user_input": "tripuserinput",
+            # trip-level additions/splits: currently only used for the time-use survey
+            "manual/trip_addition_input": "tripuserinput",
+            # place-level additions/splits: currently only used for the
+            # time-use survey, and potentially 
+            "manual/place_addition_input": "placeuserinput",
             # user confirmation of the destination (unsure how this will
             # interact with purpose
             "manual/destination_confirm": "userlabel",

emission/core/wrapper/placeuserinput.py

@@ -0,0 +1,31 @@
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from future import standard_library
+standard_library.install_aliases()
+from builtins import *
+import logging
+import emission.core.wrapper.wrapperbase as ecwb
+
+class Placeuserinput(ecwb.WrapperBase):
+    props = {"enter_ts": ecwb.WrapperBase.Access.RO,    # geojson representation of the point
+             "enter_local_dt": ecwb.WrapperBase.Access.RO, # start datetime in local time
+             "enter_fmt_time": ecwb.WrapperBase.Access.RO, # start formatted time (in timezone of point)
+             "exit_ts": ecwb.WrapperBase.Access.WORM, # end UTC timestamp (in secs)
+             "exit_local_dt": ecwb.WrapperBase.Access.RO, # end datetime in local time
+             "exit_fmt_time": ecwb.WrapperBase.Access.RO, # end formatted time (in timezone of point)
+             "label": ecwb.WrapperBase.Access.RO, # string summary of the survey
+             "version": ecwb.WrapperBase.Access.RO, # the survey version
+             "name": ecwb.WrapperBase.Access.RO, # the survey name
+             "xmlResponse": ecwb.WrapperBase.Access.RO, # the XML string representation of the survey response
+             "jsonDocResponse": ecwb.WrapperBase.Access.RO # the JSON representation of the survey response
+    }
+
+    enums = {}
+    geojson = []
+    nullable = []
+    local_dates = ["enter_local_dt", "exit_local_dt"]
+
+    def _populateDependencies(self):
+        pass

emission/core/wrapper/tripuserinput.py

@@ -6,23 +6,29 @@
 standard_library.install_aliases()
 from builtins import *
 import logging
+import enum
 import emission.core.wrapper.wrapperbase as ecwb
 
+class InputStatus(str, enum.Enum):
+    ACTIVE = "ACTIVE"
+    DELETED = "DELETED"
+
 class Tripuserinput(ecwb.WrapperBase):
     props = {"start_ts": ecwb.WrapperBase.Access.RO,    # geojson representation of the point
              "start_local_dt": ecwb.WrapperBase.Access.RO, # start datetime in local time
              "start_fmt_time": ecwb.WrapperBase.Access.RO, # start formatted time (in timezone of point)
              "end_ts": ecwb.WrapperBase.Access.WORM, # end UTC timestamp (in secs)
              "end_local_dt": ecwb.WrapperBase.Access.RO, # end datetime in local time
              "end_fmt_time": ecwb.WrapperBase.Access.RO, # end formatted time (in timezone of point)
+             "status": ecwb.WrapperBase.Access.RO, # ACTIVE or DELETED
              "label": ecwb.WrapperBase.Access.RO, # string summary of the survey
              "version": ecwb.WrapperBase.Access.RO, # the survey version
              "name": ecwb.WrapperBase.Access.RO, # the survey name
              "xmlResponse": ecwb.WrapperBase.Access.RO, # the XML string representation of the survey response
              "jsonDocResponse": ecwb.WrapperBase.Access.RO # the JSON representation of the survey response
     }
 
-    enums = {}
+    enums = {"status": InputStatus}
     geojson = []
     nullable = []
     local_dates = ["start_local_dt", "end_local_dt"]

emission/net/usercache/formatters/android/place_addition_input.py

@@ -0,0 +1,12 @@
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from future import standard_library
+standard_library.install_aliases()
+from builtins import *
+import logging
+import emission.net.usercache.formatters.generic.userlabel as fgl
+
+def format(entry):
+    return fgl.format(entry)

emission/net/usercache/formatters/android/trip_addition_input.py

@@ -0,0 +1,16 @@
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from future import standard_library
+standard_library.install_aliases()
+from builtins import *
+import logging
+import emission.net.usercache.formatters.generic.userlabel as fgl
+
+def format(entry):
+    import arrow
+
+    entry["data"]["start_ts"] = 0
+    entry["data"]["end_ts"] = arrow.get()
+    return fgl.format(entry)

emission/net/usercache/formatters/ios/place_addition_input.py

@@ -0,0 +1,12 @@
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from future import standard_library
+standard_library.install_aliases()
+from builtins import *
+import logging
+import emission.net.usercache.formatters.generic.userlabel as fgl
+
+def format(entry):
+    return fgl.format(entry)

emission/net/usercache/formatters/ios/trip_addition_input.py

@@ -0,0 +1,12 @@
+from __future__ import unicode_literals
+from __future__ import print_function
+from __future__ import division
+from __future__ import absolute_import
+from future import standard_library
+standard_library.install_aliases()
+from builtins import *
+import logging
+import emission.net.usercache.formatters.generic.userlabel as fgl
+
+def format(entry):
+    return fgl.format(entry)

emission/storage/decorations/trip_queries.py

@@ -15,6 +15,7 @@
 import emission.core.get_database as edb
 import emission.core.wrapper.rawtrip as ecwrt
 import emission.core.wrapper.entry as ecwe
+import emission.core.wrapper.tripuserinput as ecwtui
 
 import emission.storage.timeseries.abstract_timeseries as esta
 import emission.storage.timeseries.cache_series as estsc
@@ -153,6 +154,21 @@ def final_candidate(filter_fn, potential_candidates):
         (most_recent_entry.metadata.write_fmt_time, entry_detail(most_recent_entry)))
     return most_recent_entry
 
+def get_not_deleted_candidates(filter_fn, potential_candidates):
+    potential_candidate_objects = [ecwe.Entry(c) for c in potential_candidates]
+    extra_filtered_potential_candidates = list(filter(filter_fn, potential_candidate_objects))
+    if len(extra_filtered_potential_candidates) == 0:
+        logging.debug(f"in get_not_deleted_candidates, no candidates, returning []")
+        return []
+
+    # We want to retain all ACTIVE entries that have not been DELETED
+    all_active_list = [efpc for efpc in extra_filtered_potential_candidates if "status" not in efpc.data or efpc.data.status == ecwtui.InputStatus.ACTIVE]
+    all_deleted_id = [efpc["match_id"] for efpc in extra_filtered_potential_candidates if "status" in efpc.data and efpc.data.status == ecwtui.InputStatus.DELETED]
+    # TODO: Replace this with filter and a lambda if we decide not to match by ID after all
+    not_deleted_active = [efpc for efpc in all_active_list if efpc["match_id"] not in all_deleted_id]
+    logging.info(f"Found {len(all_active_list)} active entries, {len(all_deleted_id)} deleted entries -> {len(not_deleted_active)} non deleted active entries")
+    return not_deleted_active
+
 def get_user_input_for_trip_object(ts, trip_obj, user_input_key):
     tq = estt.TimeQuery("data.start_ts", trip_obj.data.start_ts, trip_obj.data.end_ts)
     potential_candidates = ts.find_entries([user_input_key], tq)
@@ -169,6 +185,11 @@ def get_user_input_from_cache_series(user_id, trip_obj, user_input_key):
     potential_candidates = estsc.find_entries(user_id, [user_input_key], tq)
     return final_candidate(valid_user_input(ts, trip_obj), potential_candidates)
 
+def get_additions_for_trip_object(ts, trip_obj):
+    tq = estt.TimeQuery("data.start_ts", trip_obj.data.start_ts, trip_obj.data.end_ts)
+    potential_candidates = ts.find_entries(["manual/trip_addition_input"], tq)
+    return get_not_deleted_candidates(valid_user_input(ts, trip_obj), potential_candidates)
+
 def valid_trip(ts, user_input):
     def curried(trip_obj):
         return valid_user_input_for_trip(ts, trip_obj, user_input)

emission/storage/timeseries/builtin_timeseries.py

@@ -62,7 +62,8 @@ def __init__(self, user_id):
                 "manual/purpose_confirm": self.timeseries_db,
                 "manual/replaced_mode": self.timeseries_db,
                 "manual/trip_user_input": self.timeseries_db,
-                "manual/destination_confirm": self.timeseries_db,
+                "manual/trip_addition_input": self.timeseries_db,
+                "manual/place_addition_input": self.timeseries_db,
                 "manual/demographic_survey": self.timeseries_db,
                 "segmentation/raw_trip": self.analysis_timeseries_db,
                 "segmentation/raw_place": self.analysis_timeseries_db,