Using a Python script to save data in HA

I warn immediately, this Retex is only for pedagogical purposes, I write it to share a methodology, namely that I have realized this feature differently since then, so I do not have this process in prod.

Integration to run the Python script

For a reason I never understood, the python scripting features built into HA are archi-mini, no import is possible, so I have to use an integration called pyscript.

The doc's here.

I pass the details to install it, it is very simple with HACS and everything is in the doc.

In configuration.yaml, I have these lines:

pyscript: allow_all_imports: true hass _is _global: true

I get a token

I open the Home Assistant user interface.

In the lower left corner, I click on my username (or profile icon).

I click on "Profile".

I scroll down to the ‘Long-term access tokens’ section.

I click on "Create Token" to generate a new token.

I give my token a name to make it easier for me to find it.

Once the token is created, I see its contents. I copy the value of the token, i.e. the alphanumeric character series, and paste it into my configuration (next paragraph)

Declaration of the platform that makes the link between Python and the sensor

In configuration.yaml, I add these lines:

rest_command: send_data_to_api: url: 'http://homeassistant.local:8123/api/states/sensor.wallbox _charges' method: 'post' content_type: 'application/json' headers: Authorization: 'Bearer TOKEN' # Replace TOKEN with the token cf previous paragraph payload: '{{ data | tojson }}'

This is the front door that will be supplied by the Python script (below) and will send the information to the sensor.wallbox _charges.

My Python script

I place the _charges.py wallbox file in the /config/pyscript folder

import datetime import pytz @service def wallbox _charges(lastdata=None, action=None, price="123", km="123", dateStart="01/01/2000 12:00:00", value="123"): _time _paris = pytz.timezone("Europe/Paris") now _timestamp = int(datetime.datetime.now().timestamp()) now _time _paris = datetime.datetime.fromtimestamp(now _timestamp, _time _paris) now _formatee = now _time _paris.strftime("%d/%m/%Y %H:%M:%S") nowMoins20min = datetime.datetime.now() - datetime.timedelta(minutes=20) nowMoins20min _formatee = nowMoins20min.strftime("%H:%M") dateDebutRecue = datetime.datetime.strptime(dateDebut, '%d/%m/%Y %H:%M:%S').replace(tzinfo=fuseau_heure_paris) debut_timestamp_formatee = dateDebutRecue.strftime("%d/%m/%Y %H:%M:%S") debut_timestamp_formatee_SansSecondes = dateDebutRecue.strftime("%d/%m/%Y %H:%M") # Turn strings into datetime objects now _dt = datetime.datetime.strptime(now _formatee, "%d/%m/%Y %H:%M:%S") debut_dt = datetime.datetime.strptime(debut_timestamp_formatee, "%d/%m/%Y %H:%M:%S") # Calculate the difference between timetime objectsDeCharge = now_dt - debut_dt - datetime.timedelta(minutes=20) # Format the duration in hours, minutes and seconds hours, rest = divmod(timeDeCharge.seconds, 3600) minutes, seconds = divmod(rest, 60) # Display duration as HH:MM:SS timeDeCharge_formatee = f"{hours:02}:{minutes:02}" # Write an information message in the newspaper #log.info(action) if action == "register": valueaadd = [{ "saved_at": now _timestamp, "beginning": debut_timestamp_formatee_SansSecondes, "end": nowLess20min _formatee, "duration": timeDeCharge_formatee, "consumption": value, "km": km, "cost": round(value * price, 2) }] if lastdata is not None: valuea add = lastdata + valuea add data = { "state": debut_timestamp_formatee_SansSecondes, "attributes": { "updated _at": now _dt.strftime("%d/%m/%Y %H:%M:%S"), "charge session": valueadd } } if action == "init": data = { "state": "on", "attributes": { "updated _at": ‘14/10/2023 16:42:39’, ‘sessionscharge’: [ { "consumption": 2.14, "Cost": 0.43, "duration": "00:27", "beginning": ‘19/09/2023 20:36’, ‘end’: "21:03", "km": 17, "saved _at": 1695150359 }, { "consumption": 23.29, "cost": 4.66, "duration": "04:19", "beginning": ‘21/09/2023 21:06’, ‘end’: "01:25", "km": 192, "saved _at": 1695323159 }, { "consumption": 7.68, "cost": 1.54, "duration": "01:23", "beginning": ‘22/09/2023 08:45’, ‘end’: "10:08", "km": 63, "saved_at": 1695409559 }, { "consumption": 23.80, "cost": 4.76, "duration": "04:26", "beginning": ‘23/09/2023 16:57’, ‘end’: "21:23", "km": 196, "saved _at": 1695495959 } ] } } # Call the REST service to send the data service _data = { "token": "xxxxxx", # Replace with the token "data": data } if action == "register" or action == "init": # Call the REST service asynchronously await hass.services.async_call('rest_command', 'send_data_to_api', _data service) # Write an error message in the log #log.error("End script")

Add two automations

One to initialise the sensor, the other to add data

Initialize data

alias: Initializes load sessions description: "" trigger: [] condition: [] action: - service: pyscript.wallbox _charges data: Action: init mode: single

Save data

