From 5caf4e48174533710243e8eeaf52567795ec9d01 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 7 May 2018 09:06:04 +0200 Subject: move files to backend subdir, push data in submodule --- api.js | 42 ------------- backend/api.js | 42 +++++++++++++ backend/config.js | 59 ++++++++++++++++++ backend/etl.js | 176 ++++++++++++++++++++++++++++++++++++++++++++++++++++++ backend/file.js | 39 ++++++++++++ backend/index.js | 27 +++++++++ backend/params.js | 39 ++++++++++++ backend/reduce.js | 51 ++++++++++++++++ backend/source.js | 39 ++++++++++++ backend/utils.js | 72 ++++++++++++++++++++++ config.js | 59 ------------------ data | 1 + data/.gitkeep | 0 data_/.gitkeep | 0 etl.js | 176 ------------------------------------------------------ file.js | 39 ------------ index.js | 27 --------- params.js | 39 ------------ reduce.js | 51 ---------------- source.js | 39 ------------ utils.js | 72 ---------------------- 21 files changed, 545 insertions(+), 544 deletions(-) delete mode 100644 api.js create mode 100644 backend/api.js create mode 100644 backend/config.js create mode 100644 backend/etl.js create mode 100644 backend/file.js create mode 100644 backend/index.js create mode 100644 backend/params.js create mode 100644 backend/reduce.js create mode 100644 backend/source.js create mode 100644 backend/utils.js delete mode 100644 config.js create mode 160000 data delete mode 100644 data/.gitkeep create mode 100644 data_/.gitkeep delete mode 100644 etl.js delete mode 100644 file.js delete mode 100644 index.js delete mode 100644 params.js delete mode 100644 reduce.js delete mode 100644 source.js delete mode 100644 utils.js diff --git a/api.js b/api.js deleted file mode 100644 index 30b4937..0000000 --- a/api.js +++ /dev/null @@ -1,42 +0,0 @@ -#!/usr/bin/node - -const requestP = require('request-promise'), - config = require('./config'), - Future = require('fluture'), - FutureRetry = require('fluture-retry'); - -const API_KEY = process.env.API_KEY || require('./secret').apikey; - -const requestF = Future.encaseP(requestP); - -function apiRequest(path, query) { - return FutureRetry.retryLinearly(requestF({ - uri: config.api.baseUrl + path, - headers: { - 'X-Title-Id': 'semc-vainglory', - 'Authorization': API_KEY - }, - qs: query, - json: true, - gzip: true, - forever: true, - timeout: config.api.timeout * 1000, - strictSSL: true - })); -} - -function awsRequest(url) { - return FutureRetry.retryLinearly(requestF({ - uri: url, - json: true, - gzip: true, - forever: true, - timeout: config.api.timeout * 1000, - strictSSL: true - })); -} - -module.exports = { - apiRequest, - awsRequest, -}; diff --git a/backend/api.js b/backend/api.js new file mode 100644 index 0000000..30b4937 --- /dev/null +++ b/backend/api.js @@ -0,0 +1,42 @@ +#!/usr/bin/node + +const requestP = require('request-promise'), + config = require('./config'), + Future = require('fluture'), + FutureRetry = require('fluture-retry'); + +const API_KEY = process.env.API_KEY || require('./secret').apikey; + +const requestF = Future.encaseP(requestP); + +function apiRequest(path, query) { + return FutureRetry.retryLinearly(requestF({ + uri: config.api.baseUrl + path, + headers: { + 'X-Title-Id': 'semc-vainglory', + 'Authorization': API_KEY + }, + qs: query, + json: true, + gzip: true, + forever: true, + timeout: config.api.timeout * 1000, + strictSSL: true + })); +} + +function awsRequest(url) { + return FutureRetry.retryLinearly(requestF({ + uri: url, + json: true, + gzip: true, + forever: true, + timeout: config.api.timeout * 1000, + strictSSL: true + })); +} + +module.exports = { + apiRequest, + awsRequest, +}; diff --git a/backend/config.js b/backend/config.js new file mode 100644 index 0000000..fe51ca3 --- /dev/null +++ b/backend/config.js @@ -0,0 +1,59 @@ +#!/usr/bin/node + +const crypto = require('crypto'); + +const sha256 = (x) => crypto.createHash('sha256').update(x, 'utf8').digest('hex'); + +const hashSettings = (config) => sha256( + config.api.regions + + config.api.modes + + config.api.requestsPerInterval + + config.api.interval + + config.api.intervalUnit +).substring(0, 8); + +module.exports = { + api: { + baseUrl: 'https://api.dc01.gamelockerapp.com', + timeout: 3, + /* available: eu, na, sg, cn, sa, ea */ + regions: ['sg', 'eu', 'na'], + /* available: casual, 5v5_pvp_casual, ranked, 5v5_pvp_ranked, blitz_pvp_ranked, casual_aral */ + modes: ['blitz_pvp_ranked', 'casual_aral'], + /* available: 2, 3, 4, 5 */ + matchesPerRequest: 5, + /* How many equidistant requests to create + * matchesPerRequest * requestsPerInterval = sample size for that interval + * currently requestsPerInterval <= API key rate limit but should work if it isn't significantly higher + * rough estimate of upper limits (na/eu/sg, depends on region and time of day): + * blitz: 240 + * battle royale: 30 + * 3v3 casual: 30 + * 3v3 ranked: 100 + * 5v5 casual: 30 + * 5v5 ranked: 40 + * if filtering for multiple game modes, use the sum of the upper limits */ + requestsPerInterval: 50, + /* the size of a batch */ + interval: 1, + intervalUnit: 'hours', + /* How many batches to request per minute + * requestsPerInterval * intervalsPer Minute <= API key rate limit */ + intervalsPerMinute: 1, + }, + etl: { + /* attributes to be extracted and renamed from main API data */ + participant: [ ['Winner', ['attributes', 'stats', 'winner']] ], + roster: [ ], + match: [ ['Mode', ['attributes', 'gameMode']] ], + }, + reduce: { + /* criteria or 'filters' */ + group: ['Actor', 'Talent', 'Mode'], + /* stats */ + sum: ['Level', 'Winner'], + }, + file: { + pattern: (moment) => `../data/${hashSettings(module.exports)}/${moment.toISOString()}.json`, + }, +}; diff --git a/backend/etl.js b/backend/etl.js new file mode 100644 index 0000000..7334b7d --- /dev/null +++ b/backend/etl.js @@ -0,0 +1,176 @@ +#!/usr/bin/node + +const R = require('ramda'), + Future = require('fluture'), + utils = require('./utils'), + api = require('./api'); + +// API data cleanup +const mapSideToTeam = (side) => side == 'left/blue' ? 'Left' : 'Right'; + +function loadFPayloads(config) { + const objOfPathsConfig = { + match: R.map(R.apply(utils.objOfPath), config.match), + roster: R.map(R.apply(utils.objOfPath), config.roster), + participant: R.map(R.apply(utils.objOfPath), config.participant), + }; // TODO Ramda-ify + + /* + * 1 Pull API JSON + * 2 Extract Participant, Roster, Match, Assets + * 3.1 Transform Participant, Roster, Match to flat structure + * 3.2 Pull and filter Assets for Events + * 3.2.1 Transform Events: Add Match meta data + * 4 Join Events to flat structure + */ + + const propIncludedFilterType = R.curry((type) => + R.compose(R.filter(R.propEq('type', type)), R.prop('included')) + ); + + const extractJsonToAssets = propIncludedFilterType('asset'); + const extractJsonToParticipants = propIncludedFilterType('participant'); + const extractJsonToRosters = propIncludedFilterType('roster'); + const extractJsonToMatches = R.prop('data'); + + const filterTypeEqTalentEquipped = R.pipe( + R.filter(R.propEq('type', 'TalentEquipped')), + R.map(R.prop('payload')) + ); + + const transformParticipantToPayload = R.converge(utils.mergeAllArgs, + R.concat(objOfPathsConfig.participant, [ + utils.objOfPath('Actor', ['attributes', 'actor']), + utils.objOfPath('_participant_id', ['id']), + ]) + ); + + const transformRosterToPayload = R.converge(utils.mergeAllArgs, + R.concat(objOfPathsConfig.roster, [ + R.pipe( + R.path(['attributes', 'stats', 'side']), + mapSideToTeam, + R.objOf('Team'), + ), + utils.objOfPath('_roster_id', ['id']), + R.pipe( + R.path(['relationships', 'participants', 'data']), + R.map(R.prop('id')), + R.objOf('_participant_ids') + ) + ]) + ); + + const transformMatchToPayload = R.converge(utils.mergeAllArgs, + R.concat(objOfPathsConfig.match, [ + utils.objOfPath('_match_id', ['id']), + utils.objOfPath('_asset_id', ['relationships', 'assets', 'data', 0, 'id']), + R.pipe( + R.path(['relationships', 'rosters', 'data']), + R.map(R.prop('id')), + R.objOf('_roster_ids') + ) + ]) + ); + + /* SQL style join using relationship info stored in _ attrs */ + // TODO refactor + const joinParticipantPayloadToRosterPayload = R.pipe( + R.xprod, + R.filter( + R.converge(R.contains, [ + R.path([0, '_participant_id']), + R.path([1, '_participant_ids']) + ]) + ), + R.map(R.mergeAll) + ); + + const joinParticipantRosterPayloadToMatchPayload = R.pipe( + R.xprod, + R.filter( + R.converge(R.contains, [ + R.path([0, '_roster_id']), + R.path([1, '_roster_ids']) + ]) + ), + R.map(R.mergeAll) + ); + + // join participant -> roster -> match + const joinedParticipantRosterPayloads = R.converge(joinParticipantPayloadToRosterPayload, [ + R.compose(R.map(transformParticipantToPayload), extractJsonToParticipants), + R.compose(R.map(transformRosterToPayload), extractJsonToRosters), + ]); + const joinedParticipantRosterMatchPayloads = R.converge(joinParticipantRosterPayloadToMatchPayload, [ + joinedParticipantRosterPayloads, + R.compose(R.map(transformMatchToPayload), extractJsonToMatches), + ]); + + // chain json -> constructed payload, + // wrap in future for compatibility with `loadFTelemetryPayloads` + const loadFParticipantRosterMatchPayloads = R.pipe( + joinedParticipantRosterMatchPayloads, + Future.of, + ); + + // join [telemetry, asset id] -> { telemetry, asset id } + const joinTelemetryPayloadsToId = R.pipe( + R.apply(R.xprod), + R.map(R.mergeAll), + ); + + // chain json -> api payload, + // add additional meta information to events + const loadFTelemetryPayloads = R.pipe( + extractJsonToAssets, + R.map(R.converge(R.pipe( + Future.both, + R.map(joinTelemetryPayloadsToId), + ), [ + R.pipe( + R.path(['attributes', 'URL']), + api.awsRequest, + R.map(filterTypeEqTalentEquipped), + ), + R.pipe( + utils.objOfPath('_asset_id', ['id']), + R.of, + Future.of, + ), + ])), + Future.parallel(Infinity), + R.map(R.unnest), + ); + + // given an array of telemetry payloads and an array of constructed payloads, + // return merged telemetry + constructed payloads for identical players & matches + // problem: some do not have 'NoTalent' (5v5, 3v3 standard)! -> need right join not inner join + const joinParticipantRosterMatchPayloadsToTelemetryPayloads = R.pipe( + R.apply(utils.naturalJoinRight), + R.map(R.mergeAll), + ); + + const loadFParticipantRosterMatchTelemetryPayloads = R.pipe( + api.apiRequest, + R.chain(R.converge(R.pipe( + Future.both, + R.map(R.pipe( + joinParticipantRosterMatchPayloadsToTelemetryPayloads, + R.map(R.pipe( + R.omit(['_asset_id', '_participant_id', '_participant_ids', '_roster_id', '_roster_ids', '_match_id', 'Team']), + R.set(R.lensProp('Count'), 1), + )), + )), + ), [ + loadFParticipantRosterMatchPayloads, + loadFTelemetryPayloads, + ])) + ); + + return loadFParticipantRosterMatchTelemetryPayloads; +} + +module.exports = { + loadFPayloads, +}; diff --git a/backend/file.js b/backend/file.js new file mode 100644 index 0000000..b750921 --- /dev/null +++ b/backend/file.js @@ -0,0 +1,39 @@ +#!/usr/bin/node + +const R = require('ramda'), + Future = require('fluture'), + fs = require('fs'), + fsPath = require('fs-path'); + +function loadFPayloads(config) { + return R.curry((file) => + Future.node((cb) => fs.readFile(file, 'utf8', cb)) + .chain(Future.encase(JSON.parse)) + ); +} + +function saveFPayloads(config) { + return R.curry((file, data) => + Future.node((cb) => fsPath.writeFile(file, JSON.stringify(data), cb)) + .map(() => data) + ); +} + +function saveFPayloadsTimestamped(config) { + return R.curry((timestamp, data) => + saveFPayloads(config)(config.pattern(timestamp), data) + ); +} + +function loadFPayloadsTimestamped(config) { + return R.curry((timestamp) => + loadFPayloads(config)(config.pattern(timestamp)) + ); +} + +module.exports = { + loadFPayloads, + loadFPayloadsTimestamped, + saveFPayloads, + saveFPayloadsTimestamped, +}; diff --git a/backend/index.js b/backend/index.js new file mode 100644 index 0000000..5da0f82 --- /dev/null +++ b/backend/index.js @@ -0,0 +1,27 @@ +#!/usr/bin/node + +const R = require('ramda'), + Future = require('fluture'), + moment = require('moment'), + reduce = require('./reduce'), + source = require('./source'), + config = require('./config'); + +const loadFTimestamped = source.loadFTimestamped(config); +const deriveStatistics = reduce.deriveStatistics(config.reduce); +const aggregatePayloads = reduce.aggregatePayloads(config.reduce); + +function main() { + const firstOfMay = moment('2018-05-01'); + + const later = R.curry((base, hs) => base.clone().add(hs, 'hours')); + const hours = R.range(0, 24 * 5); + const laterMoments = R.map(later(firstOfMay), hours); + + const futures = R.map(loadFTimestamped, laterMoments); + Future.parallel(1, futures) + .map(R.compose(deriveStatistics, aggregatePayloads, R.unnest)) + .fork(console.error, (d) => console.log(JSON.stringify(d, null, 2))); +} + +main(); diff --git a/backend/params.js b/backend/params.js new file mode 100644 index 0000000..5edf658 --- /dev/null +++ b/backend/params.js @@ -0,0 +1,39 @@ +#!/usr/bin/node + +const R = require('ramda'), + moment = require('moment'); + +function paramsForHourSample(config) { + const requestArgsForRegionStartEnd = R.curry((region, start, end) => [ + `/shards/${region}/matches`, { + 'sort': '-createdAt', + 'filter[gameMode]': R.join(',', config.modes), + 'filter[createdAt-start]': start.toISOString(), + 'filter[createdAt-end]': end.toISOString(), + 'page[limit]': config.matchesPerRequest, + 'page[offset]': '0', + } + ]); + + const intervalMinutes = moment.duration(config.interval, config.intervalUnit).as('minutes'); + const requestsPerMinutePerRegion = Math.floor(config.requestsPerInterval / config.regions.length); + const splitDurationMinutes = Math.floor(intervalMinutes / requestsPerMinutePerRegion); + const splitMoments = (hour) => R.map( + (offsetIndex) => [ + hour.clone().minutes(offsetIndex * splitDurationMinutes), + hour.clone().minutes((offsetIndex + 1) * splitDurationMinutes), + ], + R.range(0, requestsPerMinutePerRegion), + ); + + return R.pipe( + splitMoments, + R.xprod(config.regions), + R.map(R.unnest), + R.map(R.apply(requestArgsForRegionStartEnd)), + ); +} + +module.exports = { + paramsForHourSample, +}; diff --git a/backend/reduce.js b/backend/reduce.js new file mode 100644 index 0000000..3ac9da2 --- /dev/null +++ b/backend/reduce.js @@ -0,0 +1,51 @@ +#!/usr/bin/node + +const R = require('ramda'), + utils = require('./utils'); + +function aggregatePayloads(config) { + // Ramda group() only groups neighbors, so sort beforehand + const sorter = R.props(config.group); + const grouper = utils.eqPropses(config.group); + + // sum & count + const aggregator = R.pipe( + R.sortBy(sorter), + R.groupWith(grouper), + R.map( + R.converge(utils.mergeAllArgs, [ + R.pipe( + R.project(config.group), + R.prop(0), + ), + R.pipe( + R.project(R.concat(config.sum, R.of('Count'))), + utils.mergeAllWith(R.add), + utils.unlist, + ), + ]), + ), + ); + + return aggregator; +} + +function deriveStatistics(config) { + const sorter = R.compose(R.defaultTo(0), R.prop('Winner')); + + // divide sums by counts = averages + const transformer = R.map( + R.converge(utils.mergeAllArgs, [ + utils.projectProps(config.group), + utils.divPropsByProp('Count', config.sum), + utils.projectProps(['Count']), + ]) + ); + + return R.compose(transformer, R.sortBy(sorter)); +} + +module.exports = { + aggregatePayloads, + deriveStatistics, +}; diff --git a/backend/source.js b/backend/source.js new file mode 100644 index 0000000..0f4fd28 --- /dev/null +++ b/backend/source.js @@ -0,0 +1,39 @@ +#!/usr/bin/node + +const R = require('ramda'), + Future = require('fluture'), + etl = require('./etl'), + params = require('./params'), + reduce = require('./reduce'), + file = require('./file'); + +function doFetchProcessStoreHour(config) { + const intervalSleep = 60 / config.api.intervalsPerMinute * 1000; + + const paramsForHourSample = params.paramsForHourSample(config.api); + const loadFApi = etl.loadFPayloads(config.etl); + const saveFPayloads = file.saveFPayloadsTimestamped(config.file); + const aggregatePayloads = reduce.aggregatePayloads(config.reduce); + + const requestsForHour = (timestamp) => R.map(R.apply(loadFApi), paramsForHourSample(timestamp)); + + return (timestamp) => Future.parallel(1, requestsForHour(timestamp)) + .map(R.compose(aggregatePayloads, R.unnest)) + .chainRej((err) => { console.error(err); return Future.of({}); }) // fail silently + .chain(saveFPayloads(timestamp)); + //.chain(Future.after(intervalSleep)); + // TODO should sleep after request; throttling rpm should happen + // at a higher level though. CPU / IO / AWS requests stall + // as well +} + +function loadFTimestamped(config) { + // load from file or fall back to API and store + return (timestamp) => + file.loadFPayloads(config)(config.file.pattern(timestamp)) + .or(doFetchProcessStoreHour(config)(timestamp)); +}; + +module.exports = { + loadFTimestamped, +}; diff --git a/backend/utils.js b/backend/utils.js new file mode 100644 index 0000000..6cff356 --- /dev/null +++ b/backend/utils.js @@ -0,0 +1,72 @@ +#!/usr/bin/node + +const R = require('ramda'); + +const lengthEq = (n) => R.compose(R.equals(n), R.length); +const list = R.unapply(R.identity); +const unlist = R.apply(R.identity); +const mergeAllArgs = R.compose(R.mergeAll, list); +const objOfPath = R.curry((objOf, path) => R.compose(R.objOf(objOf), R.path(path))); +const andAll = R.all(R.equals(R.T())); +const andAllArgs = R.compose(andAll, list); + +const projectProps = R.curry( + (proj, d) => R.compose(R.prop(0), R.project(proj), R.of)(d) +); + +const joinRight = R.curry((mapper1, mapper2, t1, t2) => { + let indexed = R.indexBy(mapper1, t1); + return t2.map((t2row) => R.merge(t2row, indexed[mapper2(t2row)])); +}); + +const commonKeys = R.useWith(R.intersection, [ + R.keys, + R.keys, +]); + +const naturalJoinRight = R.curry((t1, t2) => joinRight( + R.props(commonKeys(t1[0], t2[0])), + R.props(commonKeys(t1[0], t2[0])), + t2, + t1, +)); + +const eqPropses = R.curry((props) => // TODO should create function with arity 3 but doesn't because it's nested + R.converge(andAllArgs, R.map(R.eqProps, props)) +); + +const mergeAllWith = (merger) => R.until(lengthEq(1), R.pipe( // TODO same issue + R.splitEvery(2), + R.map( + R.ifElse(lengthEq(2), // TODO refactor + R.apply(R.mergeWith(merger)), + R.apply(R.identity)) + ) +)); + +const divPropsByProp = R.curry((prop, props, data) => { // TODO refactor + return R.fromPairs( + R.map((aProp) => + R.contains(aProp, props) ? + [aProp, R.prop(aProp, data) / R.prop(prop, data)] + : [aProp, R.prop(aProp, data)], + R.keys(data), + ) + ); +}); + +module.exports = { + list, + unlist, + mergeAllArgs, + objOfPath, + andAll, + andAllArgs, + joinRight, + naturalJoinRight, + eqPropses, + lengthEq, + mergeAllWith, + projectProps, + divPropsByProp, +}; diff --git a/config.js b/config.js deleted file mode 100644 index 0a7e048..0000000 --- a/config.js +++ /dev/null @@ -1,59 +0,0 @@ -#!/usr/bin/node - -const crypto = require('crypto'); - -const sha256 = (x) => crypto.createHash('sha256').update(x, 'utf8').digest('hex'); - -const hashSettings = (config) => sha256( - config.api.regions + - config.api.modes + - config.api.requestsPerInterval + - config.api.interval + - config.api.intervalUnit -).substring(0, 8); - -module.exports = { - api: { - baseUrl: 'https://api.dc01.gamelockerapp.com', - timeout: 3, - /* available: eu, na, sg, cn, sa, ea */ - regions: ['sg', 'eu', 'na'], - /* available: casual, 5v5_pvp_casual, ranked, 5v5_pvp_ranked, blitz_pvp_ranked, casual_aral */ - modes: ['blitz_pvp_ranked', 'casual_aral'], - /* available: 2, 3, 4, 5 */ - matchesPerRequest: 5, - /* How many equidistant requests to create - * matchesPerRequest * requestsPerInterval = sample size for that interval - * currently requestsPerInterval <= API key rate limit but should work if it isn't significantly higher - * rough estimate of upper limits (na/eu/sg, depends on region and time of day): - * blitz: 240 - * battle royale: 30 - * 3v3 casual: 30 - * 3v3 ranked: 100 - * 5v5 casual: 30 - * 5v5 ranked: 40 - * if filtering for multiple game modes, use the sum of the upper limits */ - requestsPerInterval: 50, - /* the size of a batch */ - interval: 1, - intervalUnit: 'hours', - /* How many batches to request per minute - * requestsPerInterval * intervalsPer Minute <= API key rate limit */ - intervalsPerMinute: 1, - }, - etl: { - /* attributes to be extracted and renamed from main API data */ - participant: [ ['Winner', ['attributes', 'stats', 'winner']] ], - roster: [ ], - match: [ ['Mode', ['attributes', 'gameMode']] ], - }, - reduce: { - /* criteria or 'filters' */ - group: ['Actor', 'Talent', 'Mode'], - /* stats */ - sum: ['Level', 'Winner'], - }, - file: { - pattern: (moment) => `./data/${hashSettings(module.exports)}/${moment.toISOString()}.json`, - }, -}; diff --git a/data b/data new file mode 160000 index 0000000..638ae1e --- /dev/null +++ b/data @@ -0,0 +1 @@ +Subproject commit 638ae1e647bf8621da9f8ddd3d93cacb0788bb14 diff --git a/data/.gitkeep b/data/.gitkeep deleted file mode 100644 index e69de29..0000000 diff --git a/data_/.gitkeep b/data_/.gitkeep new file mode 100644 index 0000000..e69de29 diff --git a/etl.js b/etl.js deleted file mode 100644 index 7334b7d..0000000 --- a/etl.js +++ /dev/null @@ -1,176 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - Future = require('fluture'), - utils = require('./utils'), - api = require('./api'); - -// API data cleanup -const mapSideToTeam = (side) => side == 'left/blue' ? 'Left' : 'Right'; - -function loadFPayloads(config) { - const objOfPathsConfig = { - match: R.map(R.apply(utils.objOfPath), config.match), - roster: R.map(R.apply(utils.objOfPath), config.roster), - participant: R.map(R.apply(utils.objOfPath), config.participant), - }; // TODO Ramda-ify - - /* - * 1 Pull API JSON - * 2 Extract Participant, Roster, Match, Assets - * 3.1 Transform Participant, Roster, Match to flat structure - * 3.2 Pull and filter Assets for Events - * 3.2.1 Transform Events: Add Match meta data - * 4 Join Events to flat structure - */ - - const propIncludedFilterType = R.curry((type) => - R.compose(R.filter(R.propEq('type', type)), R.prop('included')) - ); - - const extractJsonToAssets = propIncludedFilterType('asset'); - const extractJsonToParticipants = propIncludedFilterType('participant'); - const extractJsonToRosters = propIncludedFilterType('roster'); - const extractJsonToMatches = R.prop('data'); - - const filterTypeEqTalentEquipped = R.pipe( - R.filter(R.propEq('type', 'TalentEquipped')), - R.map(R.prop('payload')) - ); - - const transformParticipantToPayload = R.converge(utils.mergeAllArgs, - R.concat(objOfPathsConfig.participant, [ - utils.objOfPath('Actor', ['attributes', 'actor']), - utils.objOfPath('_participant_id', ['id']), - ]) - ); - - const transformRosterToPayload = R.converge(utils.mergeAllArgs, - R.concat(objOfPathsConfig.roster, [ - R.pipe( - R.path(['attributes', 'stats', 'side']), - mapSideToTeam, - R.objOf('Team'), - ), - utils.objOfPath('_roster_id', ['id']), - R.pipe( - R.path(['relationships', 'participants', 'data']), - R.map(R.prop('id')), - R.objOf('_participant_ids') - ) - ]) - ); - - const transformMatchToPayload = R.converge(utils.mergeAllArgs, - R.concat(objOfPathsConfig.match, [ - utils.objOfPath('_match_id', ['id']), - utils.objOfPath('_asset_id', ['relationships', 'assets', 'data', 0, 'id']), - R.pipe( - R.path(['relationships', 'rosters', 'data']), - R.map(R.prop('id')), - R.objOf('_roster_ids') - ) - ]) - ); - - /* SQL style join using relationship info stored in _ attrs */ - // TODO refactor - const joinParticipantPayloadToRosterPayload = R.pipe( - R.xprod, - R.filter( - R.converge(R.contains, [ - R.path([0, '_participant_id']), - R.path([1, '_participant_ids']) - ]) - ), - R.map(R.mergeAll) - ); - - const joinParticipantRosterPayloadToMatchPayload = R.pipe( - R.xprod, - R.filter( - R.converge(R.contains, [ - R.path([0, '_roster_id']), - R.path([1, '_roster_ids']) - ]) - ), - R.map(R.mergeAll) - ); - - // join participant -> roster -> match - const joinedParticipantRosterPayloads = R.converge(joinParticipantPayloadToRosterPayload, [ - R.compose(R.map(transformParticipantToPayload), extractJsonToParticipants), - R.compose(R.map(transformRosterToPayload), extractJsonToRosters), - ]); - const joinedParticipantRosterMatchPayloads = R.converge(joinParticipantRosterPayloadToMatchPayload, [ - joinedParticipantRosterPayloads, - R.compose(R.map(transformMatchToPayload), extractJsonToMatches), - ]); - - // chain json -> constructed payload, - // wrap in future for compatibility with `loadFTelemetryPayloads` - const loadFParticipantRosterMatchPayloads = R.pipe( - joinedParticipantRosterMatchPayloads, - Future.of, - ); - - // join [telemetry, asset id] -> { telemetry, asset id } - const joinTelemetryPayloadsToId = R.pipe( - R.apply(R.xprod), - R.map(R.mergeAll), - ); - - // chain json -> api payload, - // add additional meta information to events - const loadFTelemetryPayloads = R.pipe( - extractJsonToAssets, - R.map(R.converge(R.pipe( - Future.both, - R.map(joinTelemetryPayloadsToId), - ), [ - R.pipe( - R.path(['attributes', 'URL']), - api.awsRequest, - R.map(filterTypeEqTalentEquipped), - ), - R.pipe( - utils.objOfPath('_asset_id', ['id']), - R.of, - Future.of, - ), - ])), - Future.parallel(Infinity), - R.map(R.unnest), - ); - - // given an array of telemetry payloads and an array of constructed payloads, - // return merged telemetry + constructed payloads for identical players & matches - // problem: some do not have 'NoTalent' (5v5, 3v3 standard)! -> need right join not inner join - const joinParticipantRosterMatchPayloadsToTelemetryPayloads = R.pipe( - R.apply(utils.naturalJoinRight), - R.map(R.mergeAll), - ); - - const loadFParticipantRosterMatchTelemetryPayloads = R.pipe( - api.apiRequest, - R.chain(R.converge(R.pipe( - Future.both, - R.map(R.pipe( - joinParticipantRosterMatchPayloadsToTelemetryPayloads, - R.map(R.pipe( - R.omit(['_asset_id', '_participant_id', '_participant_ids', '_roster_id', '_roster_ids', '_match_id', 'Team']), - R.set(R.lensProp('Count'), 1), - )), - )), - ), [ - loadFParticipantRosterMatchPayloads, - loadFTelemetryPayloads, - ])) - ); - - return loadFParticipantRosterMatchTelemetryPayloads; -} - -module.exports = { - loadFPayloads, -}; diff --git a/file.js b/file.js deleted file mode 100644 index b750921..0000000 --- a/file.js +++ /dev/null @@ -1,39 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - Future = require('fluture'), - fs = require('fs'), - fsPath = require('fs-path'); - -function loadFPayloads(config) { - return R.curry((file) => - Future.node((cb) => fs.readFile(file, 'utf8', cb)) - .chain(Future.encase(JSON.parse)) - ); -} - -function saveFPayloads(config) { - return R.curry((file, data) => - Future.node((cb) => fsPath.writeFile(file, JSON.stringify(data), cb)) - .map(() => data) - ); -} - -function saveFPayloadsTimestamped(config) { - return R.curry((timestamp, data) => - saveFPayloads(config)(config.pattern(timestamp), data) - ); -} - -function loadFPayloadsTimestamped(config) { - return R.curry((timestamp) => - loadFPayloads(config)(config.pattern(timestamp)) - ); -} - -module.exports = { - loadFPayloads, - loadFPayloadsTimestamped, - saveFPayloads, - saveFPayloadsTimestamped, -}; diff --git a/index.js b/index.js deleted file mode 100644 index 5da0f82..0000000 --- a/index.js +++ /dev/null @@ -1,27 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - Future = require('fluture'), - moment = require('moment'), - reduce = require('./reduce'), - source = require('./source'), - config = require('./config'); - -const loadFTimestamped = source.loadFTimestamped(config); -const deriveStatistics = reduce.deriveStatistics(config.reduce); -const aggregatePayloads = reduce.aggregatePayloads(config.reduce); - -function main() { - const firstOfMay = moment('2018-05-01'); - - const later = R.curry((base, hs) => base.clone().add(hs, 'hours')); - const hours = R.range(0, 24 * 5); - const laterMoments = R.map(later(firstOfMay), hours); - - const futures = R.map(loadFTimestamped, laterMoments); - Future.parallel(1, futures) - .map(R.compose(deriveStatistics, aggregatePayloads, R.unnest)) - .fork(console.error, (d) => console.log(JSON.stringify(d, null, 2))); -} - -main(); diff --git a/params.js b/params.js deleted file mode 100644 index 5edf658..0000000 --- a/params.js +++ /dev/null @@ -1,39 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - moment = require('moment'); - -function paramsForHourSample(config) { - const requestArgsForRegionStartEnd = R.curry((region, start, end) => [ - `/shards/${region}/matches`, { - 'sort': '-createdAt', - 'filter[gameMode]': R.join(',', config.modes), - 'filter[createdAt-start]': start.toISOString(), - 'filter[createdAt-end]': end.toISOString(), - 'page[limit]': config.matchesPerRequest, - 'page[offset]': '0', - } - ]); - - const intervalMinutes = moment.duration(config.interval, config.intervalUnit).as('minutes'); - const requestsPerMinutePerRegion = Math.floor(config.requestsPerInterval / config.regions.length); - const splitDurationMinutes = Math.floor(intervalMinutes / requestsPerMinutePerRegion); - const splitMoments = (hour) => R.map( - (offsetIndex) => [ - hour.clone().minutes(offsetIndex * splitDurationMinutes), - hour.clone().minutes((offsetIndex + 1) * splitDurationMinutes), - ], - R.range(0, requestsPerMinutePerRegion), - ); - - return R.pipe( - splitMoments, - R.xprod(config.regions), - R.map(R.unnest), - R.map(R.apply(requestArgsForRegionStartEnd)), - ); -} - -module.exports = { - paramsForHourSample, -}; diff --git a/reduce.js b/reduce.js deleted file mode 100644 index 3ac9da2..0000000 --- a/reduce.js +++ /dev/null @@ -1,51 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - utils = require('./utils'); - -function aggregatePayloads(config) { - // Ramda group() only groups neighbors, so sort beforehand - const sorter = R.props(config.group); - const grouper = utils.eqPropses(config.group); - - // sum & count - const aggregator = R.pipe( - R.sortBy(sorter), - R.groupWith(grouper), - R.map( - R.converge(utils.mergeAllArgs, [ - R.pipe( - R.project(config.group), - R.prop(0), - ), - R.pipe( - R.project(R.concat(config.sum, R.of('Count'))), - utils.mergeAllWith(R.add), - utils.unlist, - ), - ]), - ), - ); - - return aggregator; -} - -function deriveStatistics(config) { - const sorter = R.compose(R.defaultTo(0), R.prop('Winner')); - - // divide sums by counts = averages - const transformer = R.map( - R.converge(utils.mergeAllArgs, [ - utils.projectProps(config.group), - utils.divPropsByProp('Count', config.sum), - utils.projectProps(['Count']), - ]) - ); - - return R.compose(transformer, R.sortBy(sorter)); -} - -module.exports = { - aggregatePayloads, - deriveStatistics, -}; diff --git a/source.js b/source.js deleted file mode 100644 index 0f4fd28..0000000 --- a/source.js +++ /dev/null @@ -1,39 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'), - Future = require('fluture'), - etl = require('./etl'), - params = require('./params'), - reduce = require('./reduce'), - file = require('./file'); - -function doFetchProcessStoreHour(config) { - const intervalSleep = 60 / config.api.intervalsPerMinute * 1000; - - const paramsForHourSample = params.paramsForHourSample(config.api); - const loadFApi = etl.loadFPayloads(config.etl); - const saveFPayloads = file.saveFPayloadsTimestamped(config.file); - const aggregatePayloads = reduce.aggregatePayloads(config.reduce); - - const requestsForHour = (timestamp) => R.map(R.apply(loadFApi), paramsForHourSample(timestamp)); - - return (timestamp) => Future.parallel(1, requestsForHour(timestamp)) - .map(R.compose(aggregatePayloads, R.unnest)) - .chainRej((err) => { console.error(err); return Future.of({}); }) // fail silently - .chain(saveFPayloads(timestamp)); - //.chain(Future.after(intervalSleep)); - // TODO should sleep after request; throttling rpm should happen - // at a higher level though. CPU / IO / AWS requests stall - // as well -} - -function loadFTimestamped(config) { - // load from file or fall back to API and store - return (timestamp) => - file.loadFPayloads(config)(config.file.pattern(timestamp)) - .or(doFetchProcessStoreHour(config)(timestamp)); -}; - -module.exports = { - loadFTimestamped, -}; diff --git a/utils.js b/utils.js deleted file mode 100644 index 6cff356..0000000 --- a/utils.js +++ /dev/null @@ -1,72 +0,0 @@ -#!/usr/bin/node - -const R = require('ramda'); - -const lengthEq = (n) => R.compose(R.equals(n), R.length); -const list = R.unapply(R.identity); -const unlist = R.apply(R.identity); -const mergeAllArgs = R.compose(R.mergeAll, list); -const objOfPath = R.curry((objOf, path) => R.compose(R.objOf(objOf), R.path(path))); -const andAll = R.all(R.equals(R.T())); -const andAllArgs = R.compose(andAll, list); - -const projectProps = R.curry( - (proj, d) => R.compose(R.prop(0), R.project(proj), R.of)(d) -); - -const joinRight = R.curry((mapper1, mapper2, t1, t2) => { - let indexed = R.indexBy(mapper1, t1); - return t2.map((t2row) => R.merge(t2row, indexed[mapper2(t2row)])); -}); - -const commonKeys = R.useWith(R.intersection, [ - R.keys, - R.keys, -]); - -const naturalJoinRight = R.curry((t1, t2) => joinRight( - R.props(commonKeys(t1[0], t2[0])), - R.props(commonKeys(t1[0], t2[0])), - t2, - t1, -)); - -const eqPropses = R.curry((props) => // TODO should create function with arity 3 but doesn't because it's nested - R.converge(andAllArgs, R.map(R.eqProps, props)) -); - -const mergeAllWith = (merger) => R.until(lengthEq(1), R.pipe( // TODO same issue - R.splitEvery(2), - R.map( - R.ifElse(lengthEq(2), // TODO refactor - R.apply(R.mergeWith(merger)), - R.apply(R.identity)) - ) -)); - -const divPropsByProp = R.curry((prop, props, data) => { // TODO refactor - return R.fromPairs( - R.map((aProp) => - R.contains(aProp, props) ? - [aProp, R.prop(aProp, data) / R.prop(prop, data)] - : [aProp, R.prop(aProp, data)], - R.keys(data), - ) - ); -}); - -module.exports = { - list, - unlist, - mergeAllArgs, - objOfPath, - andAll, - andAllArgs, - joinRight, - naturalJoinRight, - eqPropses, - lengthEq, - mergeAllWith, - projectProps, - divPropsByProp, -}; -- cgit v1.3.1