From 5c27e442c8774bdb02e9ff097cb91d560ca88627 Mon Sep 17 00:00:00 2001
From: "niehues.mark@gmail.com" <niehues.mark@gmail.com>
Date: Thu, 19 Mar 2020 17:09:45 +0100
Subject: [PATCH] wip
---
evrouting/charge/factories.py | 7 ++++---
evrouting/charge/routing.py | 26 +++++---------------------
evrouting/charge/utils.py | 30 +++++++++++++++---------------
tests/charge/test_utils.py | 3 ++-
4 files changed, 26 insertions(+), 40 deletions(-)
diff --git a/evrouting/charge/factories.py b/evrouting/charge/factories.py
index 15ffdb6..74774b8 100644
--- a/evrouting/charge/factories.py
+++ b/evrouting/charge/factories.py
@@ -1,6 +1,7 @@
-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,
diff --git a/evrouting/charge/routing.py b/evrouting/charge/routing.py
index ab13fe8..18dab9e 100644
--- a/evrouting/charge/routing.py
+++ b/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:
diff --git a/evrouting/charge/utils.py b/evrouting/charge/utils.py
index bc42556..23b3b61 100644
--- a/evrouting/charge/utils.py
+++ b/evrouting/charge/utils.py
@@ -1,4 +1,4 @@
-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}
diff --git a/tests/charge/test_utils.py b/tests/charge/test_utils.py
index d5bb2a2..ee517c9 100644
--- a/tests/charge/test_utils.py
+++ b/tests/charge/test_utils.py
@@ -1,13 +1,14 @@
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
--
GitLab