   We break this apart from dynamic-filter to force the filtering to
   occur before the analysis. Note that all variable names within this
   query are TIMESERIES, not individual values."
-  [est-map clean-src rain-src]
+  [est-map clean-src]
   (let [long-block (:long-block est-map)
         short-block (:window est-map)]
-    (<- [?s-res ?mod-h ?mod-v ?sample ?line ?start ?short ?long ?t-stat ?break]
-        (rain-src ?s-res ?mod-h ?mod-v ?sample ?line ?start _ ?precl _)
+    (<- [?s-res ?mod-h ?mod-v ?sample ?line ?start ?end ?short ?long ?t-stat ?break]
+        (clean-src ?s-res ?mod-h ?mod-v ?sample ?line ?start ?ndvi ?precl)
         (f/shorten-ts ?ndvi ?precl :> ?short-precl)
-        (clean-src ?s-res ?mod-h ?mod-v ?sample ?line ?start ?ndvi)
         (a/short-stat long-block short-block ?ndvi :> ?short)
         (a/long-stats ?ndvi ?short-precl :> ?long ?t-stat)
-        (a/hansen-stat ?ndvi :> ?break))))
+        (a/hansen-stat ?ndvi :> ?break)
+        (series-end ?ndvi ?start :> ?end)
+        (:distinct false))))
+
+(defn count-series
+  [series]
+  (count (first series)))
+
+(defbufferop consolidator
+  [tuples]
+  [[[(vec (map first tuples))]
+    [(vec (map #(nth % 1) (sort-by first tuples)))]
+    [(vec (map #(nth % 2) (sort-by first tuples)))]
+    [(vec (map #(nth % 3) (sort-by first tuples)))]
+    [(vec (map #(nth % 4) (sort-by first tuples)))]]])
+
+(defn get-max-period
+  [series]
+  (reduce max (first series)))
+
+(defn consolidate-trends
+  [trends-src]
+  (<- [?s-res ?mh ?mv ?s ?l ?start ?end-seq ?short-ts ?long-ts ?t-stat-ts ?break-ts]
+      (trends-src ?s-res ?mh ?mv ?s ?l ?start ?end ?short ?long ?t-stat ?break)
+      (consolidator ?end ?short ?long ?t-stat ?break :> ?end-seq ?short-ts ?long-ts ?t-stat-ts ?break-ts)))
+
+(defn unnest-series
+  "Remove extraneous vectors from around a series"
+  [series]
+  (vec (flatten series)))
+
+(defn unnest-all
+  "Unnest series coming out of trends-cleanup of form
+   [[\"500\" 28 8 0 0 693 694 [[0.1 0.1]] [[0.2 0.2]] [[0.3 0.3]] [[0.4 0.4]]]]"
+  [& series]
+  (map unnest-series series))
+
+(defn trends-cleanup
+  [trends-src]
+  (let [consolidated-tap (consolidate-trends trends-src)]
+    (<- [?s-res ?mh ?mv ?s ?l ?start ?end ?short-v ?long-v ?t-stat-v ?break-v]
+        (consolidated-tap ?s-res ?mh ?mv ?s ?l ?start ?end-seq ?short ?long ?t-stat ?break)
+        (unnest-all ?short ?long ?t-stat ?break :> ?short-v ?long-v ?t-stat-v ?break-v)
+        (get-max-period ?end-seq :> ?end))))
 
 (defn forma-tap
-  "Accepts an est-map and sources for ndvi, rain, and fire timeseries,
-  plus a source of static vcf pixels.
-
-  Note that all values internally discuss timeseries."
-  [dynamic-src fire-src]
-  (<- [?s-res ?period ?mh ?mv ?s ?l ?forma-val]
-      (fire-src ?s-res ?mh ?mv ?s ?l !!fire)
-      (dynamic-src ?s-res ?mh ?mv ?s ?l ?start ?short ?break ?long ?t-stat)
-      (schema/forma-seq !!fire ?short ?break ?long ?t-stat :> ?forma-seq)
-      (p/index ?forma-seq :zero-index ?start :> ?period ?forma-val)))
+  "Accepts an est-map and sources for 
+
+  Note that all values internally discuss timeseries.
+
+  Also note that !!fire is an ungrounding variable, and triggers a
+  left join with the trend result variables."
+  [t-res est-map dynamic-src fire-src]
+  (let [start (date/datetime->period t-res (:est-start est-map))]
+    (<- [?s-res ?period ?mh ?mv ?s ?l ?forma-val]
+        (fire-src ?s-res ?mh ?mv ?s ?l !!fire)
+        (dynamic-src ?s-res ?mh ?mv ?s ?l _ ?end ?short ?long ?t-stat ?break)
+        (schema/forma-seq !!fire ?short ?long ?t-stat ?break :> ?forma-seq)
+        (p/index ?forma-seq :zero-index start :> ?period ?forma-val)
+        (:distinct false))))
 
 (defmapcatop [process-neighbors [num-neighbors]]
   "Processes all neighbors... Returns the index within the chunk, the
-value, and the aggregate of the neighbors."
+  value, and the aggregate of the neighbors."
   [window]
-  (for [[idx [val neighbors]] (->> (w/neighbor-scan num-neighbors window)
-                                   (map-indexed vector))
+  (for [[idx [val neighbors]]
+        (->> (w/neighbor-scan num-neighbors window)
+             (map-indexed vector))
         :when val]
     [idx val (->> neighbors
                   (apply concat)

(ns forma.hadoop.jobs.forma
  (:use cascalog.api)
  (:require [cascalog.ops :as c]
            [forma.matrix.walk :as w]
            [forma.reproject :as r]
            [forma.date-time :as date]
            [forma.schema :as schema]
            [forma.hadoop.predicate :as p]
            [forma.trends.analysis :as a]
            [forma.ops.classify :as log]
            [forma.trends.filter :as f]))

;; We're mapping across two sequences at the end, there; the
;; long-series and the t-stat-series.

(def get-loc
  (<- [?chunk :> ?s-res ?mod-h ?mod-v ?sample ?line ?val]
      (schema/extract-location ?chunk :> ?loc)
      (schema/extract-chunk-value ?chunk :> ?val)
      (schema/unpack-pixel-location ?loc :> ?s-res ?mod-h ?mod-v ?sample ?line)))

(defn fire-tap
  "Accepts an est-map and a query source of fire timeseries. Note that
  this won't work, pulling directly from the pail!"
  [est-map fire-src]
  (<- [?s-res ?mod-h ?mod-v ?sample ?line ?fire-series]
      (fire-src ?chunk)
      (get-loc ?chunk :> ?s-res ?mod-h ?mod-v ?sample ?line ?f-series)
      (schema/adjust-fires est-map ?f-series :> ?fire-series)
      (:distinct true)))

(defn filter-query
  "Note that the ?ts here is an ArrayList of one of the items inside
  of ArrayValue."
  [vcf-src vcf-limit chunk-src]
  (<- [?s-res ?mod-h ?mod-v ?sample ?line ?start ?vals]
      (chunk-src _ ?ts-chunk)
      (vcf-src _ ?vcf-chunk)
      (get-loc ?ts-chunk :> ?s-res ?mod-h ?mod-v ?sample ?line ?series)
      (schema/unpack-timeseries ?series :> ?start _ ?ts)
      (schema/get-vals-wrap ?ts :> ?vals)
      (p/blossom-chunk ?vcf-chunk :> ?s-res ?mod-h ?mod-v ?sample ?line ?vcf)
      (>= ?vcf vcf-limit)))

(defn dynamic-filter

        (process-neighbors [neighbors] ?window :> ?win-idx ?val ?neighbor-val)
        (r/tile-position cols rows ?win-col ?win-row ?win-idx :> ?sample ?line))))

(defn gen-static-sources
  [pail-path out-dir dataset-str]
  (let [out-path (str out-dir "/" dataset-str)
        src (split-chunk-tap pail-path [dataset-str])]
    (<- [?s-res ?h ?v ?s ?l ?val]
        (src _ ?x)
        (thrift/unpack ?x :> _ ?loc ?obj _ _)
        (thrift/unpack ?loc :> ?s-res ?h ?v ?s ?l)
        (thrift/get-field-value ?obj :> ?val))))

(defn read-static-sources
  [seq-dir name-vec]
  (let [gen-path (fn [name] (str seq-dir "/" name))
        static-paths (map gen-path name-vec)]
   (map hfs-seqfile static-paths)))

(defn beta-data-prep
  "for example, static-path can equal this on dan's local machine:
   \"/mnt/hgfs/Dropbox/local/static-out\""
  [{:keys [t-res est-start min-coast-dist]} dynamic-src static-src]
  (let [first-idx (date/datetime->period t-res est-start)]
    (<- [?s-res ?pd ?mod-h ?mod-v ?s ?l ?val ?neighbor-val ?eco ?hansen]
        (dynamic-src ?s-res ?pd ?mod-h ?mod-v ?s ?l ?val ?neighbor-val)
        (static-src ?s-res ?mod-h ?mod-v ?s ?l _ _ ?eco ?hansen ?coast-dist)
        (= ?pd first-idx)
        (>= ?coast-dist min-coast-dist)
        (:distinct false))))

(defn beta-gen
  "query to return the beta vector associated with each ecoregion"
  [{:keys [t-res est-start ridge-const convergence-thresh max-iterations min-coast-dist]} src]
  (let [first-idx (date/datetime->period t-res est-start)]
    (<- [?s-res ?eco ?beta]
        (src ?s-res ?pd ?mod-h ?mod-v ?s ?l ?val ?neighbor-val ?eco ?hansen)
        (log/logistic-beta-wrap [ridge-const convergence-thresh max-iterations]
                                ?hansen ?val ?neighbor-val :> ?beta)
        (:distinct false))))

(defmapop [apply-betas [betas]]
  [eco val neighbor-val]
