stream data aggregator

aggregator.coffee

require 'coffee-mate/global'

compareDictOn = (f) ->
	(da, db) ->
		[da, db] = [f(da), f(db)]
		for k, va of da
			vb = db[k]
			if (not vb?) or (va < vb)
				return -1
			else if (not va?) or (va > vb)
				return 1
		return 0

groupOn = (f) ->
	(items) ->
		k_memo = {}
		v_memo = {}
		foreach items, (it) ->
			key = f(it)
			k = (v for _, v of key).join(',')
			k_memo[k] ?= key
			v_memo[k] ?= []
			v_memo[k].push it
		return [seek(k_memo), v_memo]

###
aggregator :: {
	keys: (List -> Dict)
	folder: {init: Dict, fold: ((Dict, Item) -> Dict)}
	values: ((Dict, Dict) -> Dict)
} -> [Item] -> [{key, value}]
###

aggregator = ({keys, folder, values}) ->
	(items) ->
		[getkey, groups] = groupOn(keys) items
		states = {}
		results = []
		foreach enumerate(groups), ([k, xs]) ->
			key = getkey(k)
			state = deepcopy(folder.init)
			foreach xs, (x) ->
				state = folder.fold(state, x)
			value = values(state, key)
			results.push {key, value}
		#f = ({key}) -> (v for _, v of key).join(',')
		#return results.sort((a, b) -> u = f(a); v = f(b); (u > v) - (u < v))
		return results.sort compareDictOn(pluck 'key')

#### helpers

each = (d) ->
	(args...) ->
		r = {}
		for k, f of d
			r[k] = f(args...)
		return r

foldEach = (d) ->
	init = {}
	fold_ = {}
	for k, v of d
		init[k] = v.init
		fold_[k] = v.fold
	fold = (r, v) ->
		rr = {}
		for k, f of fold_
			rr[k] = f(r[k], v)
		return rr
	return {init, fold}

SUM = (f) -> init: 0, fold: (r, x) -> r + f(x)
COUNT = SUM (-> 1)


#### try

f = (i, j, k, l) ->
	{date: i, type: ['袜子', '鞋子'][j], amount: k, sale: k * (0.5 + l) * [10, 200][j]}

data = list(200) zipWith(f) ranged_random_gen(14, seed: 1), ranged_random_gen(2, seed: 2), ranged_random_gen(10, seed: 3), random_gen(seed: 4)

#log -> data

ag = aggregator
	keys: (v) ->
		type: v.type
		#week: v.date // 7
		#day: v.date % 7
	folder:
		init:
			totalSale: 0
			totalAmount: 0
			count: 0
		fold: ({totalSale, totalAmount, count}, v) ->
			totalSale: totalSale + v.sale
			totalAmount: totalAmount + v.amount
			count: count + 1
	values: ({totalSale, totalAmount, count}, {week, type}) ->
		totalSale: totalSale
		totalAmount: totalAmount
		meanPrice: totalSale / totalAmount

ag2 = aggregator
	keys: each
		type: pluck('type')
		#week: (v) -> v.date // 7
		#day: (v) -> v.date % 7
	folder: foldEach
		totalSale: SUM pluck('sale')
		totalAmount: SUM pluck('amount')
		count: COUNT
	values: ({totalSale, totalAmount, count}, {week, type}) ->
		totalSale: totalSale
		totalAmount: totalAmount
		meanPrice: totalSale / totalAmount
		#count: count

r = ag2 data
log json r, ' '.repeat(4)