IoT and MongoDB: Powering Time Series Analysis of Household Power Consumption

1
const { Schema, model } = require('mongoose');
2
3
const powerConsumptionSchema = new Schema(
4
    {
5
        timestamp: { type: Date, index: true },
6
        global_active_power: { type: Number },
7
        global_reactive_power: { type: Number },
8
        voltage: { type: Number },
9
        global_intensity: { type: Number },
10
        sub_metering_1: { type: Number },
11
        sub_metering_2: { type: Number },
12
        sub_metering_3: { type: Number },
13
    },
14
    {
15
        timeseries: {
16
            timeField: 'timestamp',
17
            granularity: 'minutes',
18
        },
19
    }
20
);
21
22
const PowerConsumptions = model('PowerConsumptions', powerConsumptionSchema);
23
24
module.exports = PowerConsumptions;

1
// Load environment variables from .env file
2
require('dotenv').config();
3
4
// Import required modules
5
const fs = require('fs');
6
const mongoose = require('mongoose');
7
const PowerConsumptions = require('./models/power-consumption');
8
9
// Connect to MongoDB and process the data file
10
const processData = async () => {
11
    try {
12
        // Connect to MongoDB using the connection string from environment variables
13
        await mongoose.connect(process.env.MONGODB_CONNECTION_STRING);
14
15
        // Define the file path for the data source
16
        const filePath = 'Household_Power_Consumption.txt';
17
18
        // Read data file
19
        const rawFileContent = fs.readFileSync(filePath, 'utf8');
20
21
        // Normalize line endings and split the content into lines
22
        const lines = rawFileContent.replace(/\r\n/g, '\n').replace(/\r/g, '\n').trim().split('\n');
23
24
        // Extract column headers
25
        const headers = lines[0].split(';').map((header) => header.trim());
26
27
        // Parse the lines into structured objects
28
        const parsedRecords = lines.slice(1).map((line) => {
29
            const values = line.split(';').map((value) => value.trim());
30
            return headers.reduce((object, header, index) => {
31
                object[header] = values[index];
32
                return object;
33
            }, {});
34
        });
35
36
        // Transform and prepare data for insertion
37
        const transformedRecords = parsedRecords.map((item) => {
38
            const [day, month, year] = item.Date.split('/').map((num) => parseInt(num, 10));
39
            const [hour, minute, second] = item.Time.split(':').map((num) => parseInt(num, 10));
40
            const dateObject = new Date(year, month - 1, day, hour, minute, second);
41
42
            return {
43
                timestamp: dateObject.toISOString(),
44
                global_active_power: parseFloat(item.Global_active_power),
45
                global_reactive_power: parseFloat(item.Global_reactive_power),
46
                voltage: parseFloat(item.Voltage),
47
                global_intensity: parseFloat(item.Global_intensity),
48
                sub_metering_1: parseFloat(item.Sub_metering_1),
49
                sub_metering_2: parseFloat(item.Sub_metering_2),
50
                sub_metering_3: parseFloat(item.Sub_metering_3),
51
            };
52
        });
53
54
        // Filter out invalid data
55
        const finalData = transformedRecords.filter(
56
            (item) =>
57
            item.timestamp !== 'Invalid Date' &&
58
            !isNaN(item.global_active_power) &&
59
            !isNaN(item.global_reactive_power) &&
60
            !isNaN(item.voltage) &&
61
            !isNaN(item.global_intensity) &&
62
            !isNaN(item.sub_metering_1) &&
63
            !isNaN(item.sub_metering_2) &&
64
            !isNaN(item.sub_metering_3)
65
        );
66
67
        // Insert final data into the database in chunks of 1000
68
        const chunkSize = 1000;
69
        for (let i = 0; i < finalData.length; i += chunkSize) {
70
            const chunk = finalData.slice(i, i + chunkSize);
71
            await PowerConsumptions.insertMany(chunk);
72
        }
73
74
        console.log('Data processing and insertion completed.');
75
    } catch (error) {
76
        console.error('An error occurred:', error);
77
    }
78
};
79
80
// Call the processData function
81
processData();

1
MONGODB_CONNECTION_STRING = 'mongodb+srv://{{username}}:{{password}}@{{your_cluster_url}}/{{your_database}}?retryWrites=true&w=majority'

1
{ timestamp: { $gt: ISODate('2007-01-17T00:00:00.000-00:00'), $lt: ISODate('2007-01-18T00:00:00.000-00:00') } }