demo_api.py

#!/usr/bin/env python
# -*-coding:utf-8 -*-
'''
@File    :   custom_api.py
@Time    :   2023/04/09 17:01:04
@Author  :   NSL
@Version :   1.0
@Contact :   sroy10@uh.edu
@Desc    :   None
'''


from flask import Flask, request, jsonify
from datetime import datetime, timedelta
import pytz
import psycopg2

app = Flask(__name__)

# Connect to PostgreSQL database
conn = psycopg2.connect(host="localhost", dbname="your_database_name", user="your_username", password="your_password")

# /auth endpoint for user authentication
@app.route('/auth')
def authenticate_user():
    user_id = request.args.get('user')
    password = request.args.get('pass')
    # Validate user credentials from database
    cur = conn.cursor()
    cur.execute("SELECT * FROM users WHERE user_id=%s AND password=%s", (user_id, password))
    user = cur.fetchone()
    cur.close()
    if user:
        return jsonify({"status": "success"})
    else:
        return jsonify({"status": "fail"})

# /data endpoint for retrieving sensor data
@app.route('/data')
def get_sensor_data():
    user_id = request.args.get('user')
    password = request.args.get('pass')
    n_highres = request.args.get('nhires', type=int)
    n_lowres = request.args.get('nlowres', type=int)
    sensors = request.args.get('sensors')

    # Authenticate user
    cur = conn.cursor()
    cur.execute("SELECT * FROM users WHERE user_id=%s AND password=%s", (user_id, password))
    user = cur.fetchone()
    cur.close()
    if not user:
        return jsonify({"status": "fail", "message": "Authentication failed."})

    # Get current time in UTC timezone
    current_time = datetime.utcnow()

    # Get high resolution data
    if n_highres:
        cur = conn.cursor()
        cur.execute("SELECT Energy_Total,Energy_Today,Energy_Yesterday,Energy_Power,Energy_Factor,Energy_Voltage,Energy_Current, {} FROM sensor WHERE device_id=%s ORDER BY record_time DESC LIMIT %s".format(sensors), (user_id, n_highres))
        highres_data = cur.fetchall()
        cur.close()
        # Convert high resolution data to local timezone (Nepal)
        highres_data_local = []
        for data in highres_data:
            record_time_local = pytz.utc.localize(data[0]).astimezone(pytz.timezone('Asia/Kathmandu'))
            highres_data_local.append({"sensor": sensors, "time": record_time_local, "value": data[2]})
    else:
        highres_data_local = []

    # Get low resolution data
    if n_lowres:
        cur = conn.cursor()
        start_time = current_time - timedelta(days=n_lowres)
        cur.execute("SELECT Energy_Total,Energy_Today,Energy_Yesterday,Energy_Power,Energy_Factor,Energy_Voltage,Energy_Current, {} FROM sensor WHERE device_id=%s AND record_time >= %s GROUP BY device_id".format(sensors), (user_id, start_time))
        lowres_data = cur.fetchall()
        cur.close()
        # Convert low resolution data to local timezone (Nepal)
        lowres_data_local = []
        for data in lowres_data:
            lowres_data_local.append({"sensor": sensors, "time": start_time.astimezone(pytz.timezone('Asia/Kathmandu')), "value": data[1]})
    else:
        lowres_data_local = []

    return jsonify({"status": "success", "id": user_id, "highres": highres_data_local, "lowres": lowres_data_local})


# sonoff status endpoint
@app.route('/sonoff_status', methods=['GET'])
def sonoff_status(device_full_id):
    conn = psycopg2.connect(dbname='your_db_name', user='your_db_user', password='your_db_password', host='your_db_host')
    cur = conn.cursor()
    query = ""
    cur.execute("SELECT CASE WHEN MAX('record_time') > NOW() - INTERVAL '1 minute'\
        THEN 1 ELSE 0 END as status FROM sensor WHERE device_id=%s;", (device_full_id))
    status = cur.fetchall()
    cur.close()
    return status

# list all sonoffs endpoint
@app.route('/list_all_sonoffs', methods=['GET'])
def list_all_sonoffs():
    conn = psycopg2.connect(dbname='your_db_name', user='your_db_user', password='your_db_password', host='your_db_host')
    cur = conn.cursor()
    cur.execute("SELECT 'device_id' FROM sensors")
    sonoff_list = cur.fetchall()
    cur.close()
    return sonoff_list

# sonoff power endpoint
@app.route('/sonoff_power', methods=['GET'])
def sonoff_power():
    # implement it
    pass


if __name__ == '__main__':
    app.run()