Commit 2707e1db authored by Giannis Kepas's avatar Giannis Kepas
Browse files

Merge branch 'device-measurement' into 'main' 

(http-bridge): Refactor and add DeviceMeasurement data type

Closes #1

See merge request !1
parents 31389d27 98fb96f1

README.md

+29 −9
Original line number Diff line number Diff line
@@ -113,19 +113,39 @@ Move into the `http-bridge` folder and execute the following: 
### NGSI-LD Entity Types



| Entity Type           | Description                                                              |

| ------------------- | ---------------------------------------------------------- |

| ``WeatherObserved`` | Real-time weather conditions (temperature, humidity, etc.) |



Additional entity types can be added dynamically by extending the Type Mapping Configuration.

| --------------------- | ------------------------------------------------------------------------ |

| ``DeviceMeasurement`` | Device measurements which include records of all the sensors registered. |



To support more NGSI-LD models:



1. Add a new type mapping in ``typeMaps/``

1. Add a new type mapping in ``typeMaps/``. See the DeviceMeasurement typemap for the base structure.

2. Extend NGSITranslator.js to handle the new entity type

3. Modify generalControllers.js to process the new queries



## Testing 

## Queries & Testing 



You can view the structure of HTTP queries and test the broker/bridge by using the following Postman workspace:



**https://www.postman.com/retr0dev/workspace/uowm-depe-http-bridge**



Clone the workspace into your own account and modify it to your needs.



**Setting up Postman**



In order to test the broker and the http bridge you can use the sample queries provided at the following Postman workspace:

1. Setup the env variables.

   - `broker-ip` : localhost if running locally otherwise the IP of your Scorpio broker

   - `broker-port` : the port of the Scorpio broker

   - `http-bridge-ip` : the IP address of the bridge - **CANNOT BE LOCALHOST** - If you are running the bridge on the same machine as your broker then you need to set the bridge IP to the machine IP.

   - `http-bridge-port` : the port of the HTTP bridge

2. Register Context Source

   - In the `Scorpio-Registration` POST request you need to add any new entity types you have implemented.

3. Test the broker

    - If everything was setup correctly, you can use any of the GET requests to fetch data according to the request you selected.

    The format the requests are following (where 'XYZ' is the entity type to fetch ) is:

        - XYZ-Realtime : Latest data from the specific type.

        - XYZ-Realtime-Attrs : Latest data from the specific type and the specific attribute listed in the params.

        - XYZ-Temporal : Data from a specific date range.

        - XYZ-Temporal-Attrs : Data from a specific date range and a specific attribute listed in the params.



**https://www.postman.com/payload-geoscientist-32828968/scorpio-broker-registration/overview**
 
*Note:*

The reason all Realtime requests have a limit on the database entries it will return is because we cannot fetch millions of records from a single query. The server cannot handle that.

http-bridge/README.md

+2 −78
Original line number Diff line number Diff line
@@ -2,84 +2,8 @@ 


## **Overview**



The **HTTP Bridge** serves as a middleware between the **Scorpio Broker** and the **Smart City Heraklion API**. It is responsible for **fetching data from the API**, **translating it into the NGSI-LD format**, and **returning a structured response** that conforms to the **NGSI-LD data model**.

The **HTTP Bridge** serves as a middleware between the **Scorpio Broker** and a **API**. It is responsible for **fetching data from the API**, **translating it into the NGSI-LD format**, and **returning a structured response** that conforms to the **NGSI-LD data model**.



---



## **Getting Started**



### **1. Prerequisites**

Ensure you have the following installed:

- **Docker** 

- **Node.js** (v14 or higher, if running locally)

- **npm** (Node Package Manager)

---



### **2. Configuration**

Before running the application, configure the necessary environment variables.



### **Using `.env` File (Recommended)**

Create a `.env` file in the project root with the following content:



```

PORT=9010 # Http Bridge Port

HOST="0.0.0.0" # Http Bridge Host (DONT CHANGE id running with Docker)

API_URL=https://smartcity.heraklion.gr/open-data-api # Base URL of the API

```



### **3. Running the HTTP Bridge**

1. Install npm packages

```

npm install

```



2. Build the Docker Image

