Import labels from a table source into an image composite

metadata.json

{
  "name": "Import labels from table",
  "kind": "script",
  "description": "Add labels to an image composite, read from a table source",
  "source_code": "script.whizzml",
  "inputs":[
    {
      "name": "composite-id",
      "type": "source-id",
      "description": "The target image composite to extend"
    },
    {
      "name": "table-id",
      "type": "source-id",
      "description": "The table source with new label values"
    },
    {
      "name": "path-field",
      "type": "string",
      "default": "filename",
      "description": "The name of the field in the table that corresponds to image paths"
    },
    {
      "name": "labels",
      "type": "list",
      "default": [],
      "description": "The list of fields in the table to be added"
    },
    {
      "name": "batch-size",
      "type": "number",
      "default": 5,
      "description": "Update batch size (debugging, keep at default)"
    }],
  "outputs":[
    {
      "name": "output",
      "type": "source-id",
      "description": "The updated image composite (composite-id if it was open)"
    }]
}

script.whizzml

(define (rows-to-values rows names id-map)
  (iterate (res [] row rows)
    (let (id (id-map (head row) false))
      (if id
        (let (vs (map (lambda (v name)
                        ;; workaround for wintermute bug: numbers in value
                        ;; cause validation errors
                        {"field" name "value" (if (number? v) (str v) v)
                         "components" [id]})
                      (tail row)
                      names))
          (concat res vs))
        res))))

(define (update-row-values src vs)
  (when (not (empty? vs))
    (try (update-and-wait (wait src) {"row_values" (take batch-size vs)})
         (update-row-values src (drop batch-size vs))
         (catch e
           (log-error "Error updating with ")
           (for (v (take batch-size vs)) (log-error v))
           (log-error e)
           (raise e)))))

(define (fetch-sample-rows sample offset no)
  ((fetch sample {"mode" "linear" "row_offset" offset "rows" no})
   ["sample" "rows"] []))

(define (add-values offset max-rows sample src names img-map)
  (log-progress (+ 0.3 (* 0.65 (- 1 (/ (- max-rows offset) max-rows)))))
  (when (< offset max-rows)
    (let (rows (fetch-sample-rows sample offset 100))
      (when (not (empty? rows))
        (log-info "Importing label rows [" offset " " (+ offset (count rows)) ")")
        (update-row-values src (rows-to-values rows names img-map))
        (add-values (+ offset (count rows)) max-rows sample src names img-map)))))

(define (find-optype fds type)
  (when (not (empty? fds))
    (if (= type ((head fds) "optype"))
      ((head fds) "name")
      (find-optype (tail fds) type))))

(define (image-path-map src)
  (let (src-fields (values (resource-fields src))
        path-field (find-optype src-fields "path")
        img-field (find-optype src-fields "image")
        ds (create-dataset src {"input_fields" [path-field img-field] "temp" true})
        rows (resource-property (wait ds) "rows")
        sds (wait (create-sample {"dataset" ds "temp" true})))
    (loop (res {} offset 0)
      (if (>= offset rows)
        res
        (let (res (iterate (res res kv (fetch-sample-rows sds offset 100))
                    (assoc res (last kv) (head kv))))
          (recur res (+ offset 100)))))))

(define (field-names fds)
  (map (lambda (f) (f "name")) (if (map? fds) (values fds) fds)))

(define (maybe-add-fields composite table path labels)
  (let (composite-fields (resource-fields composite)
        table-fields (resource-fields table)
        labels (if (empty? labels)
                 (field-names table-fields)
                 (filter (lambda (l) (find-field table-fields l)) labels))
        labels (list* (remove path (set* labels)))
        new-fields (filter (lambda (l) (not (find-field composite-fields l))) labels))
    (when (not (find-field table-fields path))
      (raise "Path field not found in table source"))
    (when (empty? labels)
      (raise "None of the requested labels is present in table source"))
    (when (not (empty? new-fields))
      (log-info "Adding new fields " new-fields " to composite")
      (let (new-fields (map (lambda (n)
                              (select-keys (find-field table-fields n)
                                           ["name" "optype"]))
                            new-fields))
        (update-and-wait composite {"new_fields" new-fields})))
    labels))

(define (add-labels composite table path labels)
  (log-info "Preparing input data...")
  (let ([c cc] (if (resource-property composite "closed")
                 [(create-source {"origin" composite}) composite]
                 [composite (create-source {"origin" composite "temp" true})])
        _ (log-progress 0.1)
        _ (log-info "Retrieving image composite information...")
        labels (maybe-add-fields (wait c) table path labels)
        _ (log-info "Collecting image paths...")
        id-map (image-path-map (wait cc))
        _ (log-progress 0.2)
        _ (log-info "Preparing input data for " (count id-map) " images")
        ds (create-dataset table {"temp" true})
        sample (create-sample {"dataset" (wait ds) "temp" true
                               "input_fields" (cons path labels)})
        rows (resource-property ds "rows"))
    (log-info "Importing data from " rows " table rows...")
    (log-progress 0.25)
    (add-values 0 rows (wait sample) c labels id-map)
    c))

(define output (add-labels composite-id table-id path-field labels))