 (ns forma.hadoop.jobs.forma
-  (:use cascalog.api)
+  (:use cascalog.api
+        [forma.hadoop.pail :only (split-chunk-tap)])
   (:require [cascalog.ops :as c]
             [forma.matrix.walk :as w]
             [forma.reproject :as r]
             [forma.date-time :as date]
             [forma.schema :as schema]
+            [forma.thrift :as thrift]
             [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 consolidate-static
+  "Due to an issue with Pail, we consolidate separate sequence files
+   of static data with one big join"
+  [vcf-limit vcf-src gadm-src hansen-src ecoid-src border-src]
+  (<- [?s-res ?mod-h ?mod-v ?sample ?line ?vcf ?gadm ?ecoid ?hansen ?coast-dist]
+      (vcf-src    ?s-res ?mod-h ?mod-v ?sample ?line ?vcf)
+      (hansen-src ?s-res ?mod-h ?mod-v ?sample ?line ?hansen)
+      (ecoid-src  ?s-res ?mod-h ?mod-v ?sample ?line ?ecoid)
+      (gadm-src   ?s-res ?mod-h ?mod-v ?sample ?line ?gadm)
+      (border-src ?s-res ?mod-h ?mod-v ?sample ?line ?coast-dist)
+      (>= ?vcf vcf-limit)))
+
+(defn within-tileset?
+  [tile-set h v]
+  (let [tile [h v]]
+    (contains? tile-set tile)))
+
+(defn screen-by-tileset
+  [src tile-set]
+  (<- [?pail-path ?pixel-chunk]
+      (src ?pail-path ?pixel-chunk)
+      (thrift/unpack ?pixel-chunk :> _ ?pixel-loc _ _ _)
+      (thrift/unpack ?pixel-loc :> _ ?h ?v _ _)
+      (within-tileset? tile-set ?h ?v)))
 
 (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)))
+  (<- [?s-res ?h ?v ?sample ?line ?adjusted-ts]
+      (fire-src ?pixel-chunk)
+      (thrift/unpack ?pixel-chunk :> _ ?pixel-loc ?ts _ _)
+      (thrift/unpack ?pixel-loc :> ?s-res ?h ?v ?sample ?line)
+      (schema/adjust-fires est-map ?ts :> ?adjusted-ts)))
+
+(defn ts-unpacking
+  [ts-src]
+  (<- [?s-res ?mod-h ?mod-v ?sample ?line ?start-idx ?series]
+      (ts-src _ ?ts-chunk)
+      ;; unpack ts object
+      (thrift/unpack ?ts-chunk :> _ ?ts-loc ?ts-data _ _)
+      (thrift/unpack ?ts-data :> ?start-idx _ ?ts-array)
+      (thrift/unpack* ?ts-array :> ?series)
+      (thrift/unpack ?ts-loc :> ?s-res ?mod-h ?mod-v ?sample ?line)))
 
 (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]
+  (<- [?s-res ?mod-h ?mod-v ?sample ?line ?start-idx ?series]
       (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-src ?s-res ?mod-h ?mod-v ?sample ?line ?vcf)
+      
+      ;; unpack ts object
+       (thrift/unpack ?ts-chunk :> _ ?ts-loc ?ts-data _ _)
+       (thrift/unpack ?ts-data :> ?start-idx _ ?ts-array)
+       (thrift/unpack* ?ts-array :> ?series)
+       (thrift/unpack ?ts-loc :> ?s-res ?mod-h ?mod-v ?sample ?line)
+
+      ;; filter on vcf-limit
       (>= ?vcf vcf-limit)))
 
 (defn dynamic-filter

        training-reli-set (set training-reli)
        clean-fn (comp vector (partial f/make-clean freq good-set bad-set))]
    (if (f/reliable? good-set reli-thresh training-reli)
      (map clean-fn
           (f/tele-ts start-idx end-idx val-ts)
           (f/tele-ts start-idx end-idx reli-ts))
      [[nil]])))

(defn dynamic-clean
  "Accepts an est-map, and sources for ndvi and rain timeseries and

(defn forma-estimate
  "query to end all queries: estimate the probabilities for each
  period after the training period."
  [beta-src dynamic-src static-src]
  (let [betas (log/beta-dict beta-src)]
    (<- [?s-res ?mod-h ?mod-v ?s ?l ?prob-series]
        (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 _ _)
        (apply-betas [betas] ?eco ?val ?neighbor-val :> ?prob)
        (log/mk-timeseries ?pd ?prob :> ?prob-series)
        (:distinct false))))
