Tibber 的 start_live_feed() 默认启动无限监听循环，导致后续代码无法执行；通过传入始终返回 True 的 exit_condition 函数，可强制其仅接收一次实时测量数据后立即退出，满足定时脚本或数据库单次写入需求。
tibber 的 `start_live_feed()` 默认启动无限监听循环，导致后续代码无法执行；通过传入始终返回 `true` 的 `exit_condition` 函数，可强制其仅接收一次实时测量数据后立即退出，满足定时脚本或数据库单次写入需求。
在自动化能源监控场景中，常需定时采集 Tibber 平台的实时功耗（如 power）、电压或频率等指标，并写入数据库。但官方 tibber Python SDK 的 start_live_feed() 方法默认设计为长连接、持续接收 WebSocket 数据流——一旦调用，程序即阻塞于该异步循环中，后续逻辑（如数据库插入、日志记录或脚本退出）将永不执行。
解决关键在于利用其内置的 exit_condition 参数：它接受一个回调函数，在每次成功接收到 live_measurement 数据后被调用，若返回值为真值（truthy），则立即终止监听循环，控制权交还给主程序。
以下为精简、可靠、可直接用于定时任务（如 cron 或 GitHub Actions）的单次采集示例：
Secure Python Code Sandbox
下载
安全 Python 代码沙箱：在隔离的沙箱中执行任意 Python 代码，预装 requests、NumPy、pandas 库。返回 stdout、stderr。
import asyncio
import tibber
TOKEN_STRING = "your_tibber_access_token_here" # 替换为实际 Token
async def fetch_single_measurement():
account = tibber.Account(TOKEN_STRING)
await account.update_info() # 确保 homes 列表已加载
if not account.homes:
raise RuntimeError("No Tibber home found for this account")
home = account.homes[0]
latest_power = None
@home.event("live_measurement")
def on_measurement(data):
nonlocal latest_power
latest_power = data.power # 或 data.voltage, data.frequency 等
print(f"Received power: {latest_power} W")
# 关键：exit_condition=lambda _: True 表示收到第一条数据后立刻退出
await home.start_live_feed(
user_agent="MyScript/1.0",
exit_condition=lambda _: True
)
return latest_power
# 启动异步主函数
if __name__ == "__main__":
power_value = asyncio.run(fetch_single_measurement())
if power_value is not None:
# ✅ 此处可安全执行数据库写入（如 SQLite、PostgreSQL、InfluxDB）
print(f"✅ Ready to insert into DB: power = {power_value} W")
# 示例（伪代码）：db.insert("measurements", {"power": power_value, "timestamp": datetime.now()})⚠️ 注意事项：
立即学习“Python免费学习笔记（深入）”；
- exit_condition 的参数是接收到的原始 data 对象（类型为 tibber.LiveMeasurement），你可在其中提取任意字段（如 data.power, data.energy, data.lastMeterConsumption）；
- 必须使用 await 调用 start_live_feed()，且整个流程需包裹在 asyncio.run() 中；
- 若 account.homes 为空，请确认 Token 权限正确且账户已绑定家庭；
- 不建议在 @home.event 回调中执行耗时操作（如同步数据库写入），应优先保存数据，再在循环退出后统一处理；
- 生产环境建议添加超时保护（如 asyncio.wait_for(..., timeout=15)）防止网络异常卡死。
该方案摒弃了轮询或信号中断等复杂机制，以最小侵入方式复用 SDK 原生能力，精准实现“触发即采、采完即止”的轻量级数据采集目标。
