github/psa_car_controller
1
0
Fork 0
mirror of https://github.com/flobz/psa_car_controller.git synced 2026-08-26 10:17:18 +00:00
Code Issues Packages Projects Releases Wiki Activity

code clean up and reduce algo complexity

Browse Source
This commit is contained in:
Florian Bezannier
2021-03-21 19:31:26 +01:00
parent 25c61433b8
commit c3adb194a3
14 changed files with 387 additions and 319 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChargeControl.py
+55 -45
View File
@@ -11,9 +11,14 @@ import pytz
from MyPSACC import MyPSACC
from MyLogger import logger
DISCONNECTED = "Disconnected"
INPROGRESS = "InProgress"
FAILURE = "Failure"
STOPPED = "Stopped"
FINISHED = "Finished"
class ChargeControl:
periodicity = 120
MQTT_TIMEOUT = 60
def __init__(self, psacc: MyPSACC, vin, percentage_threshold, stop_hour):
@@ -22,7 +27,6 @@ class ChargeControl:
self.set_stop_hour(stop_hour)
self.psacc = psacc
self.retry_count = 0
self.always_check = True
def set_stop_hour(self, stop_hour):
if stop_hour is None or stop_hour == [0, 0]:
@@ -37,47 +41,53 @@ class ChargeControl:
def get_stop_hour(self):
return self._stop_hour
def control_charge_with_ack(self, charge: bool):
self.psacc.charge_now(self.vin, charge)
self.retry_count += 1
sleep(ChargeControl.MQTT_TIMEOUT)
vehicle_status = self.psacc.get_vehicle_info(self.vin)
status = vehicle_status.get_energy('Electric').charging.status
if status in (FINISHED, DISCONNECTED):
logger.warning("Car state isn't compatible with charging %s", status)
if (status == INPROGRESS) != charge:
logger.warning("retry to control the charge of %s", self.vin)
self.psacc.charge_now(self.vin, charge)
self.retry_count += 1
return False
self.retry_count = 0
return True
def force_update(self):
# force update if the car doesn't send info during 10 minutes
last_update = self.psacc.vehicles_list.get_car_by_vin(self.vin).get_status().get_energy('Electric').updated_at
if (datetime.utcnow().replace(tzinfo=pytz.UTC) - last_update).total_seconds() > 60 * 10:
self.psacc.wakeup(self.vin)
def process(self):
now = datetime.now()
vehicle_status = self.psacc.vehicles_list.get_car_by_vin(self.vin).status
try:
if self._next_stop_hour is not None and self._next_stop_hour < now:
stop_charge = True
self._next_stop_hour += timedelta(days=1)
logger.info("it's time to stop the charge")
else:
stop_charge = False
if self.percentage_threshold != 100 or stop_charge or self.always_check:
if vehicle_status is not None:
status = vehicle_status.get_energy('Electric').charging.status
level = vehicle_status.get_energy('Electric').level
logger.info("charging status of %s is %s, battery level: %d", self.vin, status, level)
if status == "InProgress":
# force update if the car doesn't send info during 10 minutes
last_update = vehicle_status.get_energy('Electric').updated_at
if (datetime.utcnow().replace(tzinfo=pytz.UTC) - last_update).total_seconds() > 60 * 10:
self.psacc.wakeup(self.vin)
if (level >= self.percentage_threshold and self.retry_count < 2) or stop_charge:
self.psacc.charge_now(self.vin, False)
self.retry_count += 1
sleep(ChargeControl.MQTT_TIMEOUT)
vehicle_status = self.psacc.get_vehicle_info(self.vin)
status = vehicle_status.get_energy('Electric').charging.status
if status == "InProgress":
logger.warning("retry to stop the charge of %s", self.vin)
self.psacc.charge_now(self.vin, False)
self.retry_count += 1
if self._next_stop_hour is not None:
next_in_second = (self._next_stop_hour - now).total_seconds()
if next_in_second < self.psacc.info_refresh_rate:
periodicity = next_in_second
self.thread = threading.Timer(periodicity, self.process, args=[vehicle_status])
self.thread.start()
vehicle_status = self.psacc.vehicles_list.get_car_by_vin(self.vin).get_status()
status = vehicle_status.get_energy('Electric').charging.status
level = vehicle_status.get_energy('Electric').level
logger.info("charging status of %s is %s, battery level: %d", self.vin, status, level)
if status == "InProgress":
self.force_update()
if level >= self.percentage_threshold and self.retry_count < 2:
self.control_charge_with_ack(False)
elif self._next_stop_hour is not None:
if self._next_stop_hour < now:
self._next_stop_hour += timedelta(days=1)
logger.info("it's time to stop the charge")
self.control_charge_with_ack(False)
else:
self.retry_count = 0
else:
logger.error("error can't retrieve vehicle info of %s", self.vin)
next_in_second = (self._next_stop_hour - now).total_seconds()
if next_in_second < self.psacc.info_refresh_rate:
periodicity = next_in_second
thread = threading.Timer(periodicity, self.process)
thread.setDaemon(True)
thread.start()
else:
self.retry_count = 0
except:
logger.error(traceback.format_exc())
@@ -95,10 +105,10 @@ class ChargeControls(dict):
def save_config(self, name="charge_config.json", force=False):
chd = {}
chargeControl: ChargeControl
for chargeControl in self.values():
chd[chargeControl.vin] = {"percentage_threshold": chargeControl.percentage_threshold,
"stop_hour": chargeControl.get_stop_hour()}
charge_control: ChargeControl
for charge_control in self.values():
chd[charge_control.vin] = {"percentage_threshold": charge_control.percentage_threshold,
"stop_hour": charge_control.get_stop_hour()}
config_str = json.dumps(chd, sort_keys=True, indent=4).encode('utf-8')
new_hash = md5(config_str).hexdigest()
if force or self._config_hash != new_hash:
@@ -121,8 +131,8 @@ class ChargeControls(dict):
try:
return self[vin]
except KeyError:
return None
pass
def start(self):
def init(self):
for charge_control in self.values():
charge_control.psacc.info_callback.append(charge_control.process)