wip

evrouting/charge/factories.py

-import networkx as nx
 from typing import Dict
 
+import networkx as nx
+
 from .T import SoCProfile, SoCFunction, ChargingFunction, Label
 from ..T import Node, SoC, Time
 from ..graph_tools import charging_cofficient, consumption
@@ -61,13 +62,13 @@ class LabelsFactory:
     def __init__(self,
                  f_soc: SoCFunctionFactory,
                  soc_profile: SoCProfileFactory):
-        self.f_soc: SoCFunctionFactory = f_soc
+        self.f_soc_factory: SoCFunctionFactory = f_soc
         self.soc_profile: SoCProfileFactory = soc_profile
 
     def spawn_label(self, current_node: Node, current_label: Label, t_charge: Time):
         # Only charge the minimum at the last charge station
         # and continue charging at this station.
-        soc_function: SoCFunction = self.f_soc(current_label)
+        soc_function: SoCFunction = self.f_soc_factory(current_label)
 
         return Label(
             t_trip=current_label.t_trip + t_charge,

evrouting/charge/routing.py

@@ -12,8 +12,8 @@ from evrouting.charge.factories import (
 )
 
 from ..graph_tools import distance
-from .T import SoCProfile, SoCFunction, Label
-from .utils import LabelPriorityQueue
+from .T import SoCFunction, Label
+from .utils import LabelPriorityQueue, keys
 
 __all__ = ['shortest_path']
 
@@ -86,7 +86,7 @@ def shortest_path(G: nx.Graph, charging_stations: set, s: Node, t: Node,
             soc_profile = label_minimum_node.soc_profile_cs_v + \
                           soc_profile_factory(minimum_node, n)
 
-            if _is_feasible_path(soc_profile, capacity):
+            if soc_profile(capacity) != -inf:
                 l_new = Label(
                     t_trip=label_minimum_node.t_trip + distance(G, minimum_node, n),
                     soc_last_cs=label_minimum_node.soc_last_cs,
@@ -112,8 +112,7 @@ def shortest_path(G: nx.Graph, charging_stations: set, s: Node, t: Node,
                     continue
 
                 if is_new_min_label:
-                    key, count = _key(l_new, f_soc_factory)
-                    prio_queue.insert(n, priority=key, count=count)
+                    prio_queue.insert(n, **keys(f_soc_factory, l_new))
 
 
 def _calc_optimal_t_charge(cf: ChargingFunctionMap, label_v: Label, v: Node, capacity: SoC) -> List[Time]:
@@ -132,15 +131,6 @@ def _calc_optimal_t_charge(cf: ChargingFunctionMap, label_v: Label, v: Node, cap
     return t_charge
 
 
-def _key(label, f_soc_factory):
-    soc_function = f_soc_factory(label)
-
-    t_min = soc_function.minimum
-    soc_min = soc_function(t_min)
-
-    return t_min, soc_min
-
-
 def _create_entry_label(
         G: nx.Graph,
         charging_stations: set,
@@ -175,11 +165,6 @@ def _create_entry_label(
     )
 
 
-def _is_feasible_path(soc_profile: SoCProfile, capacity: SoC) -> bool:
-    """Check, if possible to traverse path at least with full battery."""
-    return not soc_profile(capacity) == -inf
-
-
 def _update_priority_queue(
         f_soc: SoCFunctionFactory,
         prio_queue: PriorityQueue,
@@ -195,8 +180,7 @@ def _update_priority_queue(
         # l_uns[v] empty
         prio_queue.delete_min()
     else:
-        key, count = _key(minimum_label, f_soc)
-        prio_queue.insert(node, priority=key, count=count)
+        prio_queue.insert(node, **keys(f_soc, minimum_label))
 
 
 def _apply_final_constraints(G: nx.Graph, t: Node, final_soc: SoC) -> Node:

evrouting/charge/utils.py

-from typing import Set, Any
+from typing import Set, Any, Dict
 from math import inf
 
 from evrouting.utils import PriorityQueue
@@ -16,20 +16,7 @@ class LabelPriorityQueue(PriorityQueue):
 
     def insert(self, label: Label):
         """Breaking ties with lowest soc at t_min."""
-        soc_function = self.f_soc_factory(label)
-        t_min: Time = soc_function.minimum
-
-        # Might happen because of dummy charge stations
-        if t_min == -inf:
-            raise ValueError('Infeasible label.')
-
-        soc_min: SoC = soc_function(t_min)
-
-        super().insert(
-            item=label,
-            priority=t_min,
-            count=soc_min
-        )
+        super().insert(item=label, **keys(self.f_soc_factory, label))
 
         if self.peak_min() == label:
             self.dominance_check()
@@ -51,3 +38,16 @@ class LabelPriorityQueue(PriorityQueue):
             if self.f_soc_factory(other_label) > soc:
                 self.remove_item(label)
                 return
+
+
+def keys(f_soc_factory: SoCFunctionFactory, label: Label) -> Dict:
+    soc_function = f_soc_factory(label)
+    t_min: Time = soc_function.minimum
+
+    # Might happen because of dummy charge stations
+    if t_min == -inf:
+        raise ValueError('Infeasible label.')
+
+    soc_min: SoC = soc_function(t_min)
+
+    return {'priority': t_min, 'count': soc_min}

tests/charge/test_utils.py

 import pytest
 from evrouting.charge.utils import LabelPriorityQueue
 from evrouting.charge.T import Label
+from evrouting.charge.factories import SoCFunctionFactory
 
 
 @pytest.fixture
 def q(label, ch_function):
     _, _, cf = ch_function
     dummy_cf = {label.last_cs: cf}
-    q = LabelPriorityQueue(dummy_cf)
+    q = LabelPriorityQueue(SoCFunctionFactory(dummy_cf), l_set=set({}))
     q.insert(label)
 
     # create min