Skip to content

enrich irve socket max output #2571

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 2 commits into
base: dev
Choose a base branch
from
Open

enrich irve socket max output #2571

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e60258e
enrich irve socket max output
blemoineovh Aug 5, 2025
0ffd8c2
charging station: add tests to distinguish valid value in enumeration
blemoineovh Aug 11, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
296 changes: 252 additions & 44 deletions analysers/analyser_merge_charging_station_FR.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,11 +22,20 @@
###########################################################################

from modules.OsmoseTranslation import T_
from .Analyser_Merge import Analyser_Merge_Point, Source, CSV, Load_XY, Conflate, Select, Mapping
from .Analyser_Merge import (
Analyser_Merge_Point,
Source,
CSV,
Load_XY,
Conflate,
Select,
Mapping,
)


class Analyser_Merge_Charging_station_FR(Analyser_Merge_Point):

# Constante pour limiter les valeurs aberrantes (en kW)
MAX_POWER_KW = 401
WIKIDATA_MAP = {
"ionity": "Q42717773",
"bouygues": "Q3046208",
Expand All @@ -38,71 +47,270 @@ class Analyser_Merge_Charging_station_FR(Analyser_Merge_Point):
"Izivia": "Q86671322",
}

@staticmethod
def keepMaxValueIfEnum(str_val):

# si la valeur contient un ; on sépare et on prend la plus haute valeur
if ";" in str_val:
boom = str_val.split(";")
max_val = 0
for p in boom:
# Supprimer les unités éventuelles (comme "kw")
p_clean = p.lower().replace("kw", "").strip()
try:
p_val = int(float(p_clean))
# Ignorer les valeurs supérieures à la limite maximale connue (MAX_POWER_KW)
if (
p_val <= Analyser_Merge_Charging_station_FR.MAX_POWER_KW
and p_val > max_val
):
max_val = p_val
except ValueError:
# Ignorer les valeurs qui ne peuvent pas être converties en nombre
pass

if max_val > 0:
str_val = str(max_val)
else:
# Gérer les cas où il n'y a pas de délimiteur mais une unité est présente
if "kw" in str_val.lower():
p_clean = str_val.lower().replace("kw", "").strip()
try:
p_val = int(float(p_clean))
if p_val <= Analyser_Merge_Charging_station_FR.MAX_POWER_KW:
str_val = str(p_val)
except ValueError:
pass
return str_val

def _normalize_number(self, v: float) -> str:
"""Formate un float en supprimant les .0 inutiles et les zéros de fin."""
try:
iv = int(v)
if abs(v - iv) < 1e-9:
return str(iv)
return f"{v:.6f}".rstrip("0").rstrip(".")
except Exception:
return str(v)

def getPuissanceNominaleInKw(self, raw):
"""
Calcule la puissance nominale en kW à partir d'une valeur brute.

Règles:
- Si 'raw' est une énumération 'a;b;c', on garde la valeur max en kW <= MAX_POWER_KW.
- Si 'raw' est une valeur unique:
* suffixe 'kW' => valeur déjà en kW
* suffixe 'W' => conversion W -> kW
* sans unité:
- si > MAX_POWER_KW => on suppose des watts, on divise par 1000
- sinon => déjà en kW

Retour: 'X kW' ou None si indéterminable.
"""
if raw is None:
return None

s = str(raw).strip()
if s == "":
return None

max_kw = getattr(self, "MAX_POWER_KW", 500)

# Cas énumération: on délègue à keepMaxValueIfEnum puis on formate.
if ";" in s:
max_str = self.keepMaxValueIfEnum(s)
if not max_str:
return None
try:
v = float(
str(max_str).replace(",", ".").lower().replace("kw", "").strip()
)
except Exception:
return None
return f"{self._normalize_number(v)} kW"

# Cas valeur unique
s_norm = s.lower().replace(",", ".")
try:
if s_norm.endswith("kw"):
v = float(s_norm[:-2].strip())
return f"{self._normalize_number(v)} kW"

if s_norm.endswith("w"):
v_w = float(s_norm[:-1].strip())
v = v_w / 1000.0
return f"{self._normalize_number(v)} kW"

# Sans unité
v = float(s_norm)
if v > max_kw:
# On suppose des watts => conversion en kW
v = v / 1000.0
return f"{self._normalize_number(v)} kW"
except Exception:
return None

