怎么用脚本获取电能消耗数据

wen 实用脚本 2026-08-01 1

本文目录导读：

怎么用脚本获取电能消耗数据

从智能电表/能源监测设备获取
使用第三方API（如华为智能光伏、SolarEdge等）
使用MQTT协议（物联网设备）
读取电力公司数据（在线账单）
从本地数据库/导出文件读取
完整集成示例
使用建议

我可以帮你提供几种获取电能消耗数据的脚本方案,具体取决于你的数据来源，以下是几种常见场景：

从智能电表/能源监测设备获取

使用Python + Modbus协议（适用于工业电表）

import pymodbus
from pymodbus.client import ModbusTcpClient
class EnergyMeterReader:
    def __init__(self, host, port=502):
        self.client = ModbusTcpClient(host, port=port)
    def read_energy_data(self, unit_id=1):
        """读取电能数据"""
        try:
            self.client.connect()
            # 读取电压数据（寄存器地址示例）
            voltage = self.client.read_input_registers(0x0000, 2, unit=unit_id)
            # 读取电流数据
            current = self.client.read_input_registers(0x0002, 2, unit=unit_id)
            # 读取功率数据
            power = self.client.read_input_registers(0x0004, 2, unit=unit_id)
            # 读取累计电能
            energy = self.client.read_input_registers(0x0008, 4, unit=unit_id)
            return {
                '电压': voltage.registers,
                '电流': current.registers,
                '功率': power.registers,
                '累计电能': energy.registers
            }
        finally:
            self.client.close()
# 使用方法
reader = EnergyMeterReader('192.168.1.100')
data = reader.read_energy_data()
print(data)

使用第三方API（如华为智能光伏、SolarEdge等）

import requests
import pandas as pd
class SolarApiClient:
    def __init__(self, api_key, site_id):
        self.api_key = api_key
        self.site_id = site_id
        self.base_url = "https://api.yourprovider.com"
    def get_energy_data(self, start_date, end_date):
        """从云平台获取电能数据"""
        headers = {
            'Authorization': f'Bearer {self.api_key}',
            'Content-Type': 'application/json'
        }
        params = {
            'siteId': self.site_id,
            'startDate': start_date,
            'endDate': end_date
        }
        response = requests.get(
            f'{self.base_url}/energy',
            headers=headers,
            params=params
        )
        if response.status_code == 200:
            return response.json()
        else:
            raise Exception(f'API请求失败: {response.status_code}')
# 使用示例
client = SolarApiClient('your-api-key', 'site-123')
energy_data = client.get_energy_data('2024-01-01', '2024-01-31')

使用MQTT协议（物联网设备）

import paho.mqtt.client as mqtt
import json
import time
class EnergyMonitorMQTT:
    def __init__(self, broker, port=1883):
        self.broker = broker
        self.port = port
        self.client = mqtt.Client()
        self.energy_data = []
    def on_connect(self, client, userdata, flags, rc):
        print(f"连接成功，结果码: {rc}")
        # 订阅能量数据主题
        client.subscribe("home/energy/meter/#")
    def on_message(self, client, userdata, msg):
        """处理收到的能量数据"""
        try:
            payload = json.loads(msg.payload)
            self.energy_data.append({
                'timestamp': time.time(),
                'topic': msg.topic,
                'data': payload
            })
            # 实时打印数据
            print(f"{time.strftime('%Y-%m-%d %H:%M:%S')} - {msg.topic}: {payload}")
        except Exception as e:
            print(f"数据解析错误: {e}")
    def start_monitoring(self):
        self.client.on_connect = self.on_connect
        self.client.on_message = self.on_message
        self.client.connect(self.broker, self.port, 60)
        self.client.loop_start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            self.client.loop_stop()
            print("监控已停止")
# 使用示例
monitor = EnergyMonitorMQTT('mqtt.broker.com')
monitor.start_monitoring()

读取电力公司数据（在线账单）

import requests
from bs4 import BeautifulSoup
import json
class PowerBillReader:
    def __init__(self, username, password, utility_provider):
        self.username = username
        self.password = password
        self.utility_provider = utility_provider
        self.session = requests.Session()
    def get_consumption_data(self):
        """从电力公司网站获取用电数据"""
        # 示例：登录电力公司网站
        login_data = {
            'username': self.username,
            'password': self.password
        }
        # 登录
        login_url = f'https://{self.utility_provider}.com/login'
        response = self.session.post(login_url, data=login_data)
        if response.status_code == 200:
            # 获取用电数据页面
            usage_url = f'https://{self.utility_provider}.com/usage'
            usage_response = self.session.get(usage_url)
            # 解析HTML页面
            soup = BeautifulSoup(usage_response.content, 'html.parser')
            # 提取数据（需要根据实际网站结构调整）
            consumption_data = {}
            # 在这里添加解析逻辑
            return consumption_data
        else:
            raise Exception('登录失败')
