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

Merge branch 'master' into develop

Browse Source
This commit is contained in:
Florian Bezannier
2021-06-16 18:12:51 +02:00
parent a36a8fe900 14e639ddb2
commit 6dd5156198
9 changed files with 67 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/main.yml
+1 -1
View File
@@ -33,4 +33,4 @@ jobs:
run: |
echo Test
coverage run -m unittest || exit 1
coverage xml -o cobertura.xml && bash <(curl -Ls https://coverage.codacy.com/get.sh) report -r cobertura.xml
[ -n "$CODACY_PROJECT_TOKEN" ] && coverage xml -o cobertura.xml && bash <(curl -Ls https://coverage.codacy.com/get.sh) report -r cobertura.xml
FAQ.md
+5
View File
@@ -33,3 +33,8 @@ To fix this go to the application directory and execute this command :
# if the user is launched by pi user do
sudo chown pi: -R .
```
### 5. I doesn't receive SMS
The SMS authentication is used to be able to remote control your car.
If your car doesn't have this functionality you should disable remote control when you start psa-car-controller
by using `--remote-disable` argument.
README.md
+3
View File
@@ -57,6 +57,9 @@ For information on configuring the psa_car_controller Docker container [see this
2.7 Start/Stop preconditioning
http://localhost:5000/preconditioning/YOURVIN/1 or 0
2.8 Change charge hour (for example: set it to 22h30)
http://127.0.0.1:5000/charge_hour?vin=YOURVIN&hour=22&min=30
## III. Use the dashboard
docs/Docker.md
+1 -1
View File
@@ -17,4 +17,4 @@ docker run -d -ti --name psa_car_controller1 \
psa_car_controller
```
Go to http://127.0.0.1:5000 and follow instruction
Go to http://127.0.0.1:5000 and follow instruction
ecomix.py
+4 -2
View File
@@ -5,6 +5,7 @@ import numbers
import requests
import reverse_geocode
from pytz import UTC
from mylogger import logger
@@ -51,12 +52,13 @@ class Ecomix:
def get_data_from_co2_signal(latitude, longitude, country_code_default):
if Ecomix.co2_signal_key is not None:
try:
now = datetime.utcnow().replace(tzinfo=UTC)
country_code = Ecomix.get_country(latitude, longitude, country_code_default)
assert country_code is not None
if country_code not in Ecomix._cache:
Ecomix._cache[country_code] = []
elif len(Ecomix._cache[country_code]) > 0 and \
(datetime.now() - Ecomix._cache[country_code][-1][0]).total_seconds() < CO2_SIGNAL_REQ_INTERVAL:
(now - Ecomix._cache[country_code][-1][0]).total_seconds() < CO2_SIGNAL_REQ_INTERVAL:
return False
res = requests.get(CO2_SIGNAL_URL + "/v1/latest",
headers={"auth-token": Ecomix.co2_signal_key},
@@ -64,7 +66,7 @@ class Ecomix:
data = res.json()
value = data["data"]["carbonIntensity"]
assert isinstance(value, numbers.Number)
Ecomix._cache[country_code].append([datetime.now(), value])
Ecomix._cache[country_code].append([now, value])
return data["status"] == "ok"
except (AssertionError, NameError, KeyError):
logger.debug("ecomix:", exc_info=True)
libs/car_model.py
+8 -1
View File
@@ -55,10 +55,17 @@ carmodels = [
ElecModel("e-2008", 46, "peugeot:e2008:20:48", r"VR3UKZKX.*"),
ElecModel("e-Spacetourer", 46, "peugeot:etraveler:21:50:citroen", r"VF7VZZKX.*"),
ElecModel("corsa-e", 46, "opel:corsae:20:50", r"VXKUHZKX.*"),
# Use corsa in abrp because Mokka isn't available
ElecModel("Mokka-e", 46, "opel:corsae:20:50", r"VXKUKZKX.*"), # VXKUKZKXZM
ElecModel("Zaphira-e", 68, "peugeot:etraveler:21:75:opel", r"VXEVZZKX.*"), # VXEVZZKXZMZ
ElecModel("E-C4", 46, "citroen:ec4:21:50", r"VR7BCZKX.*"), # VR7BCZKXCM
CarModel("SUV 3008", 10.8, 43),
CarModel("308", 0, 56, reg=r"VF3L35GG.*"),
CarModel("2008", 0, 44, reg=r"VR3USHNS.*"), # VR3USHNSKM
CarModel("SUV 5008 II", 0, 56, reg=r"VF3MRHNS.*"), # vf3mrhnsum
CarModel("SUV 5008 II 2018", 0, 56, reg=r"VF3MRHNY.*"), # VF3MRHNYHH
CarModel("C5 Aircross", 10.8, 43),
CarModel("DS7 Crossback E-Tense", 13.2, 43, reg="VR1J45GBUK.*")
CarModel("DS7 Crossback E-Tense", 11.5, 43, reg="VR1J45GBUK.*"),
CarModel("DS7 Crossback E-Tense 300 4x4", 11.5, 43, reg=" VR1J45GBUL.*"),
CarModel("508 SW Hybrid", 11.5, 45, reg=r"VR3F4DGZ.*") # VR3F4DGZTL
]
libs/utils.py
+19
View File
@@ -3,6 +3,7 @@ from threading import Semaphore, Timer
import socket
import requests
from typing import List
from mylogger import logger
@@ -47,3 +48,21 @@ def rate_limit(limit, every):
def is_port_in_use(ip, port):
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
return s.connect_ex((ip, port)) == 0
def parse_hour(s):
s = s[2:]
separators = ("H", "M", "S")
res: List[int] = []
for sep in separators:
if sep in s:
n, s = s.split(sep)
else:
n = 0
res.append(int(n))
if s.isnumeric():
res.append(int(s))
break
if len(res) == 2:
res.append(0)
return res
my_psacc.py
+20 -19
View File
@@ -1,5 +1,4 @@
import json
import re
import threading
import uuid
from datetime import datetime
@@ -22,7 +21,7 @@ from otp.otp import load_otp, new_otp_session, save_otp, ConfigException, Otp
from psa_connectedcar.rest import ApiException
from mylogger import logger
from libs.utils import rate_limit
from libs.utils import rate_limit, parse_hour
from web.abrp import Abrp
from web.db import Database
@@ -38,6 +37,12 @@ realm_info = {
"app_name": "MyVauxhall"}
}
MQTT_BRANDCODE = {"AP": "AP",
"AC": "AC",
"DS": "AC",
"VX": "OV",
"OP": "OV"
}
AUTHORIZE_SERVICE = "https://api.mpsa.com/api/connectedcar/v2/oauth/authorize"
REMOTE_URL = "https://api.groupe-psa.com/connectedcar/v4/virtualkey/remoteaccess/token?client_id="
@@ -263,10 +268,14 @@ class MyPSACC:
sleep(60)
return None
def __get_mqtt_customer_id(self):
brand_code = self.customer_id[:2]
return MQTT_BRANDCODE[brand_code]+self.customer_id[2:]
# pylint: disable=unused-argument
def __on_mqtt_connect(self, client, userdata, result_code, _):
logger.info("Connected with result code %s", result_code)
topics = [MQTT_RESP_TOPIC + self.customer_id + "/#"]
topics = [MQTT_RESP_TOPIC + self.__get_mqtt_customer_id() + "/#"]
for car in self.vehicles_list:
topics.append(MQTT_EVENT_TOPIC + car.vin)
for topic in topics:
@@ -333,24 +342,17 @@ class MyPSACC:
self.refresh_token()
date = datetime.utcnow()
date_str = date.strftime(PSA_DATE_FORMAT)
data = {"access_token": self.remote_access_token, "customer_id": self.customer_id,
data = {"access_token": self.remote_access_token, "customer_id": self.__get_mqtt_customer_id(),
"correlation_id": gen_correlation_id(date), "req_date": date_str, "vin": vin,
"req_parameters": req_parameters}
return json.dumps(data)
def __get_charge_hour(self, vin):
reg = r"PT([0-9]{1,2})H([0-9]{1,2})?"
data = self.get_vehicle_info(vin)
hour_str = data.get_energy('Electric').charging.next_delayed_time
try:
hour_minute = re.findall(reg, hour_str)[0]
hour = int(hour_minute[0])
if hour_minute[1] == '':
minute = 0
else:
minute = hour_minute[1]
return hour, minute
return parse_hour(hour_str)[:2]
except IndexError:
logger.exception("Can't get charge hour: %s", hour_str)
return None
@@ -363,7 +365,7 @@ class MyPSACC:
def __veh_charge_request(self, vin, hour, minute, charge_type):
msg = self.mqtt_request(vin, {"program": {"hour": hour, "minute": minute}, "type": charge_type})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge", msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.__get_mqtt_customer_id() + "/VehCharge", msg)
def change_charge_hour(self, vin, hour, miinute):
self.__veh_charge_request(vin, hour, miinute, "delayed")
@@ -381,19 +383,18 @@ class MyPSACC:
def horn(self, vin, count):
msg = self.mqtt_request(vin, {"nb_horn": count, "action": "activate"})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Horn", msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.__get_mqtt_customer_id() + "/Horn", msg)
def lights(self, vin, duration: int):
msg = self.mqtt_request(vin, {"action": "activate", "duration": duration})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Lights", msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.__get_mqtt_customer_id() + "/Lights", msg)
@rate_limit(3, 60 * 20)
def wakeup(self, vin):
logger.info("ask wakeup to %s", vin)
msg = self.mqtt_request(vin, {"action": "state"})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/VehCharge/state", msg)
return True
def lock_door(self, vin, lock: bool):
@@ -404,7 +405,7 @@ class MyPSACC:
msg = self.mqtt_request(vin, {"action": value})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/Doors", msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.__get_mqtt_customer_id() + "/Doors", msg)
return True
def preconditioning(self, vin, activate: bool):
@@ -423,7 +424,7 @@ class MyPSACC:
}
msg = self.mqtt_request(vin, {"asap": value, "programs": programs})
logger.info(msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.customer_id + "/ThermalPrecond", msg)
self.mqtt_client.publish(MQTT_REQ_TOPIC + self.__get_mqtt_customer_id() + "/ThermalPrecond", msg)
return True
def save_config(self, name=None, force=False):
@@ -436,7 +437,7 @@ class MyPSACC:
f.write(config_str)
self._config_hash = new_hash
logger.info("save config change")
# disconnect
@staticmethod
def load_config(name="config.json"):
with open(name, "r") as f:
test/test_unit.py
+6 -1
View File
@@ -20,7 +20,7 @@ from charge_control import ChargeControls
from test.utils import DATA_DIR, record_position, latitude, longitude, date0, date1, date2, date3, record_charging, \
vehicule_list, get_new_test_db
from trip import Trips
from libs.utils import get_temp
from libs.utils import get_temp, parse_hour
from web.db import Database
from web.figures import get_figures, get_battery_curve_fig, get_altitude_fig
from deepdiff import DeepDiff
@@ -217,6 +217,11 @@ class TestUnit(unittest.TestCase):
Database.record_position(None, "xx", 11, latitude, longitude - 0.05, None, date0, 40, None, False)
assert old_dummy_value != dummy_value
def test_parse_hour(self):
expected_res = [[2, 0, 0], [3, 14, 0], [0, 0, 2], [0, 30, 0]]
assert expected_res == [parse_hour(h) for h in ["PT2H", "PT3H14", "PT2S", "PT30M"]]
if __name__ == '__main__':
my_logger(handler_level=os.environ.get("DEBUG_LEVEL", 20))
unittest.main()