def __init__(self, config, logger=None):
Analyser_Merge_Point.__init__(self, config, logger)
doc = dict(
detail = T_(
'''A car charging station may be here but is not mapped. The list of the
charging stations comes from a database published by Etalab. This database
brings together information published by the various local authorities and
networks in France.'''),
fix = T_(
'''See [the
mapping](https://wiki.openstreetmap.org/wiki/France/data.gouv.fr/Bornes_de_Recharge_pour_V%C3%A9hicules_%C3%89lectriques)
on the wiki. Add a node or add tags if already existing.'''),
trap = T_(
'''The initial information corresponds to recharging pools and devices and not to
stations, so some values are worth checking in the field. For instance, an open data point
with `capacity=6` can sometimes match to 3 charging station with `capacity=2`'''))
self.def_class_missing_official(item = 8410, id = 1, level = 3, tags = ['merge', 'fix:imagery', 'fix:survey', 'fix:picture'],
title = T_('Car charging station not integrated'), **doc)
self.def_class_possible_merge(item = 8411, id = 3, level = 3, tags = ['merge', 'fix:imagery', 'fix:survey', 'fix:picture'],
title = T_('Car charging station, integration suggestion'), **doc)
self.def_class_update_official(item = 8412, id = 4, level = 3, tags = ['merge', 'fix:imagery', 'fix:survey', 'fix:picture'],
title = T_('Car charging station update'), **doc)
detail=T_(
"""A car charging station may be here but is not mapped. The list of the
charging stations comes from a database published by Etalab. This database
brings together information published by the various local authorities and
networks in France."""
),
fix=T_(
"""See [the
mapping](https://wiki.openstreetmap.org/wiki/France/data.gouv.fr/Bornes_de_Recharge_pour_V%C3%A9hicules_%C3%89lectriques)
on the wiki. Add a node or add tags if already existing."""
),
trap=T_(
"""The initial information corresponds to recharging pools and devices and not to
stations, so some values are worth checking in the field. For instance, an open data point
with `capacity=6` can sometimes match to 3 charging station with `capacity=2`"""
),
)
self.def_class_missing_official(
item=8410,
id=1,
level=3,
tags=["merge", "fix:imagery", "fix:survey", "fix:picture"],
title=T_("Car charging station not integrated"),
**doc,
)
self.def_class_possible_merge(
item=8411,
id=3,
level=3,
tags=["merge", "fix:imagery", "fix:survey", "fix:picture"],
title=T_("Car charging station, integration suggestion"),
**doc,
)
self.def_class_update_official(
item=8412,
id=4,
level=3,
tags=["merge", "fix:imagery", "fix:survey", "fix:picture"],
title=T_("Car charging station update"),
**doc,
)

