MyEMS开源能源管理系统v4.1.0源代码

import uuid from datetime import datetime, timezone, timedelta import falcon import mysql.connector import simplejson as json from core.useractivity import user_logger, admin_control, access_control import config class MicrogridCollection: @staticmethod def __init__(): """"Initializes MicrogridCollection""" pass @staticmethod def on_options(req, resp): resp.status = falcon.HTTP_200 @staticmethod def on_get(req, resp): access_control(req) cnx = mysql.connector.connect(**config.myems_system_db) cursor = cnx.cursor() query = (" SELECT id, name, uuid " " FROM tbl_contacts ") cursor.execute(query) rows_contacts = cursor.fetchall() contact_dict = dict() if rows_contacts is not None and len(rows_contacts) > 0: for row in rows_contacts: contact_dict[row[0]] = {"id": row[0], "name": row[1], "uuid": row[2]} query = (" SELECT id, name, uuid " " FROM tbl_cost_centers ") cursor.execute(query) rows_cost_centers = cursor.fetchall() cost_center_dict = dict() if rows_cost_centers is not None and len(rows_cost_centers) > 0: for row in rows_cost_centers: cost_center_dict[row[0]] = {"id": row[0], "name": row[1], "uuid": row[2]} query = (" SELECT id, name, uuid, " " address, postal_code, latitude, longitude, capacity, " " contact_id, cost_center_id, serial_number, svg, description " " FROM tbl_microgrids " " ORDER BY id ") cursor.execute(query) rows_microgrids = cursor.fetchall() result = list() if rows_microgrids is not None and len(rows_microgrids) > 0: for row in rows_microgrids: contact = contact_dict.get(row[8], None) cost_center = cost_center_dict.get(row[9], None) meta_result = {"id": row[0], "name": row[1], "uuid": row[2], "address": row[3], "postal_code": row[4], "latitude": row[5], "longitude": row[6], "capacity": row[7], "contact": contact, "cost_center": cost_center, "serial_number": row[10], "svg": row[11], "description": row[12], "qrcode": 'microgrid:' + row[2]} result.append(meta_result) cursor.close() cnx.close() resp.text = json.dumps(result) @staticmethod @user_logger def on_post(req, resp): """Handles POST requests""" admin_control(req) try: raw_json = req.stream.read().decode('utf-8') except Exception as ex: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.FAILED_TO_READ_REQUEST_STREAM') new_values = json.loads(raw_json) if 'name' not in new_values['data'].keys() or \ not isinstance(new_values['data']['name'], str) or \ len(str.strip(new_values['data']['name'])) == 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_MICROGRID_NAME') name = str.strip(new_values['data']['name']) if 'address' not in new_values['data'].keys() or \ not isinstance(new_values['data']['address'], str) or \ len(str.strip(new_values['data']['address'])) == 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_ADDRESS_VALUE') address = str.strip(new_values['data']['address']) if 'postal_code' not in new_values['data'].keys() or \ not isinstance(new_values['data']['postal_code'], str) or \ len(str.strip(new_values['data']['postal_code'])) == 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_POSTAL_CODE_VALUE') postal_code = str.strip(new_values['data']['postal_code']) if 'latitude' not in new_values['data'].keys() or \ not (isinstance(new_values['data']['latitude'], float) or isinstance(new_values['data']['latitude'], int)) or \ new_values['data']['latitude'] < -90.0 or \ new_values['data']['latitude'] > 90.0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_LATITUDE_VALUE') latitude = new_values['data']['latitude'] if 'longitude' not in new_values['data'].keys() or \ not (isinstance(new_values['data']['longitude'], float) or isinstance(new_values['data']['longitude'], int)) or \ new_values['data']['longitude'] < -180.0 or \ new_values['data']['longitude'] > 180.0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_LONGITUDE_VALUE') longitude = new_values['data']['longitude'] if 'capacity' not in new_values['data'].keys() or \ not (isinstance(new_values['data']['capacity'], float) or isinstance(new_values['data']['capacity'], int)) or \ new_values['data']['capacity'] <= 0.0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_CAPACITY_VALUE') capacity = new_values['data']['capacity'] if 'contact_id' not in new_values['data'].keys() or \ not isinstance(new_values['data']['contact_id'], int) or \ new_values['data']['contact_id'] <= 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_CONTACT_ID') contact_id = new_values['data']['contact_id'] if 'cost_center_id' not in new_values['data'].keys() or \ not isinstance(new_values['data']['cost_center_id'], int) or \ new_values['data']['cost_center_id'] <= 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_COST_CENTER_ID') cost_center_id = new_values['data']['cost_center_id'] if 'serial_number' not in new_values['data'].keys() or \ not isinstance(new_values['data']['serial_number'], str) or \ len(str.strip(new_values['data']['serial_number'])) == 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_SERIAL_NUMBER') serial_number = str.strip(new_values['data']['serial_number']) if 'svg' not in new_values['data'].keys() or \ not isinstance(new_values['data']['svg'], str) or \ len(str.strip(new_values['data']['svg'])) == 0: raise falcon.HTTPError(status=falcon.HTTP_400, title='API.BAD_REQUEST', description='API.INVALID_SVG') svg = str.strip(new_values['data']['svg']) if