```

docker build -t http-bridge .

```



3. Run the Container



```

docker run -d --name http-bridge -p 8080:8080 --env-file .env http-bridge

```



The bridge will be accessible at: **{HOST}:{PORT}**



Default: **http://localhost:8080**



---

### **4. Stopping & Removing the Container**



1. To stop the running container:

```

docker stop http-bridge

```



2. To remove the container:

```

docker rm http-bridge

```



---



## **Supported NGSI-LD Entity Types**

| **Entity Type**      | **Description**                                        |

|----------------------|--------------------------------------------------------|

| `WeatherObserved`    | Real-time weather conditions (temperature, humidity, etc.) |



Additional entity types can be added dynamically by extending the **Type Mapping Configuration**.



---



## **Extending the Bridge**

To support more NGSI-LD models, simply:

1. **Add a new type mapping** in `typeMaps/`

2. **Extend `NGSITranslator.js`** to handle the new entity type

3. **Modify `generalControllers.js`** to process the new queries



---



## **Extra Links**

[Scorpio Broker](https://github.com/efntallaris/scorpioBroker)

[Postman Workplace](https://www.postman.com/payload-geoscientist-32828968/workspace/scorpio-broker-registration)
 
 No newline at end of file
 
**See the README on the root of this repository for instructions on how to install and setup.**

http-bridge/controllers/generalControllers.js

+89 −86
Original line number Diff line number Diff line 
const axios = require("axios");

const qs = require("qs");

const NGSITranslator = require("../translator/NGSITranslator");

const typeMaps = require("../typeMaps/typeMaps");



/**

 * Generates a modality_id parameter based on the type and attributes provided.

 * If attributes are given, it will filter the modalities to only include the ones

 * that correspond to the given attributes. Otherwise, it will return all modalities

 * available for the given type.

 * 

 * Based on Smart City Heraklion API

 *

 * @param {string} type The NGSI-LD type to generate the modality_id for

 * @param {string} type The NGSI-LD type to generate the measurement_type for

 * @param {string} attributes A comma-separated list of attributes to filter by

 * @throws {Error} If the type is invalid or no valid attributes are found

 * @returns {string} The modality_id parameter as a string

 * @returns {string[]} List of measurement types

 */

function generateModalityIdParam(type, attributes) {

    const modalityMap = typeMaps[type];

    if (!modalityMap) {

function generateMeasurementTypeParam(type, attributes) {

    const typeMap = typeMaps[type];

    if (!typeMap) {

        throw new Error(`Invalid type: ${type}`);

    }



    // Filter attributes if provided, otherwise return all

    const modalityIds = attributes

        ? attributes.split(",").map(attr => Object.entries(modalityMap)

    const measurementTypes = attributes

        ? attributes.split(",").map(attr => Object.entries(typeMap)

            .find(([_, value]) => value.slug === attr)?.[0])

            .filter(Boolean)

        : Object.keys(modalityMap);

        : Object.keys(typeMap);



    if (modalityIds.length === 0) {

        throw new Error(`No valid attributes found for type: ${type}`);

    if (measurementTypes.length === 0) {

        throw new Error(`No valid attributes found for type '${type}' and attributes '${attributes}'`);

    }



    return modalityIds;

    // console.log("Measurement types:", measurementTypes);

    return measurementTypes;

}



/**

 * Fetches data from the Smart City Heraklion API and translates it to NGSI-LD.

 * Fetches data from the provided API and translates it to NGSI-LD.

 * 

 * @param {string} type The NGSI-LD type to retrieve

 * @param {string} [attrs] A comma-separated list of attributes to filter by

 * @param {Function} fetcher The specific fetcher function to use (fetchWeatherObservedTemporalData or fetchWeatherObservedRealTimeData)

 * @param {Function} fetcher The specific fetcher function to use (fetchTemporalData or fetchRealTimeData)

 * @param {Object} [additionalParams] Additional parameters to pass to the fetcher function

 * @param {Response} res The Express response object

 * @throws {Error} If the type is invalid, no valid attributes are found, or the API request fails
@@ -55,23 +50,20 @@ const fetchEntities = async (type, attrs, fetcher, additionalParams, res) => { 
            return res.status(400).json({ error: `Unsupported entity type: ${type}` });

        }



        const modalityIdParam = generateModalityIdParam(type, attrs);

        if (!modalityIdParam) {

            return res.status(400).json({ error: "Missing modality_id parameter." });

        const measurementTypes = generateMeasurementTypeParam(type, attrs);

        if (!measurementTypes) {

            return res.status(400).json({ error: "Missing measurement_type parameter." });

        }



        // Call the specific fetcher function

        const apiResponse = await fetcher(type, modalityIdParam, additionalParams);



        if (!apiResponse || !apiResponse.data) {

            return res.status(500).json({ error: "Failed to retrieve data from API." });

        const apiResponse = await fetcher(type, measurementTypes, additionalParams);

        if (!apiResponse || !apiResponse.data || apiResponse.data.length === 0) {

            return res.status(500).json({ error: "Failed to retrieve data from API. See http-bridge logs for info." });

        }



        // Translate response to NGSI-LD

        const ngsiData = NGSITranslator.toNGSI(apiResponse.data, type);

        res.status(200).json(ngsiData);

    } catch (error) {

        console.error("Error in fetchEntities:", error.message);

        console.error(`error in fetchEntities\n${error.stack}\n`);

        res.status(500).json({ error: error.message || "Failed to fetch entities." });

    }

};
@@ -80,67 +72,65 @@ const fetchEntities = async (type, attrs, fetcher, additionalParams, res) => { 
 * Fetches the temporal data for a given NGSI-LD entity type.

 * 

 * @param {string} type The NGSI-LD type to retrieve

 * @param {string} modalityIdParam The modality_id parameter for filtering

 * @param {string} measurementTypes The measurement type parameter for filtering

 * @param {Object} [params] Additional parameters for the API request

 * @throws {Error} If the entity type is unsupported

 */

const fetchWeatherObservedTemporalData = async (type, modalityIds, params) => {

    if (!Array.isArray(modalityIds) || modalityIds.length === 0) {

        throw new Error(`Invalid modality IDs for type: ${type}`);

    }



    const results = { data: [] };



    // Loop through each modality ID and make separate requests

    for (const modalityId of modalityIds) {

const fetchTemporalData = async (type, measurementTypes, params) => {

    switch (type) {

        case "DeviceMeasurement":

            try {

            const apiResponse = await axios.post(`${process.env.API_URL}/rpc/measurements_averages`, {

                _devices: "{71,72,73,63,64,36,68,70,61,69,67,66,65,60,59,120}",

                _from: params.time,

                _to: params.endTime,

                _granularity: "1 hour",

                _modality: typeMaps[type][modalityId].key,

                _show_out_of_range: false

                const response = await axios.get(`${process.env.API_URL}/device_measurement`, {

                    params: {

                        measurement_type: `in.(${measurementTypes.join(",")})`,

                        ts: params.ts,

                    },

                    paramsSerializer: (params) => qs.stringify(params, { encode: false }) // Prevent double encoding

                });



            if (apiResponse && apiResponse.data) {

                formattedData = apiResponse.data.map(entry => ({

                    device_id: modalityId,

                    modality_id: parseInt(modalityId, 10),

                    value: entry.avg, 

                    observed_at: entry.at

                }));



                results.data.push(...formattedData);

            }

                return response;

            } catch (error) {

            console.error(`Error fetching data for modalityId ${modalityId}:`, error.message);

        }

                console.error(`error fetching temporal data\n${error.stack}`);

                return { data: [] };

            }



    return results;

        default:

            return { data: [] };

    }

};



/**

 * Fetches the latest real-time data for a given NGSI-LD entity type.

 * 

 * @param {string} type The NGSI-LD type to retrieve

 * @param {string} modalityIdParam The modality_id parameter for filtering

 * @param {string} measurementTypes The measurement type parameter for filtering

 * @throws {Error} If the entity type is unsupported

 */



const fetchWeatherObservedRealTimeData = async (type, modalityIdParam) => {

const fetchRealTimeData = async (type, measurementTypes, params) => {

    switch (type) {

        case "WeatherObserved":

            return axios.get(`${process.env.API_URL}/latest_measurements_averages`, {

                params: { modality_id: `in.(${Array.isArray(modalityIdParam) ? modalityIdParam.join(",") : modalityIdParam})` },

        case "DeviceMeasurement":

            try {

                const response = await axios.get(`${process.env.API_URL}/device_measurement`, {

                    params: {

                        measurement_type: `in.(${measurementTypes.join(",")})`,

                        order: "ts.desc",

                        // This is set in order to avoid fetching all data and not

                        // to overload the system. This can be adjusted as needed.

                        limit: params.limit || 1000,

                    }

                });



                return response;

            } catch (error) {

                console.error(`error fetching real-time data\n${error.stack}`);

                return { data: [] };

            }



        default:

            throw new Error(`Unsupported real-time entity type: ${type}`);

            return { data: [] };

    }

};





/**

 * Handles incoming requests for temporal NGSI-LD entities.

 *
@@ -149,26 +139,37 @@ const fetchWeatherObservedRealTimeData = async (type, modalityIdParam) => { 
 * @prop {string} time The timestamp to apply the temporal relationship to

 * @prop {string} endTime The end time for the time range query

 * @prop {string} attrs A comma-separated list of attributes to filter by

 * @prop {string} limit The maximum number of entities to return

 */

const getTemporalEntities = async (req, res) => {

    console.log("Incoming temporal request:", req.query);

    let { type, timerel, time, endTime, attrs } = req.query;



    // Decode URL-encoded time values (fixes %3A issue)

    time = decodeURIComponent(time);

    if (endTime) endTime = decodeURIComponent(endTime);

    let tsParams = [];



    switch (timerel) {

        case "after":

            tsParams.push(`gte.${time}`);

            break;



        case "before":

            const adjustedStartTime = new Date(new Date(time).setDate(new Date(time).getDate() - 30)).toISOString();

            tsParams.push(`gte.${adjustedStartTime}`, `lt.${time}`);

            break;



    // Ensure proper ISO 8601 formatting

    const now = new Date().toISOString();

        case "between":

            if (!endTime) {

                return res.status(400).json({ error: "endTime is required for 'between' queries" });

            }

            tsParams.push(`gte.${time}`, `lt.${endTime}`);

            break;



    if (timerel === "after") {

        endTime = now; // Set endTime to current time

    } else if (timerel === "before") {

        endTime = time; // Swap values for "before"

        time = new Date(new Date(endTime).setDate(new Date(endTime).getDate() - 30)).toISOString(); // 30 days before

        default:

            return res.status(400).json({ error: "Invalid timerel value" });

    }



    await fetchEntities(type, attrs, fetchWeatherObservedTemporalData, { time, endTime }, res);

    // console.log("Final timestamp:", tsParams.join(", "));

    await fetchEntities(type, attrs, fetchTemporalData, { ts: tsParams }, res);

};



/**
@@ -176,11 +177,13 @@ const getTemporalEntities = async (req, res) => { 
 * 

 * @prop {string} type The NGSI-LD type to retrieve

 * @prop {string} attrs A comma-separated list of attributes to filter by

 * @prop {string} limit The maximum number of entities to return

 */

const getRealTimeEntities = async (req, res) => {

    console.log("Incoming real-time request:", req.query);

    const { type, attrs } = req.query;

    await fetchEntities(type, attrs, fetchWeatherObservedRealTimeData, {}, res);

    const { type, attrs, limit } = req.query;



    await fetchEntities(type, attrs, fetchRealTimeData, { limit }, res);

};



module.exports = { getTemporalEntities, getRealTimeEntities };

http-bridge/package-lock.json

+302 −106

File changed.

Preview size limit exceeded, changes collapsed.

http-bridge/package.json

+3 −2
Original line number Diff line number Diff line 
{

  "name": "wrappertest",

  "name": "uowm-http-bridge",

  "version": "1.0.0",

  "main": "index.js",

  "scripts": {
@@ -13,6 +13,7 @@ 
    "axios": "^1.7.8",

    "dotenv": "^16.4.5",

    "express": "^4.21.1",

    "morgan": "^1.10.0"

    "morgan": "^1.10.0",

    "qs": "^6.14.0"

  }

}