# 使用示例（注意：需要正确的登录信息）
# bill_reader = PowerBillReader('username', 'password', 'your-electric-provider')
# usage_data = bill_reader.get_consumption_data()

从本地数据库/导出文件读取

import pandas as pd
import sqlite3
class EnergyDataProcessor:
    def __init__(self, data_source):
        self.data_source = data_source
    def read_from_csv(self, file_path):
        """从CSV文件读取电能数据"""
        df = pd.read_csv(file_path)
        # 数据清洗
        df['timestamp'] = pd.to_datetime(df['timestamp'])
        # 计算每日用电量
        daily_usage = df.groupby(df['timestamp'].dt.date)['energy'].sum()
        return daily_usage
    def read_from_database(self, db_path):
        """从SQLite数据库读取数据"""
        conn = sqlite3.connect(db_path)
        query = '''
        SELECT timestamp, energy_consumption as energy
        FROM energy_meter
        WHERE timestamp >= date('now', '-30 days')
        ORDER BY timestamp DESC
        '''
        df = pd.read_sql_query(query, conn)
        conn.close()
        return df
# 使用示例
processor = EnergyDataProcessor('any-source')
csv_data = processor.read_from_csv('energy_usage.csv')
database_data = processor.read_from_database('energy.db')

完整集成示例

import time
import json
import requests
from datetime import datetime
import schedule
class EnergyScheduler:
    def __init__(self, config):
        self.config = config
        self.data_sources = []
    def setup_data_sources(self):
        """初始化所有数据源"""
        # 添加不同的数据源
        for source_config in self.config['sources']:
            source_type = source_config['type']
            if source_type == 'modbus':
                from pymodbus.client import ModbusTcpClient
                client = ModbusTcpClient(
                    source_config['host'], 
                    port=source_config.get('port', 502)
                )
                self.data_sources.append(client)
            elif source_type == 'api':
                # 添加API数据源
                self.data_sources.append({
                    'type': 'api',
                    'endpoint': source_config['endpoint'],
                    'api_key': source_config['api_key']
                })
    def collect_data(self):
        """收集所有数据源的数据"""
        all_data = []
        for source in self.data_sources:
            try:
                if source.get('type') == 'api':
                    # 从API获取数据
                    data = self.fetch_api_data(source)
                else:
                    # 从Modbus设备获取
                    data = self.fetch_modbus_data(source)
                all_data.append({
                    'timestamp': datetime.now().isoformat(),
                    'source': source.get('name', 'unknown'),
                    'data': data
                })
            except Exception as e:
                print(f"收集数据失败: {e}")
        return all_data
    def fetch_modbus_data(self, source):
        """从Modbus设备获取数据"""
        # 实现具体的Modbus读取逻辑
        pass
    def fetch_api_data(self, source):
        """从API获取数据"""
        response = requests.get(
            source['endpoint'],
            headers={'Authorization': f"Bearer {source['api_key']}"}
        )
        return response.json()
    def schedule_collection(self, interval_minutes=15):
        """定时采集数据"""
        schedule.every(interval_minutes).minutes.do(self.collect_data)
        while True:
            schedule.run_pending()
            time.sleep(1)
# 配置示例
config = {
    'sources': [
        {
            'name': '主电表',
            'type': 'modbus',
            'host': '192.168.1.100',
            'port': 502
        },
        {
            'name': '太阳能系统',
            'type': 'api',
            'endpoint': 'https://api.solar-system.com/energy',
            'api_key': 'your-api-key'
        }
    ]
}
# 使用示例
energy_manager = EnergyScheduler(config)
energy_manager.setup_data_sources()
energy_manager.schedule_collection(interval_minutes=15)

使用建议

选择方案：根据你的实际设备情况选择合适的方案
安全性：涉及能源数据时注意数据安全和隐私
准确性：确保时间戳和数据校准正确
容错性：添加异常处理和重试机制
可视化：可以将收集的数据存储到数据库并用图表展示

需要我针对某个特定方案详细说明,或者调整脚本以满足你的具体需求吗？