alias: Saves a load session description: >- This automation is triggered when the sensor has not changed its value for 20 minutes and is not at 0 trigger: - platform: state entity _id: - sensor.wallbox _portal _added _range for: hours: 0 minutes: 20 seconds: 0 enabled: true condition: - condition: numeric _state entity _id: sensor.wallbox _portal _added _energy above: 0 - condition: not conditions: - condition: state entity _id: sensor.wallbox _charges state: >- {{ as_timestamp(states('sensor.wallbox _portal _added_range _min')) | int | timestamp _custom('%d/%m/%Y %H:%M:%S', true)}} enabled: false action: - service: pyscript.wallbox _charges data: Action: record value: "{{states('sensor.wallbox _portal _added_energy')}}" prize: "{{states('sensor.wallbox _portal _energy_price')}}" km: "{{states('sensor.wallbox _portal _added_range')}}" lastdata: "{{state _attr('sensor.wallbox _charges', 'sessionscharge')}}" dateStart: >- {{ as_timestamp(states('sensor.wallbox _portal _added_range _min')) | int | timestamp _custom('%d/%m/%Y %H:%M:%S', true)}} - service: variable.update _sensor data: replace _attributes: false value: >- {{ as_timestamp(states('sensor.wallbox _portal _added_range _min')) | int | timestamp _custom('%d/%m/%Y %H:%M:%S', true)}} target: entity_id: sensor.wallbox _charges enabled: false mode: single

To have the minimum value of the load, I use statistics to have the min value

This sensor is declared in sensors.yaml

- platform: statistics name: "Wallbox Portal Added Range Min" entity_id: sensor.wallbox _portal _added _range state _characteristic: datetime _value _min max_age: hours: 10

Result of storing information

Display card of all sessions

<div>Charging sessions</div> 
Last update: {{ state_attr('sensor.wallbox _charges','updated _at')}} <br><br>
{% set items = state_attr('sensor.wallbox _charges', 'sessionscharge') %}
<table width='100%'>
<tbody>
{% for i in range(0, items | count, 1) %}
<tr>
<td>{{ items[i].start }} to {{ items[i].fin }} ({{ items[i].duration }})</td>
<td>{{ items[i].consumption }} kWh</td>
<td>+{{ items[i].km }} km</td>
<td>{{ items[i].cost }} €</td>
</tr>

{% endfor %}
</tbody>
</table>

Display card with totals by year, month, week

Type: markdown content: &gt;-
  <div><h1>Charging sessions</div>  {%- set default _language = 'en' %} Latest update: {{ state_attr('sensor.wallbox _charges','updated _at')}} <br><br>


  {% set items = state_attr('sensor.wallbox _charges', 'sessionscharge') %} {% set septjours = now() - timedelta(days=7) %} {% set power_semaine = namespace(value=0) %} {% set cout_semaine = namespace(value=0) %} {% set unmois = now() - timedelta(days=30) %} {% set power_months = namespace(value=0) %} {% set cout_mois = namespace(value=0) %} {% set uneannee = now() - timedelta(days=365) %} {% set power_annee = namespace(value=0) %} {% set cout_annee = namespace(value=0) %} {% for i in range(0, items | count, 1) %} {% if items[i].saved_at | int &gt;= as_timestamp(uneannee) %} {% set power_annee.value = power_annee.value + items[i].consumption %} {% set cout_annee.value = cout_annee.value + items[i].cost %} {% if items[i].saved_at | int &gt;= as_timestamp(unmonth) %} {% set power_mois.value = power_mois.value + items[i].consumption %} {% set cout_mois.value = cout_mois.value + items[i].cost %} {% if items[i].saved_at | int &gt;= as_timestamp(sevendays) %} {% set power_semaine.value = power_semaine.value + items[i].consumption %} {% set cout_semaine.value = cout_semaine.value + items[i].cost %} {% endif %} {% endif %} {% endif %} {% endfor %}

  </ha-alert> <ha-alert alert-type="info" title="Last month: {{ power _mois.value | int }} kWh - {{ cost _mois.value | float | round(2) }} €"></ha-alert> <ha-alert alert-type="info" title="Last year: {{ power_annee.value | int }} kWh - {{ cout _annee.value | float | round(2) }} €"></ha-alert>


  <br> 


  {% set items = state_attr('sensor.wallbox _charges', 'sessionscharge') %} {% set septjours = now() - timedelta(days=7) %}

  <table width='100%'> <tbody>  {% for i in range(0, items | count, 1) %} {% if items[i].saved_at | int &gt;= as_timestamp(sevendays) %} {% set datetime _obj = strptime(items[i].start, '%d/%m/%Y %H:%M') %}

  <tr> <td><h5>{{items[i].start }} to {{ items[i].fin }} ({{ items[i].duration }})</td> <td>{{ items[i].consumption }} kWh</td> <td>+{{ items[i].km }} km</td> <td>{{ items[i].cost }} €</td> </tr> {% endif %} {% endfor %} </tbody> </table>
    <ha-alert alert-type="success" title="Last week: {{ power_semaine.value | int}} kWh - {{ cost_semaine.value | float | round(2) }}€">    
style: .: | ha-card { margin: 0px 0px 0px 0px; box-shadow: none; } ha-markdown: $: | h5 { font-size: 15px!important; font-weight: normal!important; padding: 0px 0px 0px 8px!important; border-left: 3px solid #429f46; }