                      :> ?start-idx ?precl-ts ?ndvi-ts ?reli-ts)))
 
 (defmapcatop tele-clean
-  "Return clean timeseries with telescoping window, nil if no (or not enough) good training data"
-  [{:keys [est-start est-end t-res]}
-   good-set bad-set start-period val-ts reli-ts]
+  "Return clean timeseries with telescoping window, nil if no (or not
+  enough) good training data"
+  [{:keys [est-start est-end t-res]} good-set bad-set start-period val-ts reli-ts]
   (let [reli-thresh 0.1
         freq (date/res->period-count t-res)
         [start-idx end-idx] (date/relative-period t-res start-period

(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]
