<commit_before>(ns taxi-rides-clj.core
  (:require [clojure.data.json :as json]
            [clojure.data.csv :as csv]
            [clojure.string :as string]
            [clojure.java.io :as io]
            [qbits.spandex :as s]
            [clojure.core.async :as async]
            [java-time :as t])
  (:use clojure.set)
  (:import java.util.zip.GZIPInputStream)
  (:gen-class))

; https://gist.github.com/prasincs/827272
(defn get-hash [type data] (.digest (java.security.MessageDigest/getInstance type) (.getBytes data)))
(defn sha1-hash [data] (->> data (get-hash "sha1") (map #(.substring (Integer/toString (+ (bit-and % 0xff) 0x100) 16) 1)) (apply str)))

; http://yellerapp.com/posts/2014-12-11-14-race-condition-in-clojure-println.html
(defn my-println [& more]
  (do
    (.write *out* (str (clojure.string/join "" more) "\n"))
    (flush)))

; Based on standard pmap but with configurable n
(defn my-pmap [n f coll]
  (let [rets (map #(-> % f (try (catch Exception e (my-println "Exception: " e))) future) coll)
        step (fn step [[x & xs :as vs] fs]
               (lazy-seq
                 (if-let [s (seq fs)]
                   (cons (deref x) (step xs (rest s)))
                   (map deref vs))))]
     (step rets (drop n rets))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def data-folder "/home/wrecked/projects/taxi-rides/data")
(defn find-files [re] (->> data-folder io/file file-seq (filter #(re-find re (-> % .getName))) sort))

(defmacro make-parser [f]
  `(fn [v#] (if-not (empty? v#)
              (try (~f v#)
                   (catch Exception e#
                     (do (my-println "Invalid value '" v# "' for parser " (quote ~f))
                         (my-println "Exception: " e#)))))))

(let [int-parser      (make-parser Integer.)
      double-parser   (make-parser Double.)
      digits          (into #{\T} (for [i (range 10)] (first (str i))))
      assert-len     #(if (= (count %2) %) %2 (throw (Exception. (str "Expected length " % ", got '" %2 "'!"))))
      basic-parser   #(let [d (->> % (re-seq #"[0-9]+") (map int-parser) (apply t/local-date-time)
                                     str (filter digits) (apply str))
                            d (if (and (= (count d) 13) (= (nth d 8) \T))
                                (str d "00")
                                d)]
                        (str (assert-len 15 d) "Z"))]
  (def parsers
    {:int              int-parser
     :float            double-parser
     :latlon          #(if-not (= % "0") (double-parser %))
     :keyword         #(let [s (string/trim %)] (if-not (empty? s) s))
     :basic-datetime   (make-parser basic-parser)}))

(let [[header & rows] (-> (str data-folder "/central_park_weather.csv") slurp csv/read-csv)
      cols-to-keep #{"DATE" "PRCP" "SNWD" "SNOW" "TMAX" "TMIN" "AWND"}
      n-cols       (count header)
      avg-wind-fix (fn [row] (update row :weather-AWND #(if-not (neg? %) %)))
      rows (for [row rows :when (>= (count row) n-cols)]
             (into {} (map (fn [k v] (if (cols-to-keep k) [k v])) header row)))]
  (def weather
    (->> (for [row rows]
             [(row "DATE")
              (->>
                (-> row (dissoc "DATE") seq)
                (map (fn [[k v]] [(keyword (str "weather-" k)) ((:float  parsers) v)]))
                (into {})
                avg-wind-fix)])
         (into {}))))
;(->> weather seq (take 10))

(def col-mapping
  {"VendorID"                [:int             :vendor-id]
   "lpep_pickup_datetime"    [:basic-datetime  :pickup-dt]
   "Lpep_dropoff_datetime"   [:basic-datetime  :dropoff-dt]
   "Store_and_fwd_flag"      [:keyword         :store-flag]
   "RateCodeID"              [:int             :rate-code-id]
   "Pickup_longitude"        [:latlon          :pickup-lon      :skip-in-es]
   "Pickup_latitude"         [:latlon          :pickup-lat      :skip-in-es]
   "Dropoff_longitude"       [:latlon          :dropoff-lon     :skip-in-es]
   "Dropoff_latitude"        [:latlon          :dropoff-lat     :skip-in-es]
   "Passenger_count"         [:int             :n-passengers]
   "Trip_distance"           [:float           :travel-km]
   "Fare_amount"             [:float           :paid-fare]
   "MTA_tax"                 [:float           :paid-tax]
   "Tip_amount"              [:float           :paid-tip]
   "Tolls_amount"            [:float           :paid-tolls]
   "Ehail_fee"               [:float           :paid-ehail]
   "Total_amount"            [:float           :paid-total]
   "Payment_type"            [:keyword         :payment-type]
   "Trip_type"               [:keyword         :trip-type]
   "Extra"                   nil
   
   ; These are formed by mergin latitude an longitude columns, not part of CSV
   :pickup-pos   [:geopoint :pickup-pos]
   :dropoff-pos  [:geopoint :dropoff-pos]
   :dlat-km      [:float    :dlat-km]
   :dlon-km      [:float    :dlon-km]
   :pickup-time  [:float    :pickup-time]
   :dropoff-time [:float    :dropoff-time]
   :travel-h     [:float    :travel-h]
   :speed-kmph   [:float    :speed-kmph]})

(def fields
  (->> (concat
         (for [weather-field (-> weather vals first keys)] [weather-field :float])
         (for [[field-type field-name & flag] (vals col-mapping) :when (and field-name (not ((set flag) :skip-in-es)))] [field-name field-type]))
      (into {})))

(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))
          to-time  (fn [datetime]
                     (let [time (apply + (map / (map #(->> % (apply str) ((:int parsers))) (->> datetime (drop 9) (partition 2))) [1.0 60.0 3600.0]))]
                       (if (not= time 0.0) (round time))))
          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))]
                       (-> doc
                         (assoc :id      (->> doc-id-fields (map doc) (interleave (repeat "/")) (apply str) sha1-hash))
                         (assoc :index   (apply str "taxicab-" dataset "-" (take 4 (:pickup-dt doc))))
                         (into           (weather (-> doc :pickup-dt (subs 0 8)))))))]
        (for [doc docs :when (pos? (:travel-km doc))]
          (-> doc
              (update :travel-km   mile2km)
              (assoc :pickup-pos   (if (:pickup-lon  doc) (mapv doc [:pickup-lon  :pickup-lat])))
              (assoc :dropoff-pos  (if (:dropoff-lon doc) (mapv doc [:dropoff-lon :dropoff-lat])))
              (assoc :dlat-km      (delta-km (:pickup-lat doc) (:dropoff-lat doc)))
              (assoc :dlon-km      (delta-km (:pickup-lon doc) (:dropoff-lon doc)))
              (assoc :pickup-time  (to-time (:pickup-dt  doc)))
              (assoc :dropoff-time (to-time (:dropoff-dt doc)))
              (#(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))))))))
             ;(update :pickup-time  to-int-str)
             ;(update :dropoff-time to-int-str)
              (#(assoc % :speed-kmph (if-let [travel-h (:travel-h %)] (round (/ (:travel-km %) travel-h)))))
              (#(if (< (dist (:dlat-km %) (:dlon-km %)) 0.2) (reduce dissoc % [:dlat-km :dlon-km]) %))
              drop-fields)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def es-types
    {:float            {:type "float"}
     :latlon           {:type "float"}
     :double           {:type "double"}
     :int              {:type "integer"}
     :keyword          {:type "keyword"}
     :basic-date       {:type "date" :format "basic_date"}
     :basic-datetime   {:type "date" :format "basic_date_time_no_millis||year_month_day||epoch_millis"}
     :geopoint         {:type "geo_point"}})

(def taxi-mapping
  {:mappings {:ride  {:properties (->> (map vector (keys fields) (map es-types (vals fields))) (into {}))}}
   :settings {:number_of_shards 1 :number_of_replicas 0}})
;(pprint taxi-mapping)

(def es-config
  {:server      (or (System/getenv "ES_SERVER")      "192.168.0.100:9200")})
<commit_msg>Code tweaks<commit_after>(ns taxi-rides-clj.core
  (:require [clojure.data.json :as json]
            [clojure.data.csv :as csv]
            [clojure.string :as string]
            [clojure.java.io :as io]
            [qbits.spandex :as s]
            [clojure.core.async :as async]
            [java-time :as t])
  (:use clojure.set)
  (:import java.util.zip.GZIPInputStream)
  (:gen-class))

; https://gist.github.com/prasincs/827272
(defn get-hash [type data] (.digest (java.security.MessageDigest/getInstance type) (.getBytes data)))
(defn sha1-hash [data] (->> data (get-hash "sha1") (map #(.substring (Integer/toString (+ (bit-and % 0xff) 0x100) 16) 1)) (apply str)))

; http://yellerapp.com/posts/2014-12-11-14-race-condition-in-clojure-println.html
(defn my-println [& more]
  (do
    (.write *out* (str (clojure.string/join "" more) "\n"))
    (flush)))

; Based on standard pmap but with configurable n
(defn my-pmap [n f coll]
  (let [rets (map #(-> % f (try (catch Exception e (my-println "Exception: " e))) future) coll)
        step (fn step [[x & xs :as vs] fs]
               (lazy-seq
                 (if-let [s (seq fs)]
                   (cons (deref x) (step xs (rest s)))
                   (map deref vs))))]
     (step rets (drop n rets))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def data-folder "/home/wrecked/projects/taxi-rides/data")
(defn find-files [re] (->> data-folder io/file file-seq (filter #(re-find re (-> % .getName))) sort))

(defmacro make-parser [f]
  `(fn [v#] (if-not (empty? v#)
              (try (~f v#)
                   (catch Exception e#
                     (do (my-println "Invalid value '" v# "' for parser " (quote ~f))
                         (my-println "Exception: " e#)))))))

(let [int-parser      (make-parser Integer.)
      double-parser   (make-parser Double.)
      digits          (into #{\T} (for [i (range 10)] (first (str i))))
      assert-len     #(if (= (count %2) %) %2 (throw (Exception. (str "Expected length " % ", got '" %2 "'!"))))
      basic-parser   #(let [d (->> % (re-seq #"[0-9]+") (map int-parser) (apply t/local-date-time)
                                     str (filter digits) (apply str))
                            d (if (and (= (count d) 13) (= (nth d 8) \T))
                                (str d "00")
                                d)]
                        (str (assert-len 15 d) "Z"))]
  (def parsers
    {:int              int-parser
     :float            double-parser
     :latlon          #(if-not (= % "0") (double-parser %))
     :keyword         #(let [s (string/trim %)] (if-not (empty? s) s))
     :basic-datetime   (make-parser basic-parser)}))

(let [[header & rows] (-> (str data-folder "/central_park_weather.csv") slurp csv/read-csv)
      cols-to-keep #{"DATE" "PRCP" "SNWD" "SNOW" "TMAX" "TMIN" "AWND"}
      n-cols       (count header)
      avg-wind-fix (fn [row] (update row :weather-AWND #(if-not (neg? %) %)))
      rows (for [row rows :when (>= (count row) n-cols)]
             (into {} (map (fn [k v] (if (cols-to-keep k) [k v])) header row)))]
  (def weather
    (->> (for [row rows]
             [(row "DATE")
              (->>
                (-> row (dissoc "DATE") seq)
                (map (fn [[k v]] [(keyword (str "weather-" k)) ((:float  parsers) v)]))
                (into {})
                avg-wind-fix)])
         (into {}))))
;(->> weather seq (take 10))

(def col-mapping
  {"VendorID"                [:int             :vendor-id]
   "lpep_pickup_datetime"    [:basic-datetime  :pickup-dt]
   "Lpep_dropoff_datetime"   [:basic-datetime  :dropoff-dt]
   "Store_and_fwd_flag"      [:keyword         :store-flag]
   "RateCodeID"              [:int             :rate-code-id]
   "Pickup_longitude"        [:latlon          :pickup-lon]
   "Pickup_latitude"         [:latlon          :pickup-lat]
   "Dropoff_longitude"       [:latlon          :dropoff-lon]
   "Dropoff_latitude"        [:latlon          :dropoff-lat]
   "Passenger_count"         [:int             :n-passengers]
   "Trip_distance"           [:float           :travel-km]
   "Fare_amount"             [:float           :paid-fare]
   "MTA_tax"                 [:float           :paid-tax]
   "Tip_amount"              [:float           :paid-tip]
   "Tolls_amount"            [:float           :paid-tolls]
   "Ehail_fee"               [:float           :paid-ehail]
   "Total_amount"            [:float           :paid-total]
   "Payment_type"            [:keyword         :payment-type]
   "Trip_type"               [:keyword         :trip-type]
   "Extra"                   nil
   
   ; These are generated, not part of CSV
   :company      [:keyword  :company]
   :pickup-pos   [:geopoint :pickup-pos]
   :dropoff-pos  [:geopoint :dropoff-pos]
   :dlat-km      [:float    :dlat-km]
   :dlon-km      [:float    :dlon-km]
   :pickup-time  [:float    :pickup-time]
   :dropoff-time [:float    :dropoff-time]
   :travel-h     [:float    :travel-h]
   :speed-kmph   [:float    :speed-kmph]})

(def fields
  (->> (concat
         (for [weather-field (-> weather vals first keys)] [weather-field :float])
         (for [[field-type field-name & flag] (vals col-mapping) :when (and field-name (not ((set flag) :skip-in-es)))] [field-name field-type]))
      (into {})))

(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))
          to-time  (fn [datetime]
                     (let [time (apply + (map / (map #(->> % (apply str) ((:int parsers))) (->> datetime (drop 9) (partition 2))) [1.0 60.0 3600.0]))]
                       (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))]
                       (-> doc
                         (assoc :id      (->> doc-id-fields (map doc) (interleave (repeat "/")) (apply str) sha1-hash))
                         (assoc :index   (apply str "taxicab-" dataset "-" (take 4 (:pickup-dt doc))))
                         (into           (weather (-> doc :pickup-dt (subs 0 8)))))))]
        (for [doc docs :when (pos? (:travel-km doc))]
          (-> doc
              (assoc  :company       "Green")
              (update :travel-km     mile2km)
              (assoc  :pickup-pos    (if (:pickup-lon  doc) (mapv doc [:pickup-lon  :pickup-lat])))
              (assoc  :dropoff-pos   (if (:dropoff-lon doc) (mapv doc [:dropoff-lon :dropoff-lat])))
              (assoc  :dlat-km       (delta-km (:pickup-lat doc) (:dropoff-lat doc)))
              (assoc  :dlon-km       (delta-km (:pickup-lon doc) (:dropoff-lon doc)))
              (assoc  :pickup-time   (to-time (:pickup-dt  doc)))
              (assoc  :dropoff-time  (to-time (:dropoff-dt doc)))
              (update :paid-ehail   #(or % 0.0))
              (#(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))))))))
             ;(update :pickup-time  to-int-str)
             ;(update :dropoff-time to-int-str)
              (#(assoc % :speed-kmph (if-let [travel-h (:travel-h %)] (if (> travel-h 0.17) (round (/ (:travel-km %) travel-h))))))
              (#(if (< (dist (:dlat-km %) (:dlon-km %)) 0.2) (reduce dissoc % [:dlat-km :dlon-km]) %))
              drop-fields)))))
;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;;

(def es-types
    {:float            {:type "float"}
     :latlon           {:type "float"}
     :double           {:type "double"}
     :int              {:type "integer"}
     :keyword          {:type "keyword"}
     :basic-date       {:type "date" :format "basic_date"}
     :basic-datetime   {:type "date" :format "basic_date_time_no_millis||year_month_day||epoch_millis"}
     :geopoint         {:type "geo_point"}})

(def taxi-mapping
  {:mappings {:ride  {:properties (->> (map vector (keys fields) (map es-types (vals fields))) (into {}))}}
   :settings {:number_of_shards 1 :number_of_replicas 0}})
;(pprint taxi-mapping)

(def es-config
  {:server      (or (System/getenv "ES_SERVER")      "192.168.0.100:9200")})

{'fraction_non_alphanumeric': 0.19186226863819666, 'fraction_numerical': 0.01199462309998966, 'mean_word_length': 1.835385517443565, 'pattern_counts': {'":': 0, '<': 7, '<?xml version=': 0, '>': 90, 'https://': 2, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 22, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3619841', 'n_tokens_mistral': 6981, 'n_tokens_neox': 6449, 'n_words': 1642}