diff --git a/evrouting/T.py b/evrouting/T.py
index e95dbd259a2e81b808ad5844072e165ad02020fa..117ad593ad10b6622604fcdc8d8619c8524e2c64 100644
--- a/evrouting/T.py
+++ b/evrouting/T.py
@@ -1,9 +1,19 @@
-from typing import Tuple, Union, NewType
+from dataclasses import dataclass
+from typing import Tuple, Union, NewType, Dict, Any
from math import inf
Node = int
Edge = Tuple[Node, Node]
+
+@dataclass
+class DummyNode:
+ n: Node
+
+
+NodeData = Dict[str, Any]
+EdgeData = Dict[str, Any]
+
Wh = NewType('Wh', Union[float, int])
SoC = NewType('SoC', Union[-inf, Wh])
diff --git a/evrouting/charge/T.py b/evrouting/charge/T.py
index 4d8c3084601878a3219a5f4b7c9e5ebe28ca9002..be75e784b5f86b2695c995955d80f24ace1b6e47 100644
--- a/evrouting/charge/T.py
+++ b/evrouting/charge/T.py
@@ -1,18 +1,17 @@
-from typing import Tuple
+from typing import Union
from copy import copy
from collections import namedtuple
from math import inf
import networkx as nx
-
-from evrouting.T import SoC, Wh, ChargingCoefficient, Time, Node
+from evrouting.T import SoC, Wh, ChargingCoefficient, Time, Node, DummyNode
from evrouting.graph_tools import charging_cofficient, consumption
class Label(namedtuple):
t_trip: Time
beta_u: SoC
- u: Node
+ u: Union[Node, DummyNode]
SoCProfile: SoCProfile
@@ -28,14 +27,27 @@ class ChargingFunction:
return beta if beta < self.U else self.U
+class DummyChargingFunction:
+ def __init__(self, beta_s: SoC):
+ self.beta_s = beta_s
+
+ def __call__(self, t: Time, beta: SoC = 0) -> SoC:
+ return self.beta_s
+
+
class SoCFunction:
def __init__(self, G: nx.Graph, l: Label, U: SoC):
self.t_trip: Time = l.t_trip
self.beta_u: SoC = l.beta_u
- self.u: Node = l.u
+ self.u: Union[Node, DummyNode] = l.u
self.b_u_v: SoCProfile = l.SoCProfile
- self.cf: ChargingFunction = ChargingFunction(G, l.u, U)
+
+ self.cf: Union[ChargingFunction, DummyChargingFunction]
+ if isinstance(self.u, DummyNode):
+ self.cf = DummyChargingFunction(self.beta_u)
+ else:
+ self.cf = ChargingFunction(G, self.u, U)
def __call__(self, t: Time) -> SoC:
if t < self.t_trip:
@@ -43,8 +55,11 @@ class SoCFunction:
return self.b_u_v(self.cf(t - self.t_trip, self.beta_u))
- def get_minimum(self) -> Time:
+ def get_minimum(self, beta_t: SoC = 0) -> Time:
"""TODO: Explain."""
+ if isinstance(self.cf, DummyChargingFunction):
+ return 0
+
cost_p = self.b_u_v.cost
return max(self.t_trip, (cost_p - self.beta_u) /
self.cf.c + self.t_trip)
diff --git a/evrouting/charge/routing.py b/evrouting/charge/routing.py
index 2d3778cb36c8a52c0cb5811f6687b98e02f7f0c8..5612d70b4bd6fa4525f9485397d2abba06e5ee9b 100644
--- a/evrouting/charge/routing.py
+++ b/evrouting/charge/routing.py
@@ -5,7 +5,8 @@ import networkx as nx
from evrouting.T import Node, SoC, Time
from evrouting.utils import PriorityQueue
-from .T import SoCProfile, SoCFunction, Label, ChargingFunction
+from .T import SoCProfile, SoCFunction, Label, ChargingFunction, DummyNode
+from .utils import LabelPriorityQueue
from ..graph_tools import distance
@@ -24,18 +25,15 @@ def shortest_path(G: nx.Graph, S: set, s: Node, t: Node,
"""
q = PriorityQueue()
l_set: Dict[int, set] = {v: set() for v in G}
- l_uns: Dict[int, PriorityQueue] = {v: PriorityQueue() for v in G}
+ l_uns: Dict[int, LabelPriorityQueue] = {v: LabelPriorityQueue(G, U) for v in G}
# Dummy vertex without incident edges that is (temporarily) added to G
- v_0: Node = len(G.nodes)
- G.add_node(v_0)
- S.add(v_0)
+ dummy_node: DummyNode = DummyNode(len(G.nodes))
+ G.add_node(dummy_node)
+ S.add(dummy_node)
- cf_v_0 = [(0, beta_s)]
- l_uns[s] = PriorityQueue()
-
- l = Label(0, beta_s, v_0, SoCProfile(G, U, s))
- l_uns[s].insert(item=l, priority=key(l))
+ l = Label(0, beta_s, dummy_node, SoCProfile(G, U, s))
+ l_uns[s].insert(l)
q.insert(s, 0)
@@ -51,7 +49,7 @@ def shortest_path(G: nx.Graph, S: set, s: Node, t: Node,
l_set[v].add(l)
if v == t:
- return SoCFunction(G, l, U).get_minimum()
+ return SoCFunction(G, l, U).get_minimum(beta_t)
# handle charging stations
t_trip, beta_u, u, b_u_v = l
@@ -68,7 +66,7 @@ def shortest_path(G: nx.Graph, S: set, s: Node, t: Node,
u=v,
SoCProfile=SoCProfile(G, U, v)
)
- l_uns[v].insert(l_new, priority=key(l_new))
+ l_uns[v].insert(l_new)
# update priority queue
if l_uns[v]:
@@ -89,4 +87,3 @@ def shortest_path(G: nx.Graph, S: set, s: Node, t: Node,
def key(l: Label) -> Time:
return l.t_trip
-
diff --git a/evrouting/charge/utils.py b/evrouting/charge/utils.py
new file mode 100644
index 0000000000000000000000000000000000000000..2d885d89ac56e003d2e060c8f721626eb2dc4e9a
--- /dev/null
+++ b/evrouting/charge/utils.py
@@ -0,0 +1,22 @@
+import networkx as nx
+from evrouting.utils import PriorityQueue
+from evrouting.T import SoC
+from .T import Label, SoCFunction
+
+
+class LabelPriorityQueue(PriorityQueue):
+ def __init__(self, G: nx.Graph, U: SoC):
+ super().__init__()
+ self.G = G
+ self.U = U
+
+ def insert(self, item: Label):
+ """Breaking ties with lowest soc at t_min."""
+ soc_function = SoCFunction(self.G, item, self.U)
+ t_min = soc_function.get_minimum()
+
+ super().insert(
+ item,
+ priority=t_min,
+ count=soc_function(t_min)
+ )
diff --git a/evrouting/graph_tools.py b/evrouting/graph_tools.py
index e08c11920d292675279e18cb89938c79d69e8ee2..ea494a64baaed79ef93495765dbc9ac1f62a0a37 100644
--- a/evrouting/graph_tools.py
+++ b/evrouting/graph_tools.py
@@ -1,20 +1,13 @@
-from typing import Dict, Tuple, Any
from collections import namedtuple
import networkx as nx
-from evrouting.T import Wh, ChargingCoefficient, Time
+from evrouting.T import Wh, ChargingCoefficient, Time, Node, NodeData, EdgeData
TemplateEdge = namedtuple('Edge', ['u', 'v', 'distance', 'consumption'])
TemplateNode = namedtuple(
'Node', ['label', 'charging_coeff'], defaults=(None, None)
)
-NodeData = Dict[str, Any]
-EdgeData = Dict[str, Any]
-
-Node = int
-Edge = Tuple[int, int]
-
def node_convert(n: TemplateNode) -> NodeData:
return {'label': n.label, 'c': n.charging_coeff}
diff --git a/evrouting/utils.py b/evrouting/utils.py
index 8aa001fc41dee4cc845c37141a8aa834d71e47c1..0919e515ac4d0a6dfccb4ff7011d539eb1ccfa0c 100644
--- a/evrouting/utils.py
+++ b/evrouting/utils.py
@@ -11,11 +11,11 @@ class PriorityQueue:
self.entry_finder = {} # mapping of tasks to entries
self.counter = itertools.count() # unique sequence count as tie break
- def insert(self, item: Any, priority=0):
+ def insert(self, item: Any, priority: Any = 0, count: Any = None):
"""Add a new task or update the priority of an existing task"""
if item in self.entry_finder:
self.remove_item(item)
- count = next(self.counter)
+ count = count or next(self.counter)
entry = [priority, count, item]
self.entry_finder[item] = entry
heappush(self.pq, entry)