Add aggregated power per socket type

group_opendata_by_station.py

@@ -5,11 +5,21 @@
 import re
 
 from collections import Counter
+from enum import IntFlag, auto
+
+class Socket(IntFlag):
+    EF = auto()
+    T2 = auto()
+    CHADEMO = auto()
+    CCS = auto()
 
 station_list = {}
 station_attributes = [ 'nom_amenageur', 'siren_amenageur', 'contact_amenageur', 'nom_operateur', 'contact_operateur', 'telephone_operateur', 'nom_enseigne', 'id_station_itinerance', 'id_station_local', 'nom_station', 'implantation_station', 'code_insee_commune', 'nbre_pdc', 'station_deux_roues', 'raccordement', 'num_pdl', 'date_mise_en_service', 'observations', 'adresse_station' ]
 pdc_attributes = [ 'id_pdc_itinerance', 'id_pdc_local', 'puissance_nominale', 'prise_type_ef', 'prise_type_2', 'prise_type_combo_ccs', 'prise_type_chademo', 'prise_type_autre', 'gratuit', 'paiement_acte', 'paiement_cb', 'paiement_autre', 'tarification', 'condition_acces', 'reservation', 'accessibilite_pmr', 'restriction_gabarit', 'observations', 'date_maj', 'cable_t2_attache', 'datagouv_organization_or_owner', 'horaires' ]
+socket_attributes = { 'prise_type_ef': Socket.EF, 'prise_type_2': Socket.T2, 'prise_type_chademo': Socket.CHADEMO, 'prise_type_combo_ccs': Socket.CCS }
 
+errors = []
+power_stats = []
 wrong_ortho = {}
 
 with open('fixes_networks.csv', 'r') as csv_file:
@@ -57,6 +67,37 @@ def cleanPhoneNumber(phone):
     else:
         return None
 
+def compute_max_power_per_socket_type(station, errors):
+    """
+    Computes the aggregated max power per socket type accross all PDCs (PDLs) associated with the given station.
+    Sockets are limited to the max power rating for their type. This is a safe guess needed when a given PDC
+    has multiple sockets for a given nominal power rating:
+        - E/F: max 4 kw
+        - Type 2: max 43 kw
+        - Chademo: max 63 kw (in France -> Version 1 only)
+        - Combo CCS: currently unlimited
+    """
+    power_ef = power_t2 = power_chademo = power_ccs = 0
+    for pdc in station['pdc_list']:
+        socket_mask = sum([ flag for socket_attr, flag in socket_attributes.items() if stringBoolToInt(pdc[socket_attr])==1 ])
+        socket_mask = Socket(socket_mask)
+        power = float(pdc['puissance_nominale'])
+        if power >= 1000:
+            errors.append({"station_id" :  station['attributes']['id_station_itinerance'],
+                        "source": station['attributes']['source_grouped'],
+                        "error": "puissance nominale déclarée suspecte",
+                        "detail": "puissance: {}, prises: {}".format(pdc['puissance_nominale'], socket_mask.name)
+                        })
+            # Convert from W to kW (>2MW should not exist)
+            # FIXME: Probably not usefull anymore. Data looks fine.
+            if power_ccs >= 2000:
+                power_ccs /= 1000
+        power_ef = max(power_ef, min(4, power if Socket.EF in socket_mask else 0))
+        power_t2 = max(power_t2, min(43, power if Socket.T2 in socket_mask else 0))
+        power_chademo = max(power_chademo, min(63, power if Socket.CHADEMO in socket_mask else 0))
+        power_ccs = max(power_ccs, power if Socket.CCS in socket_mask else 0)
+    return (power_ef, power_t2, power_chademo, power_ccs)
+
 def stringBoolToInt(strbool):
     return 1 if strbool.lower() == 'true' else 0
 
@@ -256,13 +297,22 @@ def transformRef(refIti, refLoc):
                        "detail": "nb pdc: %s" % (len(station['pdc_list']))
                        })
 
+    power_grouped_values = compute_max_power_per_socket_type(station, errors)
+    power_stats.append(power_grouped_values)
+    power_props = ['power_ef_grouped', 'power_t2_grouped', 'power_chademo_grouped', 'power_ccs_grouped']
+    station['attributes'].update(zip(power_props, power_grouped_values))
+
+print("Computed power stats:")
+print("       EF  |   T2  | Chademo |  CCS  |")
+for power_set, count in Counter(power_stats).most_common():
+    print(" >    {:4.2f}   {:5.2f}    {:5.2f}   {:6.2f} : {} occurences".format(*power_set, count))
+
 print("{} stations".format(len(station_list)))
 
 print("{} points de charge avec des erreurs :".format(len(errors)))
 for error_type, error_count in Counter([elem['error'] for elem in errors]).items():
     print(" > {} : {} éléments".format(error_type, error_count))
 
-
 with open("output/opendata_errors.csv", 'w') as ofile:
     tt = csv.DictWriter(ofile, fieldnames=errors[0].keys())
     tt.writeheader()