 (defn parse-trips [file]
   (with-open [in (-> file clojure.java.io/input-stream java.util.zip.GZIPInputStream.)]
     (let [doc-id-fields [:pickup-dt :dropoff-dt :pickup-lat :pickup-lon :dropoff-lat :dropoff-lon :travel-km :paid-total]
-          dataset (->> file str (re-find #"([a-z]+)[^/]+$") second)
-          [header & rows] (-> in slurp csv/read-csv)
-          header   (map string/trim header)
-          missing  (clojure.set/difference (->> col-mapping keys (filter string?) set) (-> header set))
-          check    (if-not (empty? missing) (throw (Exception. (str "Missing cols for " missing " at " file))))
-          drop-fields identity ; #(reduce dissoc % [:pickup-lat :pickup-lon :dropoff-lat :dropoff-lon])
-          header   (map col-mapping header)
-          n-cols   (count header)
-          round    (fn [n] (let [scale 1000.0] (-> n (* scale) Math/round (/ scale))))
-          to-int-str (fn [v] (if v (let [v (int v)] (str (if (< v 10) "0" "") v))))
-          deg2km   (/ 40075.0 360.0)
-          mile2km  (fn [mile] (-> mile (* 1.60934) round))
-          delta-km (fn [from to] (if (and from to) (-> (- to from) (* deg2km) round)))
-          dist     (fn [a b] (if (and a b) (Math/sqrt (+ (* a a) (* b b))) 0.0))
+          company           (first (filter #(.contains (str file) %) ["green" "yellow" "uber"]))
+          vendor-lookup     (if (= company "green") {1 "Creative Mobile Technologies" 2 "VeriFone"})
+          dataset           (->> file str (re-find #"([a-z]+)[^/]+$") second)
+          [header & rows]   (-> in slurp csv/read-csv)
+          header            (map (comp string/lower-case string/trim) header)
+          missing           (clojure.set/difference (-> header set) (->> col-mapping keys (filter string?) set))
+          check             (if-not (empty? missing) (throw (Exception. (str "Missing cols for " missing " at " file))))
+          drop-extra-fields identity ; #(reduce dissoc % [:pickup-lat :pickup-lon :dropoff-lat :dropoff-lon])
+          header            (map col-mapping header)
+          n-cols            (count header)
+          coalesce-to-zero #(or % 0.0)
+          round            #(let [scale 1000.0] (-> % (* scale) Math/round (/ scale)))
+          deg2km            (/ 40075.0 360.0)
+          mile2km          #(-> % (* 1.60934) round)
+          delta-km          (fn [from to] (if (and from to) (-> (- to from) (* deg2km) round)))
+          dist              (fn [a b]     (if (and a b)     (Math/sqrt (+ (* a a) (* b b))) 0.0))
+          filter-km-deltas #(if (< (dist (:dlat-km %) (:dlon-km %)) 0.2) (reduce dissoc % [:dlat-km :dlon-km]) %)
+          assoc-speed-kmph #(assoc % :speed-kmph (if-let [travel-h (:travel-h %)] (if (> travel-h 0.17) (round (/ (:travel-km %) travel-h)))))
+          assoc-travel-h   #(assoc % :travel-h   (let [pickup-time  (:pickup-time  %)
+                                                       dropoff-time (:dropoff-time %)]
+                                                   (if (and dropoff-time pickup-time (not= dropoff-time pickup-time))
+                                                     (round (let[delta (- dropoff-time pickup-time)]
+                                                              (if (pos? delta) delta (+ delta 24)))))))
           to-time  (fn [datetime]
-                     (let [time (apply + (map / (map #(->> % (apply str) ((:int parsers))) (->> datetime (drop 9) (partition 2))) [1.0 60.0 3600.0]))]
+                     (let [time-scales [1.0 60.0 3600.0]
+                           time (apply + (map / (map #(->> % (apply str) ((:int parsers))) (->> datetime (drop 9) (partition 2))) time-scales))]
                        (if (not= time 0.0) (min (round time) 23.999))))
           docs     (for [row rows :when (>= (count row) n-cols)]
                      (let [doc (into (sorted-map) (map (fn [h v] (if h [(second h) ((-> h first parsers) v)])) header row))]

   :travel-h     [:float    :travel-h]
   :speed-kmph   [:float    :speed-kmph]})

(def fields
  (->> (concat
         (for [weather-field (-> weather vals first keys)] [weather-field :float])

  {:server      (or (System/getenv "ES_SERVER")      "192.168.0.100:9200")})

; curl -s 'http://localhost:9200/_template/*?pretty'
; curl -XDELETE 'http://localhost:9200/_template/taxicab' && curl -XDELETE 'http://localhost:9200/taxicab-green-*'
(defn create-template [client template-name body]
  (let [url (str "_template/" template-name)]
    (if (= 404 (:status (s/request client {:url url :method :head})))
