session limits are rearranged according to rated limits

iso15118/evcc/comm_session_handler.py

@@ -190,9 +190,11 @@ def create_sap(self) -> Union[SupportedAppProtocolReq, None]:
             priority += 1
             app_protocol_entry = AppProtocol(
                 protocol_ns=protocol.ns.value,
-                major_version=2
-                if protocol in [Protocol.ISO_15118_2, Protocol.DIN_SPEC_70121]
-                else 1,
+                major_version=(
+                    2
+                    if protocol in [Protocol.ISO_15118_2, Protocol.DIN_SPEC_70121]
+                    else 1
+                ),
                 minor_version=0,
                 schema_id=schema_id,
                 priority=priority,

iso15118/evcc/controller/simulator.py

@@ -3,6 +3,7 @@
 retrieve data from the EV. The DummyEVController overrides all abstract methods from
 EVControllerInterface.
 """
+
 import logging
 import random
 from typing import List, Optional, Tuple, Union

iso15118/evcc/states/evcc_state.py

@@ -2,6 +2,7 @@
 This module contains the abstract class for an EVCC-specific state,
 which extends the state shared between the EVCC and SECC.
 """
+
 import logging
 import time
 from abc import ABC

iso15118/evcc/states/iso15118_20_states.py

@@ -455,9 +455,9 @@ async def process_message(
             service_discovery_res.service_renegotiation_supported
         )
 
-        req_energy_services: List[
-            ServiceV20
-        ] = await self.comm_session.ev_controller.get_supported_energy_services()
+        req_energy_services: List[ServiceV20] = (
+            await self.comm_session.ev_controller.get_supported_energy_services()
+        )
 
         for energy_service in service_discovery_res.energy_service_list.services:
             for requested_energy_service in req_energy_services:

iso15118/evcc/states/iso15118_2_states.py

@@ -186,9 +186,9 @@ async def process_message(
         await self.select_energy_transfer_mode()
 
         charge_service: ChargeService = service_discovery_res.charge_service
-        offered_energy_modes: List[
-            EnergyTransferModeEnum
-        ] = charge_service.supported_energy_transfer_mode.energy_modes
+        offered_energy_modes: List[EnergyTransferModeEnum] = (
+            charge_service.supported_energy_transfer_mode.energy_modes
+        )
 
         if self.comm_session.selected_energy_mode not in offered_energy_modes:
             self.stop_state_machine(

iso15118/secc/comm_session_handler.py

@@ -201,9 +201,9 @@ def __init__(
         # The comm_sessions dict keys are of type str (the IPv6 address), the
         # values are a tuple containing the SECCCommunicationSession and the
         # associated ayncio.Task object (so we can cancel the task when needed)
-        self.comm_sessions: Dict[
-            str, Tuple[SECCCommunicationSession, asyncio.Task]
-        ] = {}
+        self.comm_sessions: Dict[str, Tuple[SECCCommunicationSession, asyncio.Task]] = (
+            {}
+        )
 
     async def start_session_handler(
         self, iface: str, start_udp_server: Optional[bool] = True

iso15118/secc/controller/ev_data.py

@@ -218,9 +218,9 @@ def __init__(
         self.max_soc: Optional[int] = max_soc  # 0-100
         self.max_v2x_energy_request: Optional[float] = max_v2x_energy_request
         self.min_v2x_energy_request: Optional[float] = min_v2x_energy_request
-        self.remaining_time_to_target_soc: Optional[
-            float
-        ] = remaining_time_to_target_soc  # noqa: E501
+        self.remaining_time_to_target_soc: Optional[float] = (
+            remaining_time_to_target_soc  # noqa: E501
+        )
         # In -2 is FullSOC
         self.remaining_time_to_max_soc: Optional[float] = remaining_time_to_max_soc
         self.remaining_time_to_min_soc: Optional[float] = remaining_time_to_min_soc

iso15118/secc/controller/evse_data.py

@@ -173,21 +173,21 @@ def __init__(
         self.max_power_asymmetry: Optional[float] = max_power_asymmetry
 
         #  Optional in 15118-2 CPD
-        self.current_regulation_tolerance: Optional[
-            float
-        ] = current_regulation_tolerance
+        self.current_regulation_tolerance: Optional[float] = (
+            current_regulation_tolerance
+        )
         self.peak_current_ripple: Optional[float] = peak_current_ripple
         self.energy_to_be_delivered: Optional[float] = energy_to_be_delivered
         # Metering
-        self.present_active_power: Optional[
-            float
-        ] = present_active_power  # Optional in AC Scheduled CL
-        self.present_active_power_l2: Optional[
-            float
-        ] = present_active_power_l2  # Optional in AC Scheduled CL
-        self.present_active_power_l3: Optional[
-            float
-        ] = present_active_power_l3  # Optional in AC Scheduled CL
+        self.present_active_power: Optional[float] = (
+            present_active_power  # Optional in AC Scheduled CL
+        )
+        self.present_active_power_l2: Optional[float] = (
+            present_active_power_l2  # Optional in AC Scheduled CL
+        )
+        self.present_active_power_l3: Optional[float] = (
+            present_active_power_l3  # Optional in AC Scheduled CL
+        )
 
         # Required for -2 DC CurrentDemand, -20 DC CL
         self.present_current: Union[float, int] = present_current
@@ -218,28 +218,44 @@ def update_ac_charge_parameters_v2(
     ) -> None:
         """Update the EVSE data context with the AC charge parameters."""
         self.current_type = CurrentType.AC
-        rated_limits = self.rated_limits.ac_limits = EVSEACCPDLimits()
-        self.session_limits.ac_limits = EVSEACCLLimits()
+        if self.rated_limits.ac_limits is None:
+            self.rated_limits.ac_limits = EVSEACCPDLimits()
+        rated_limits = self.rated_limits.ac_limits
+        if self.session_limits.ac_limits is None:
+            self.session_limits.ac_limits = EVSEACCLLimits()
+        session_limits = self.session_limits.ac_limits
         self.nominal_voltage = (
             ac_charge_parameter.evse_nominal_voltage.get_decimal_value()
         )  # noqa: E501
         rated_limits.max_current = (
             ac_charge_parameter.evse_max_current.get_decimal_value()
         )
+
         rated_limits.max_charge_power = rated_limits.max_current * self.nominal_voltage
+        rated_limits.max_charge_power_l2 = rated_limits.max_charge_power
+        rated_limits.max_charge_power_l3 = rated_limits.max_charge_power
         rated_limits.max_discharge_power = 0
         rated_limits.min_charge_power = 0
         rated_limits.min_discharge_power = 0
         # Create the session limits based on the rated limits
-        self.session_limits.ac_limits.update(rated_limits.as_dict())
+        # without exceeding the rated limits
+        for value in vars(rated_limits):
+            if hasattr(session_limits, value):
+                rated_value = getattr(rated_limits, value)
+                session_value = getattr(session_limits, value)
+                if not session_value or (session_value > rated_value):
+                    setattr(session_limits, value, rated_value)
 
     def update_dc_charge_parameters(
         self, dc_charge_parameter: DCEVSEChargeParameter
     ) -> None:
         """Update the EVSE data context with the DC charge parameters."""
         self.current_type = CurrentType.DC
-        rated_limits = self.rated_limits.dc_limits = EVSEDCCPDLimits()
-        self.session_limits.dc_limits = EVSEDCCLLimits()
+        if not self.rated_limits.dc_limits:
+            self.rated_limits.dc_limits = EVSEDCCPDLimits()
+        rated_limits = self.rated_limits.dc_limits
+        if not self.session_limits.dc_limits:
+            self.session_limits.dc_limits = EVSEDCCLLimits()
         rated_limits.max_charge_power = (
             dc_charge_parameter.evse_maximum_power_limit.get_decimal_value()
         )
@@ -264,8 +280,14 @@ def update_dc_charge_parameters(
             self.energy_to_be_delivered = (
                 dc_charge_parameter.evse_energy_to_be_delivered.get_decimal_value()
             )
-        # Create the session limits based on the rated limits
-        self.session_limits.dc_limits.update(rated_limits.as_dict())
+            # Create the session limits based on the rated limits
+            # without exceeding the rated limits
+            for value in vars(rated_limits):
+                if hasattr(self.session_limits.dc_limits, value):
+                    rated_value = getattr(rated_limits, value)
+                    session_value = getattr(self.session_limits.dc_limits, value)
+                    if not session_value or (session_value > rated_value):
+                        setattr(self.session_limits.dc_limits, value, rated_value)
 
     def update_ac_charge_parameters_v20(
         self,

iso15118/secc/controller/interface.py

@@ -2,6 +2,7 @@
 This module contains the abstract class for an SECC to retrieve data from the EVSE
 (Electric Vehicle Supply Equipment).
 """
+
 import logging
 from abc import ABC, abstractmethod
 from dataclasses import dataclass

iso15118/secc/controller/simulator.py

@@ -2,6 +2,7 @@
 This module contains the code to retrieve (hardware-related) data from the EVSE
 (Electric Vehicle Supply Equipment).
 """
+
 import base64
 import logging
 import time
@@ -316,7 +317,7 @@ async def get_scheduled_se_params(
         """Overrides EVSEControllerInterface.get_scheduled_se_params()."""
         charging_power_schedule_entry = PowerScheduleEntry(
             duration=3600,
-            power=RationalNumber(exponent=3, value=10)
+            power=RationalNumber(exponent=3, value=10),
             # Check if AC ThreePhase applies (Connector parameter within parameter set
             # of SelectedEnergyService) if you want to add power_l2 and power_l3 values
         )
@@ -394,7 +395,7 @@ async def get_scheduled_se_params(
 
         discharging_power_schedule_entry = PowerScheduleEntry(
             duration=3600,
-            power=RationalNumber(exponent=3, value=10)
+            power=RationalNumber(exponent=3, value=10),
             # Check if AC ThreePhase applies (Connector parameter within parameter set
             # of SelectedEnergyService) if you want to add power_l2 and power_l3 values
         )

iso15118/secc/states/din_spec_states.py

@@ -507,9 +507,9 @@ async def process_message(
             # Any relays in the DC output circuit of the DC station shall
             # be closed during the insulation test
             # If None is returned, then contactor close operation is ongoing.
-            contactors_closed_for_cable_check: Optional[
-                bool
-            ] = await self.comm_session.evse_controller.is_contactor_closed()
+            contactors_closed_for_cable_check: Optional[bool] = (
+                await self.comm_session.evse_controller.is_contactor_closed()
+            )
 
             if contactors_closed_for_cable_check is not None:
                 if contactors_closed_for_cable_check:

iso15118/secc/states/iso15118_20_states.py

@@ -3,6 +3,7 @@
 V2GMessage objects of the ISO 15118-20 protocol, from SessionSetupReq to
 SessionStopReq.
 """
+
 import asyncio
 import logging
 import time
@@ -1427,9 +1428,9 @@ async def process_message(
         if ac_charge_loop_req.meter_info_requested:
             meter_info = await self.comm_session.evse_controller.get_meter_info_v20()
 
-        evse_status: Optional[
-            EVSEStatus
-        ] = await self.comm_session.evse_controller.get_evse_status()
+        evse_status: Optional[EVSEStatus] = (
+            await self.comm_session.evse_controller.get_evse_status()
+        )
 
         response_code = ResponseCode.OK
         params = None
@@ -1642,9 +1643,9 @@ async def process_message(
                 # Any relays in the DC output circuit of the DC station shall
                 # be closed during the insulation test
                 # If None is returned, then contactor close operation is ongoing.
-                contactors_closed_for_cable_check: Optional[
-                    bool
-                ] = await self.comm_session.evse_controller.is_contactor_closed()
+                contactors_closed_for_cable_check: Optional[bool] = (
+                    await self.comm_session.evse_controller.is_contactor_closed()
+                )
 
                 if contactors_closed_for_cable_check is not None:
                     if contactors_closed_for_cable_check:
@@ -1842,9 +1843,9 @@ async def _build_dc_charge_loop_res(
                 )
             )  # noqa
 
-        evse_status: Optional[
-            EVSEStatus
-        ] = await self.comm_session.evse_controller.get_evse_status()
+        evse_status: Optional[EVSEStatus] = (
+            await self.comm_session.evse_controller.get_evse_status()
+        )
 
         meter_info: Optional[MeterInfo] = None
         if meter_info_requested:

iso15118/secc/states/iso15118_2_states.py

@@ -3,6 +3,7 @@
 V2GMessage objects of the ISO 15118-2 protocol, from SessionSetupReq to
 SessionStopReq.
 """
+
 import asyncio
 import base64
 import logging
@@ -1357,9 +1358,9 @@ async def process_message(
             ev_data_context.selected_energy_mode.value.startswith("AC")
         )
 
-        max_schedule_entries: Optional[
-            int
-        ] = charge_params_req.max_entries_sa_schedule_tuple
+        max_schedule_entries: Optional[int] = (
+            charge_params_req.max_entries_sa_schedule_tuple
+        )
 
         ac_evse_charge_params: Optional[ACEVSEChargeParameter] = None
         dc_evse_charge_params: Optional[DCEVSEChargeParameter] = None
@@ -1471,9 +1472,9 @@ async def process_message(
 
         charge_params_res = ChargeParameterDiscoveryRes(
             response_code=ResponseCode.OK,
-            evse_processing=EVSEProcessing.FINISHED
-            if sa_schedule_list
-            else EVSEProcessing.ONGOING,
+            evse_processing=(
+                EVSEProcessing.FINISHED if sa_schedule_list else EVSEProcessing.ONGOING
+            ),
             sa_schedule_list=SAScheduleList(schedule_tuples=sa_schedule_list),
             ac_charge_parameter=ac_evse_charge_params,
             dc_charge_parameter=dc_evse_charge_params,

iso15118/secc/states/secc_state.py

@@ -68,9 +68,9 @@ class StateSECC(State, ABC):
     # The response code can be set by various methods on which a State's
     # process_message() method might rely on, such as is_message_valid().
     # The default response code 'OK' can be overwritten as needed.
-    response_code: Union[
-        ResponseCodeDINSPEC, ResponseCodeV2, ResponseCodeV20
-    ] = ResponseCodeV2.OK
+    response_code: Union[ResponseCodeDINSPEC, ResponseCodeV2, ResponseCodeV20] = (
+        ResponseCodeV2.OK
+    )
 
     def __init__(
         self, comm_session: "SECCCommunicationSession", timeout: Union[float, int] = 0

iso15118/shared/exi_codec.py

@@ -262,9 +262,7 @@ def to_exi(self, msg_element: BaseModel, protocol_ns: str) -> bytes:
 
         return exi_stream
 
-    def from_exi(
-        self, exi_message: bytes, namespace: str
-    ) -> Union[
+    def from_exi(self, exi_message: bytes, namespace: str) -> Union[
         SupportedAppProtocolReq,
         SupportedAppProtocolRes,
         V2GMessageV2,

iso15118/shared/messages/iso15118_2/body.py

@@ -9,6 +9,7 @@
 (or class) that matches the definitions in the XSD schema, including the XSD
 element names by using the 'alias' attribute.
 """
+
 import logging
 from abc import ABC
 from typing import Optional, Tuple, Type

iso15118/shared/messages/iso15118_2/msgdef.py

@@ -12,6 +12,7 @@
 (or class) that matches the definitions in the XSD schema, including the XSD
 element names by using the 'alias' attribute.
 """
+
 from pydantic import Field
 
 from iso15118.shared.messages import BaseModel

iso15118/shared/messages/iso15118_20/common_types.py

@@ -10,6 +10,7 @@
 (or class) that matches the definitions in the XSD schema, including the XSD
 element names by using the 'alias' attribute.
 """
+
 from abc import ABC
 from enum import Enum
 from typing import List, Optional, Union