-
-
Save cbpowell/cb9cf9c34f68530374eb2cfa777b597f to your computer and use it in GitHub Desktop.
sources: | |
- mutable: | |
plugs: | |
- TesSense: | |
alias: "Tesla Charger" | |
mac: 50:c7:bf:f6:4f:39 # matches value in TesSenseLink.py line 245 | |
power: 0 # will be updated with live values |
import asyncio | |
from SenseLink import * | |
import sys | |
username = 'elon@tesla.com' # Sense's and Tesla's login | |
password = 'password' # Sense's password, Tesla will prompt for it's own | |
""" | |
TesSense -Randy Spencer 2022 | |
Python charge monitoring utility for those who own the Sense Energy Monitor | |
Uses the stats for Production and Utilization of electricity to control | |
your main Tesla's AC charging to use excess production only when charging. | |
Simply plug in your car, update your info above, and type> python3 tessense.py | |
Added: another minute to remove powering off due to spurius spikes/clouds | |
Add: reporting the Tesla's charging to Sense as if plugged into a TP-Link/Kasa | |
Add: ability to find TP-Link devices on the network and control them. | |
""" | |
import datetime | |
from time import sleep | |
#/c Moved these inside the new async function due to async variable scopes - there are other ways to handle this | |
# but I did it this way for my own simplicity. You could pass them in as function arguments as well | |
# rate = minrate = 0 # Minimum rate you want to set the charger to | |
# volts = 120 # Minimum volts, until detected by the charger | |
# pshown = loop = charge = charging = False # init more variables | |
#/c Set stdout as logging handler | |
root_log = logging.getLogger() | |
root_log.setLevel(logging.DEBUG) | |
handler = logging.StreamHandler(sys.stdout) | |
def printmsg(msg): # Timestamped message | |
print(datetime.datetime.now().strftime("%a %I:%M %p"), msg) | |
def printerror(error, err): # Error message with truncated data | |
print(datetime.datetime.now().strftime("%a %I:%M %p"), error + "\n", str(err).split("}")[0], "}") | |
# To install support module: | |
# Python3 -m pip install sense_energy | |
print("Initating connection to Sense...") | |
from sense_energy import Senseable | |
sense = Senseable(wss_timeout=30, api_timeout=30) | |
sense.authenticate(username, password) | |
# Python3 -m pip install teslapy | |
print("Starting connection to Tesla...") | |
import teslapy | |
#/c wrap main code in async function | |
async def run_tes_sense(mutable_plug): | |
rate = minrate = 0 # Minimum rate you want to set the charger to | |
volts = 120 # Minimum volts, until detected by the charger | |
pshown = loop = charge = charging = False # init more variables | |
with teslapy.Tesla(username) as tesla: | |
#/c I swapped to token auth here, simply because I already had it for another program | |
if not tesla.authorized: | |
tesla.refresh_token(refresh_token=input('Enter SSO refresh token: ')) | |
vehicles = tesla.vehicle_list() | |
car = vehicles[0].get_vehicle_summary() | |
print(car['display_name'], "is", car['state'], "\n") | |
# print( "Tesla Info:\n", vehicles[0].get_vehicle_data(), "\n\n" ) # shows all car data available | |
# print( "TeslaPy:\n", dir( teslapy ), "\n\n" ) # shows all the TeslaPy API functions | |
# print( "Senseable:\n", dir( Senseable ), "\n\n" ) # shows all the Sense API functions | |
while (True): | |
try: | |
sense.update_realtime() # Update Sense info | |
asp = int(sense.active_solar_power) | |
ap = int(sense.active_power) | |
power_diff = asp - ap # Watts being set back to the grid | |
except: | |
printmsg("Sense Timeout") | |
#/c changed to asyncio sleep | |
await asyncio.sleep(10) | |
continue # back to the top of the order | |
#/c Assume not charging to start | |
charging = False | |
if vehicles[0].available(): | |
try: | |
chargedata = vehicles[0].get_vehicle_data()['charge_state'] | |
except teslapy.HTTPError as e: | |
printerror("Tesla failed to update, please wait a minute...", e) | |
#/c changed to asyncio sleep | |
await asyncio.sleep(60) | |
continue | |
if chargedata['charging_state'] == "Disconnected": # Loop w/o msgs until connected | |
if not pshown: | |
printmsg(" Please plug the vehicle in...\n\n") | |
pshown = True | |
await asyncio.sleep(60) | |
continue | |
else: | |
pshown = False | |
if chargedata['battery_level'] >= chargedata['charge_limit_soc']: # Loop when full | |
#/c Changed this to a continue, to avoid exiting just due to full battery, as my car was full! | |
#/c exit("Full Battery!") | |
#/c set 0 watts to plug | |
mutable_plug.data_source.power = 0 | |
#/c added a (shorter) asyncio sleep | |
await asyncio.sleep(20) | |
continue | |
if chargedata['fast_charger_present']: # Loop while DC Fast Charging | |
printmsg("Supercharging...") | |
if charge != chargedata['battery_level']: | |
charge = chargedata['battery_level'] | |
print("\nLevel:", | |
chargedata['battery_level'], "%, Limit", | |
chargedata['charge_limit_soc'], "%, Rate", | |
chargedata['charger_power'], "kW, ", | |
chargedata['charger_voltage'], "Volts, ", | |
chargedata['fast_charger_type'], | |
chargedata['minutes_to_full_charge'], "Minutes remaining\n") | |
# /c set 0 watts to plug | |
mutable_plug.data_source.power = 0 | |
#/c use asyncio sleep | |
await asyncio.sleep(60) | |
continue | |
if chargedata['charging_state'] == "Charging": # Collect charging info | |
volts = chargedata['charger_voltage'] | |
rate = chargedata['charge_current_request'] | |
maxrate = chargedata['charge_current_request_max'] | |
newrate = min(rate + int(power_diff / volts), maxrate) | |
if newrate == 1 or newrate == 2: newrate = 0 | |
charging = True | |
#/c Could alternatively calc the new power and set it to the plug here | |
charge_power = chargedata['charger_power'] | |
mutable_plug.data_source.power = charge_power | |
else: | |
charging = False | |
if charging: # check if need to change rate or stop | |
if power_diff > 1: | |
print("Charging at", rate, "amps, with", power_diff, "watts surplus") | |
if newrate > rate: | |
print("Increasing charging to", newrate, "amps") | |
try: | |
vehicles[0].command('CHARGING_AMPS', charging_amps=newrate) | |
if newrate < 5: vehicles[0].command('CHARGING_AMPS', charging_amps=newrate) | |
rate = newrate | |
except teslapy.VehicleError as e: | |
printerror("Error up", e) | |
except teslapy.HTTPError as e: | |
printerror("Failed up", e) | |
elif power_diff < -1: # Not enough free power to continue charging | |
print("Charging at", rate, "amps, with", power_diff, "watts usage") | |
if newrate < minrate: # Stop charging when below minrate | |
if not loop: | |
print("Just Once More") # Delay powering off once for suprious data | |
loop = True | |
else: | |
print("Stopping charge") | |
try: | |
vehicles[0].command('STOP_CHARGE') | |
except teslapy.VehicleError as e: | |
printerror("Won't stop charging", e) | |
except teslapy.HTTPError as e: | |
printerror("Failed to stop", e) | |
loop = False | |
elif newrate < rate: # Slow charging to match free solar | |
print("Slowing charging to", newrate, "amps") | |
try: | |
vehicles[0].command('CHARGING_AMPS', charging_amps=newrate) | |
if newrate < 5: vehicles[0].command('CHARGING_AMPS', charging_amps=newrate) | |
rate = newrate | |
except teslapy.VehicleError as e: | |
printerror("Error down", e) | |
except teslapy.HTTPError as e: | |
printerror("Failed down", e) | |
else: | |
print("Charging at", rate, "amps") # -1, 0 or 1 watt diff, so don't say "watts" | |
else: # NOT Charging, check if time to start | |
print("Not Charging, Spare power at", power_diff, "watts") | |
if power_diff > (minrate * volts): # Minimum free watts to wake car and charge | |
print("Starting charge") | |
try: | |
vehicles[0].sync_wake_up() | |
except teslapy.VehicleError as e: | |
printerror("Won't wake up", e) | |
except teslapy.VehicleError as e: | |
printerror("Wake Timeout", e) | |
if vehicles[0].available(): | |
try: # Check LIVE data first | |
chargedata = vehicles[0].get_vehicle_data()['charge_state'] | |
if chargedata['charging_state'] == "Disconnected": | |
print(" Please plug the vehicle in...") | |
pshown = True | |
continue | |
elif chargedata['charging_state'] != "Charging": | |
vehicles[0].command('START_CHARGE') | |
vehicles[0].command('CHARGING_AMPS', charging_amps=minrate) | |
vehicles[0].command('CHARGING_AMPS', charging_amps=minrate) | |
charging = True | |
except teslapy.VehicleError as e: | |
print("Won't start charging", e) | |
if charging: # Display stats every % charge | |
if charge != chargedata['battery_level']: | |
charge = chargedata['battery_level'] | |
print("\nLevel:", | |
chargedata['battery_level'], "%, Limit", | |
chargedata['charge_limit_soc'], "%, Rate is", | |
chargedata['charge_amps'], "of a possible", | |
chargedata['charge_current_request_max'], "Amps,", | |
chargedata['charger_voltage'], "Volts, ", | |
chargedata['time_to_full_charge'], "Hours remaining\n") | |
printmsg(" Wait a minute...") | |
# /c changed to asyncio sleep | |
await asyncio.sleep(60) # The fastest the Sense API will update | |
continue | |
async def main(): | |
# Get config | |
config = open('config.yml', 'r') | |
# Create controller, with config | |
controller = SenseLink(config) | |
# Create instances | |
controller.create_instances() | |
# Get Mutable controller object, and create task to update it | |
mutable_plug = controller.plug_for_mac("50:c7:bf:f6:4f:39") | |
# Pass plug to TesSense, where TesSense can update it | |
tes_task = run_tes_sense(mutable_plug) | |
# Get SenseLink tasks to add these | |
tasks = controller.tasks | |
tasks.add(tes_task) | |
tasks.add(controller.server_start()) | |
logging.info("Starting SenseLink controller") | |
await asyncio.gather(*tasks) | |
if __name__ == "__main__": | |
try: | |
asyncio.run(main()) | |
except KeyboardInterrupt: | |
logging.info("Interrupt received, stopping SenseLink") |
Thanks, I downloaded this and since this code is calling other code I went to the full SenseLink folder and grabbed all the other files there. I had to change the TesSenseLink and config.yml files to have my already used MAC address. I ran the script and fixed a bunch of errors in modules having a period in front of their name (perhaps they should be in another folder?
This seemed to get the app running but I get an error trying to assign the current watts to the plug:
File "/Users/israndy/Documents/Xcode/Python3/SenseApps/SenseLink/tessenselink.py", line 145, in run_tes_sense
mutable_plug.power = charge_power
AttributeError: can't set attribute
@israndy Sorry that was a mistake on my part, that I didn't catch due to what I'd simplified for testing. Check the newest revision, essentially the call needs to be mutable_plug.data_source.power = …
Thx, trying it now
OK, so I love the ability to send this info to Sense, all is working great, but I had to download your entire SenseLink directory to get it working. I have now gotten SenseLink from PyPi installed and want to use that copy so I don't need all the supporting files around the TesSenseLink.py file but I am getting an error on the line "from SenseLink import *" that there is no module named SenseLink.
requests-oauthlib 1.3.0
sense-energy 0.10.2
SenseLink 2.0.1
setuptools 49.2.1
I thought I got it working with the sense_energy module, but it failed at the create instance call and I looked at the code and it looks like that's not in the sense_energy's SenseLink code.
Give the code in the pull request a shot - I got it all working there, and having it on a branch from your TesSense code should help merge in any changes you've made since that published version. Specifically, here's the full working file: https://github.com/israndy/TesSense/blob/3d636a8d7b9d7e9c90c247d61a33616f2f8d4152/TesSense.py
The import statement when using the PyPI-installed package should be: from senselink import SenseLink
. I get your annoyance with the Python import statements though, I can't ever get them right.
The sense_energy module does not use the same code. He forked that from my original SenseLink a while back, and it's more aimed at working hand in hand with HomeAssistant than as a standalone (in my impression).
Good stuff!
Yeah I can see how the broad config option is less useful in your case. While I might eventually change this up to make it a little more streamlined, you can replace the config file by doing something the following - using my module_usage_exampe.py
code as an example below, but it's highly similar to your latest TesSense code:
Original:
# pip install
from senselink import SenseLink
...
async def main():
# Get config
config = open('config.yml', 'r')
# Create controller, with config
controller = SenseLink(config)
# Create instances
controller.create_instances()
# Get Mutable controller object, and create task to update it
mutable_plug = controller.plug_for_mac("50:c7:bf:f6:4f:39")
Change to:
from senselink import SenseLink
from senselink.plug_instance import PlugInstance
from senselink.data_source import MutableSource
...
async def main():
# Create controller, with NO config
controller = SenseLink(None)
# Create a PlugInstance, setting at least the identifier and MAC
mutable_plug = PlugInstance("mutable",mac="50:c7:bf:f6:4f:39")
# Create and assign a Mutable Data Source to that plug
mutable_data_source = MutableSource("mutable", None)
mutable_plug.data_source = mutable_data_source
# Add that plug to the controller
controller.add_instances({mutable_plug.identifier: mutable_plug})
# Note: you can then use the mutable_plug variable directly instead of needing a controller.plug_for_mac() call.
# They are the same object.
Just a heads up though, I see some areas for improvement around this, so it could change in future releases of SenseLink!
Yeah sure thing! It's definitely helped me make some improvements to SenseLink.
As for it waking up the car, there is a way to query the Tesla API in such a way that it doesn't wake up the car, but I think it might have to be via the streaming API. Unfortunately it seems like the Teslapy docs on streaming are a little light/nonexistent. I gather that if you use the streaming API it doesn't wake the car up, but you can tell if it is awake you can then query it like normal. I have telsamate set up for myself, which I can verify doesn't wake the car up.
Comments I added to supply some extra info are prefixed by
#/c
, just to make them identifiable.See this revision (between Rev 2 and Rev 3) for the major changes. The Rev 1 to Rev 2 change is just to indent the majority of the code once, so that adding the async function wrapper doesn't just make it one huge singular change block. Rev 1 is the original TesSense.py code.