FrostClient.js

"use strict";
var __awaiter = (this && this.__awaiter) || function (thisArg, _arguments, P, generator) {
    function adopt(value) { return value instanceof P ? value : new P(function (resolve) { resolve(value); }); }
    return new (P || (P = Promise))(function (resolve, reject) {
        function fulfilled(value) { try { step(generator.next(value)); } catch (e) { reject(e); } }
        function rejected(value) { try { step(generator["throw"](value)); } catch (e) { reject(e); } }
        function step(result) { result.done ? resolve(result.value) : adopt(result.value).then(fulfilled, rejected); }
        step((generator = generator.apply(thisArg, _arguments || [])).next());
    });
};
var __importDefault = (this && this.__importDefault) || function (mod) {
    return (mod && mod.__esModule) ? mod : { "default": mod };
};
Object.defineProperty(exports, "__esModule", { value: true });
exports.FrostClient = void 0;
const node_fetch_1 = __importDefault(require("node-fetch"));
const logger_1 = require("../logger/logger");
const THINGS_PATH = "Things";
const LOCATIONS_PATH = "Locations";
const FEATURES_OF_INTEREST_PATH = "FeaturesOfInterest";
const DATASTREAMS_PATH = "Datastreams";
const OBSERVATIONS_PATH = "Observations";
const SENSORS_PATH = "Sensors";
const OBSERVED_PROPERTIES_PATH = "ObservedProperties";
const getPathFromType = (type) => {
    switch (type) {
        case "thing":
            return THINGS_PATH;
        case "location":
            return LOCATIONS_PATH;
        case "featureOfInterest":
            return FEATURES_OF_INTEREST_PATH;
        case "datastream":
            return DATASTREAMS_PATH;
        case "observation":
            return OBSERVATIONS_PATH;
        case "sensor":
            return SENSORS_PATH;
        case "observedProperty":
            return OBSERVED_PROPERTIES_PATH;
    }
    throw new Error(`Unknown type ${type}!`);
};
const DEFAULT_PROJECTION = ["@iot.id"];
class FrostClient {
    constructor(baseUrl) {
        this.baseUrl = baseUrl;
        // Initialize Here sensor and Speed observed property if they are not already inserted
        this.insertItemIfNotExists("sensor", {
            "@iot.id": "here",
            name: "Here",
            description: "Here - Traffic Flow API",
            encodingType: "text/html",
            metadata: "https://developer.here.com/documentation/traffic/dev_guide/topics_v6.1/example-flow.html"
        })
            .then(inserted => {
            if (inserted)
                console.log("Here sensor was inserted.");
        })
            .catch(err => console.error(err));
        this.insertItemIfNotExists("observedProperty", {
            "@iot.id": "speed",
            name: "Speed",
            definition: "http://www.qudt.org/qudt/owl/1.0.0/quantity/Instances.html#Speed",
            description: "The Speed."
        })
            .then(inserted => {
            if (inserted)
                console.log("Speed observed property was inserted.");
        })
            .catch(err => console.error(err));
    }
    /**
     * Returns the first thing with the specified id, or null if not found.
     * @param id
     * @param projection
     */
    getSingleThingById(id, projection) {
        return __awaiter(this, void 0, void 0, function* () {
            return this.getSingleItemById("thing", id, projection);
        });
    }
    /**
     * Returns the first location with the specified id, or null if not found.
     * @param id
     * @param projection
     */
    getSingleLocationById(id, projection) {
        return __awaiter(this, void 0, void 0, function* () {
            return this.getSingleItemById("location", id, projection);
        });
    }
    /**
     * Returns the first feature of interest with the specified id, or null if not found.
     * @param id
     * @param projection
     */
    getSingleFeatureOfInterestById(id, projection) {
        return __awaiter(this, void 0, void 0, function* () {
            return this.getSingleItemById("featureOfInterest", id, projection);
        });
    }
    getSingleDatastreamById(id, projection) {
        return __awaiter(this, void 0, void 0, function* () {
            return this.getSingleItemById("datastream", id, projection);
        });
    }
    /**
     * Returns the number of inserted observations
     * @param hereFlow
     * @param forceUpdateLocations
     */
    mapAndInsertHereFlowResponse(hereFlow, forceUpdateLocations = false) {
        return __awaiter(this, void 0, void 0, function* () {
            const data = hereFlow.RWS
                .flatMap(rws => rws.RW)
                .flatMap(rw => {
                const timeStamp = rw.PBT;
                const locationsAndResults = rw.FIS
                    .flatMap(fis => fis.FI)
                    .map(fi => {
                    var _a, _b, _c;
                    const thingId = `${rw.LI}-${fi.TMC.DE}-${fi.TMC.PC}`;
                    const thingDescription = `${rw.DE} direction to ${fi.TMC.DE}`;
                    const thing = {
                        "@iot.id": thingId,
                        name: thingId,
                        description: thingDescription
                    };
                    const location = {
                        "@iot.id": thingId,
                        name: thingId,
                        description: thingDescription,
                        encodingType: "application/vnd.geo+json",
                        location: {
                            type: "MultiLineString",
                            coordinates: fi.SHP
                                .map(v => v.value.toString())
                                .map(str => str
                                .trim()
                                .split(" ")
                                .map(coord => coord.split(",").reverse().map(x => +x)))
                        },
                        properties: {
                            length: fi.TMC.LE //In Kilometers
                        },
                        Things: [{
                                "@iot.id": thingId
                            }],
                    };
                    const featureOfInterest = {
                        "@iot.id": location["@iot.id"],
                        name: location.name,
                        description: location.description,
                        encodingType: location.encodingType,
                        feature: location.location,
                        properties: location.properties,
                    };
                    const speed = (_a = fi.CF[0]) === null || _a === void 0 ? void 0 : _a.SP;
                    const confidence = (_b = fi.CF[0]) === null || _b === void 0 ? void 0 : _b.CN;
                    const jamFactor = (_c = fi.CF[0]) === null || _c === void 0 ? void 0 : _c.JF;
                    return { thing, location, featureOfInterest, timeStamp, speed, confidence, jamFactor };
                })
                    .filter(data => !!data.speed); // if failed to obtain speed data, don't consider it
                /***** Log if ID repetitions Found *****/
                Object.entries(locationsAndResults.map(data => data.featureOfInterest["@iot.id"])
                    .reduce(function (prev, cur) {
                    prev[cur] = (prev[cur] || 0) + 1;
                    return prev;
                }, {}))
                    .forEach(([foiId, repetitions]) => {
                    if (repetitions > 1) {
                        console.error(`FeaturesOfInterest: ${foiId} is repeated ${repetitions} times`);
                    }
                });
                /***** *************************** *****/
                return locationsAndResults;
            });
            return Promise.all(data.map(roadwayData => {
                const datastreamId = `here-${roadwayData.thing["@iot.id"]}`;
                return this.insertItemIfNotExists("thing", roadwayData.thing)
                    .then(() => {
                    const datastream = {
                        "@iot.id": datastreamId,
                        name: datastreamId,
                        description: `Speed readings from Here for roadway ${roadwayData.thing.description}`,
                        observationType: "http://www.opengis.net/def/observationType/OGC-OM/2.0/OM_Measurement",
                        unitOfMeasurement: {
                            name: "Kilometers per hour",
                            symbol: "km/h",
                            definition: "http://www.qudt.org/qudt/owl/1.0.0/unit/Instances.html#KilometerPerHour"
                        },
                        ObservedProperty: {
                            "@iot.id": "speed"
                        },
                        Sensor: {
                            "@iot.id": "here"
                        },
                        Thing: {
                            "@iot.id": roadwayData.thing["@iot.id"]
                        },
                    };
                    return this.insertItemIfNotExists("datastream", datastream);
                })
                    .then(() => this.insertItemIfNotExists("location", roadwayData.location, forceUpdateLocations))
                    .then(() => this.insertItemIfNotExists("featureOfInterest", roadwayData.featureOfInterest, forceUpdateLocations))
                    .then(() => {
                    const observation = {
                        phenomenonTime: roadwayData.timeStamp,
                        result: roadwayData.speed,
                        resultQuality: roadwayData.confidence,
                        resultTime: roadwayData.timeStamp,
                        FeatureOfInterest: {
                            "@iot.id": roadwayData.featureOfInterest["@iot.id"]
                        },
                        Datastream: {
                            "@iot.id": datastreamId
                        },
                        parameters: {
                            jamFactor: roadwayData.jamFactor
                        }
                    };
                    return this.insertItem("observation", observation);
                })
                    .then(() => 1);
            })).then(length => length.reduce((a, b) => a + b, 0));
        });
    }
    insertItemIfNotExists(type, frostItemWithId, forceUpdate = false) {
        return __awaiter(this, void 0, void 0, function* () {
            const existingThing = yield this.getSingleItemById(type, frostItemWithId["@iot.id"]);
            if (existingThing && forceUpdate) {
                logger_1.info(`Force updating item ${frostItemWithId["@iot.id"]} with type ${type}`);
                return this.updateItem(type, frostItemWithId);
            }
            if (existingThing)
                return false;
            return this.insertItem(type, frostItemWithId);
        });
    }
    insertItem(type, frostItem) {
        return __awaiter(this, void 0, void 0, function* () {
            // if (type == "location")
            //     console.debug(`inserting item with type ${type}`);
            const path = getPathFromType(type);
            let url = `${this.baseUrl}/${path}`;
            return node_fetch_1.default(url, {
                method: "POST",
                body: JSON.stringify(frostItem)
            })
                .then(res => {
                if (!res.ok) {
                    console.error(`Error inserting item ${JSON.stringify(frostItem)} with type ${type}`);
                    res.text().then(text => console.error(text));
                    const err = new Error(`Error inserting item ${JSON.stringify(frostItem)} with type ${type}`);
                    err.name = "COULD_NOT_INSERT_ITEM";
                    throw err;
                }
                return res.ok;
            });
        });
    }
    updateItem(type, frostItem) {
        var _a;
        return __awaiter(this, void 0, void 0, function* () {
            // if (type == "location")
            //     console.debug(`inserting item with type ${type}`);
            const path = getPathFromType(type);
            const encodedId = (_a = frostItem["@iot.id"]) === null || _a === void 0 ? void 0 : _a.split("/").map(x => encodeURIComponent(x)).join("/");
            let url = `${this.baseUrl}/${path}('${encodedId}')`;
            return node_fetch_1.default(url, {
                method: "PATCH",
                body: JSON.stringify(frostItem)
            })
                .then(res => {
                if (!res.ok) {
                    console.error(`Error updating item ${JSON.stringify(frostItem)} with type ${type}`);
                    res.text().then(text => console.error(text));
                    const err = new Error(`Error inserting item ${JSON.stringify(frostItem)} with type ${type}`);
                    err.name = "COULD_NOT_INSERT_ITEM";
                    throw err;
                }
                return res.ok;
            });
        });
    }
    getSingleItemById(type, id, projection = DEFAULT_PROJECTION) {
        return __awaiter(this, void 0, void 0, function* () {
            const path = getPathFromType(type);
            const params = Object.assign({ "$filter": `id eq '${id}'`, "$top": 1 }, (projection ? { "$select": projection === null || projection === void 0 ? void 0 : projection.join(",") } : {}));
            return this.get(path, params).then(res => res.value[0]);
        });
    }
    get(urlPath, params) {
        return __awaiter(this, void 0, void 0, function* () {
            let url = `${this.baseUrl}/${urlPath}`;
            if (!params)
                params = {};
            if (Object.keys(params)) {
                url += "?";
                url += Object.entries(params)
                    .map(([key, value]) => [key, encodeURIComponent(value)])
                    .map(([key, value]) => `${key}=${value}`)
                    .join("&");
            }
            return node_fetch_1.default(url)
                .then(res => res.json())
                .then(json => {
                // console.debug(`Getting url "${url}"`);
                // console.debug(json);
                return json;
            });
        });
    }
}
exports.FrostClient = FrostClient;
//# sourceMappingURL=FrostClient.js.map