self.init(
"https://transport.data.gouv.fr/datasets/fichier-consolide-des-bornes-de-recharge-pour-vehicules-electriques/",
"Stations de recharge pour véhicules électriques",
CSV(Source(attribution="data.gouv.fr:Etalab", millesime="05/2022",
fileUrl="https://raw.githubusercontent.com/Jungle-Bus/ref-EU-EVSE/gh-pages/opendata_stations.csv")),
CSV(
Source(
attribution="data.gouv.fr:Etalab",
millesime="05/2022",
fileUrl="https://raw.githubusercontent.com/Jungle-Bus/ref-EU-EVSE/gh-pages/opendata_stations.csv",
)
),
Load_XY("Xlongitude", "Ylatitude"),
Conflate(
select=Select(
types=["nodes", "ways"],
tags={"amenity": "charging_station"}),
types=["nodes", "ways"], tags={"amenity": "charging_station"}
),
conflationDistance=100,
osmRef="ref:EU:EVSE",
mapping=Mapping(
static1={
"amenity": "charging_station",
"motorcar": "yes"},
static1={"amenity": "charging_station", "motorcar": "yes"},
static2={"source": self.source},
mapping1={
"operator": "nom_operateur",
"network": "nom_enseigne",
"owner": "nom_amenageur",
"ref:EU:EVSE": "id_station_itinerance"
"ref:EU:EVSE": "id_station_itinerance",
},
mapping2={
"charging_station:output": lambda fields: self.getPuissanceNominaleInKw(
fields["puissance_nominale"]
),
"operator:phone": "telephone_operateur",
"operator:email": "contact_operateur",
"start_date": "date_mise_en_service",
"capacity": "nbre_pdc",
"bicycle": lambda fields: "yes" if fields["station_deux_roues"] == "true" else None,
"motorcycle": lambda fields: "yes" if fields["station_deux_roues"] == "true" else None,
"moped": lambda fields: "yes" if fields["station_deux_roues"] == "true" else None,
"motorcar": lambda fields: "no" if fields["station_deux_roues"] == "true" else "yes",
"bicycle": lambda fields: (
"yes" if fields["station_deux_roues"] == "true" else None
),
"motorcycle": lambda fields: (
"yes" if fields["station_deux_roues"] == "true" else None
),
"moped": lambda fields: (
"yes" if fields["station_deux_roues"] == "true" else None
),
"motorcar": lambda fields: (
"no" if fields["station_deux_roues"] == "true" else "yes"
),
"opening_hours": "horaires_grouped",
"fee": lambda fields: "yes" if fields["gratuit_grouped"] == "false" else ("no" if fields["gratuit_grouped"] == "true" else None),
"authentication:none": lambda fields: "yes" if fields["paiement_acte_grouped"] == "true" else None,
"payment:credit_cards": lambda fields: "yes" if fields["paiement_cb_grouped"] == "true" else ("no" if fields["paiement_cb_grouped"] == "false" else None),
"reservation": lambda fields: "yes" if fields["reservation_grouped"] == "true" else None,
"wheelchair": lambda fields: "yes" if fields["accessibilite_pmr_grouped"] == "Accessible mais non réservé PMR" else ("no" if fields["accessibilite_pmr_grouped"] == "Non accessible" else None),
"socket:typee": lambda fields: fields["nb_EF_grouped"] if fields["nb_EF_grouped"] != "0" else None,
"socket:type2": lambda fields: fields["nb_T2_grouped"] if fields["nb_T2_grouped"] != "0" else None,
"socket:type2_combo": lambda fields: fields["nb_combo_ccs_grouped"] if fields["nb_combo_ccs_grouped"] != "0" else None,
"socket:chademo": lambda fields: fields["nb_chademo_grouped"] if fields["nb_chademo_grouped"] != "0" else None,
"wikimedia:network": lambda fields: self.WIKIDATA_MAP.get(fields["nom_enseigne"].lower(), None) if fields["nom_enseigne"] != "0" else None,
"fee": lambda fields: (
"yes"
if fields["gratuit_grouped"] == "false"
else ("no" if fields["gratuit_grouped"] == "true" else None)
),
"authentication:none": lambda fields: (
"yes" if fields["paiement_acte_grouped"] == "true" else None
),
"payment:credit_cards": lambda fields: (
"yes"
if fields["paiement_cb_grouped"] == "true"
else (
"no"
if fields["paiement_cb_grouped"] == "false"
else None
)
),
"reservation": lambda fields: (
"yes" if fields["reservation_grouped"] == "true" else None
),
"wheelchair": lambda fields: (
"yes"
if fields["accessibilite_pmr_grouped"]
== "Accessible mais non réservé PMR"
else (
"no"
if fields["accessibilite_pmr_grouped"]
== "Non accessible"
else None
)
),
"socket:typee": lambda fields: (
fields["nb_EF_grouped"]
if fields["nb_EF_grouped"] != "0"
else None
),
"socket:type2": lambda fields: (
fields["nb_T2_grouped"]
if fields["nb_T2_grouped"] != "0"
else None
),
"socket:type2_combo": lambda fields: (
fields["nb_combo_ccs_grouped"]
if fields["nb_combo_ccs_grouped"] != "0"
else None
),
"socket:chademo": lambda fields: (
fields["nb_chademo_grouped"]
if fields["nb_chademo_grouped"] != "0"
else None
),
"wikimedia:network": lambda fields: (
self.WIKIDATA_MAP.get(fields["nom_enseigne"].lower(), None)
if fields["nom_enseigne"] != "0"
else None
),
},
text=lambda tags, fields: {
"en": "{0}, {1}, {2}".format(
fields["nom_station"],
fields["adresse_station"],
(
fields["observations"]
if fields["observations"] != "null"
else "-"
),
)
},
text=lambda tags, fields: {"en": "{0}, {1}, {2}".format(fields["nom_station"], fields["adresse_station"], fields["observations"] if fields["observations"] != "null" else "-")})))
),
),
)
Loading