Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 | 210x 1x 1x 1x 1x 11x 11x 11x 11x 11x 11x 1x 1x 1x 1x 4x 2x 2x 2x 2x 2x 2x 2x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 11x 5x 5x 4x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 5x 1x 1x 1x 1x 1x 1x 1x 1x 1x 1x 8x 8x 8x 2x | // import logger from 'winston'
// import makeDebug from 'debug'
import { getItems, replaceItems, discard } from 'feathers-hooks-common'
import { ObjectID } from 'mongodb'
import _ from 'lodash'
// const debug = makeDebug('weacast:weacast-core')
const discardFeaturesField = discard('features')
export function marshallResultsQuery (hook) {
let query = hook.params.query
Eif (query) {
// Need to convert from client/server side types : string
Iif ((hook.service.app.db.adapter === 'mongodb') && (typeof query.probeId === 'string')) {
query.probeId = new ObjectID(query.probeId)
}
}
}
export async function aggregateResultsQuery (hook) {
let query = hook.params.query
Eif (query) {
// Perform aggregation
if (query.$aggregate) {
const collection = hook.service.Model
const ids = typeof query.$groupBy === 'string' // Group by matching ID(s), ie single ID or array of field to create a compound ID
? { [query.$groupBy.replace('properties.', '')]: '$' + query.$groupBy }
// Aggregated in an accumulator to avoid conflict with feature properties
: query.$groupBy.reduce((object, id) => Object.assign(object, { [id.replace('properties.', '')]: '$' + id }), {})
let groupBy = {
_id: ids,
forecastTime: { $push: '$forecastTime' }, // Keep track of all forecast times
runTime: { $push: '$runTime' }, // Keep track of all run times
geometry: { $last: '$geometry' }, // geometry is similar for all results, keep last
type: { $last: '$type' }, // type is similar for all results, keep last
properties: { $last: '$properties' } // properties are similar for all results, keep last
}
// The query contains the match stage except options relevent to the aggregation pipeline
let match = _.omit(query, ['$groupBy', '$aggregate'])
// Ensure we do not mix results with/without relevant element values
// by separately querying each element then merging
let aggregatedResults
await Promise.all(query.$aggregate.map(async element => {
let elementResults = await collection.aggregate([
// Find matching probre results only
{ $match: Object.assign({ ['properties.' + element]: { $exists: true } }, match) },
// Ensure they are ordered by increasing forecast time and most recent forecast first
{ $sort: Object.assign({ forecastTime: 1, runTime: -1 }, query.$sort) },
// Keep track of all element values
{ $group: Object.assign({ [element]: { $push: '$properties.' + element } }, groupBy) }
]).toArray()
// Rearrange data so that we get ordered arrays indexed by element
elementResults.forEach(result => {
result.forecastTime = { [element]: result.forecastTime }
result.runTime = { [element]: result.runTime }
// Set back the element values as properties because we aggregated in an accumulator
// to avoid conflict with probe properties
result.properties[element] = result[element]
// Delete accumulator
delete result[element]
})
// Now merge
if (!aggregatedResults) aggregatedResults = elementResults
else {
elementResults.forEach(result => {
let previousResult = aggregatedResults.find(aggregatedResult => {
const keys = _.keys(ids)
return (_.isEqual(_.pick(aggregatedResult, keys), _.pick(result, keys)))
})
// Merge with previous matching feature if any
Eif (previousResult) {
Object.assign(previousResult.forecastTime, result.forecastTime)
Object.assign(previousResult.runTime, result.runTime)
previousResult.properties[element] = result.properties[element]
} else {
aggregatedResults.push(result)
}
})
}
}))
delete query.$aggregate
// Set result to avoid service DB call
hook.result = aggregatedResults
}
}
return hook
}
export function checkProbingType (hook) {
let query = hook.params.query
// When performing on-demand probing nothing will be created in the DB
// Simply return the probe object to be used by hooks
if (!_.isNil(query) && !_.isNil(query.forecastTime)) {
hook.result = hook.data
}
// Otherwise let create the probe object
return hook
}
export async function performProbing (hook) {
let query = hook.params.query
let items = getItems(hook)
const isArray = Array.isArray(items)
items = (isArray ? items : [items])
let probePromises = []
items.forEach(item => {
probePromises.push(hook.service.probe(item, query))
})
await Promise.all(probePromises)
replaceItems(hook, isArray ? items : items[0])
return hook
}
export async function removeResults (hook) {
let resultService = hook.service.app.getService('probe-results')
let items = getItems(hook)
const isArray = Array.isArray(items)
items = (isArray ? items : [items])
let removePromises = []
items.forEach(item => {
// We have to remove listeners for results update first
hook.service.unregisterForecastUpdates(item)
// Then result objects
removePromises.push(resultService.remove(null, {
query: {
probeId: item._id
}
}))
})
await Promise.all(removePromises)
return hook
}
export function removeFeatures (hook) {
let params = hook.params
let query = params.query
// Only discard if not explicitely asked by $select or when performing
// on-demand probing (in this case the probing time is given)
if (_.isNil(query) || (!(!_.isNil(query.$select) && query.$select.includes('features')) && _.isNil(query.forecastTime))) {
discardFeaturesField(hook)
}
}
|