         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))
+      (vec (map vector (f/tele-ts start-idx end-idx val-ts)))
+      ;;  (map clean-fn
+      ;;     (f/tele-ts start-idx end-idx val-ts)
+      ;;     (f/tele-ts start-idx end-idx reli-ts)
       [[nil]])))
+
+(defmapcatop telescope-ts
+  [{:keys [est-start est-end t-res]}
+   start-period val-ts]
+  (let [[start-idx end-idx] (date/relative-period t-res start-period
+                                                  [est-start est-end])]
+    (vec (map (comp vector vec) (f/tele-ts start-idx end-idx val-ts)))))
 
 (defn dynamic-clean
   "Accepts an est-map, and sources for ndvi and rain timeseries and

(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

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