From 4570fa9134d7b646c79f3d4417136a3f68659587 Mon Sep 17 00:00:00 2001
From: "christoph.von.oy" <christoph.von.oy@rwth-aachen.de>
Date: Wed, 27 Sep 2023 11:46:30 +0200
Subject: [PATCH] Implemented GlobalDynamic
---
dynamics/Dynamic.py | 270 ++++++++++++++++++++++++++++++++++++--------
1 file changed, 222 insertions(+), 48 deletions(-)
diff --git a/dynamics/Dynamic.py b/dynamics/Dynamic.py
index 2bf1155..e4af138 100644
--- a/dynamics/Dynamic.py
+++ b/dynamics/Dynamic.py
@@ -25,6 +25,59 @@ THE SOFTWARE.
import pandas as pd
import pyomo.environ as pyo
+class GlobalDynamic:
+ def __init__(self, d_steps):
+ self.d_steps = d_steps
+ self.root_dynamic = TrivialDynamic(self.d_steps, self)
+ self.dynamics = dict()
+ self.dynamics[self.root_dynamic] = (dict(), dict())
+ self.symbol_table = dict()
+ self.symbol_table[self.root_dynamic] = 0
+ self.assignments = dict()
+
+ def root(self):
+ return self.root_dynamic
+
+ # indices are relative to the root dynamic
+ def sub_dynamic(self, dynamic, indices):
+ if isinstance(dynamic, PartialDynamic):
+ dynamic = dynamic.reference
+ if tuple(indices) in self.dynamics[dynamic][0].keys():
+ return self.dynamics[dynamic][0][tuple(indices)]
+ sub_dynamic = BackedDynamic(dynamic, indices, self)
+ self.dynamics[dynamic][0][tuple(indices)] = sub_dynamic
+ self.dynamics[sub_dynamic] = (dict(), dict())
+ self.symbol_table[sub_dynamic] = len(self.symbol_table)
+ return sub_dynamic
+
+ # start/end are relative to dynamic
+ def partial_dynamic(self, dynamic, start, end):
+ if start == 0 and end == dynamic.number_of_steps():
+ return dynamic
+ if isinstance(dynamic, PartialDynamic):
+ start += dynamic.start
+ end += dynamic.start
+ dynamic = dynamic.reference
+ if (start, end) in self.dynamics[dynamic][1].keys():
+ return self.dynamics[dynamic][1][(start, end)]
+ partial_dynamic = PartialDynamic(dynamic, start, end, self)
+ self.dynamics[dynamic][1][(start, end)] = partial_dynamic
+ self.dynamics[partial_dynamic] = (dict(), dict())
+ self.symbol_table[partial_dynamic] = len(self.symbol_table)
+ return partial_dynamic
+
+ def display(self):
+ output = ''
+ for dynamic, (sub_dynamics, partial_dynamics) in self.dynamics.items():
+ if len(sub_dynamics) == 0 and len(partial_dynamics) == 0:
+ continue
+ output += f'{self.symbol_table[dynamic]}:\n'
+ for indices, sub_dynamic in sub_dynamics.items():
+ output += f'\t{indices}: {self.symbol_table[sub_dynamic]}\n'
+ for (start, end), partial_dynamic in partial_dynamics.items():
+ output += f'\t{start}-{end}: {self.symbol_table[partial_dynamic]}\n'
+ return output
+
# The root dynamic is
# for TrivialDynamic: self
# for BackedDynamic: root dynamic of its reference
@@ -48,6 +101,10 @@ class Dynamic:
def time_steps(self):
pass
+ # returns all indices in this dynamic
+ def all_indices(self):
+ pass
+
# return true if the dynamic contains the given index
# index is relative to the root dynamic
def has_index(self, index):
@@ -107,20 +164,115 @@ class Dynamic:
def partial_dynamic(self, i_start, i_end):
pass
- def to_string(self):
- to_print = "["
- for p in range(self.number_of_steps()):
- to_print += "(" + str(self.index_of(p)) + ", " + str(self.step_size_p(p)) + ")"
- if p != self.number_of_steps() - 1:
- to_print += ", "
- to_print += "]"
- return to_print
+ def display(self):
+ numbers = self.all_indices()
+ rows = [[]]
+ row_ends = [0]
+ for number in numbers:
+ local_position = number - numbers[0]
+ row_index = 0
+ while row_index < len(rows) and row_ends[row_index] > local_position:
+ row_index += 1
+ if row_index == len(rows):
+ rows.append([])
+ row_ends.append(0)
+ rows[row_index].append(number)
+ row_ends[row_index] = local_position + len(str(number)) + 1
+ row_strings = []
+ for items in rows:
+ row_string = ''
+ string_end = 0
+ for number in items:
+ local_position = number - numbers[0]
+ row_string += ' ' * (local_position - string_end)
+ row_string += str(number)
+ string_end = local_position + len(str(number))
+ row_strings.append(row_string)
+ output = ''
+ for row_index in range(len(row_strings)):
+ output += row_strings[len(row_strings) - row_index - 1] + '\n'
+ string_end = 0
+ for number in numbers:
+ local_position = number - numbers[0]
+ output += ' ' * (local_position - string_end)
+ output += '|'
+ string_end = local_position + 1
+ output += '\n'
+ return output
+
+ def display_alignment(self, other):
+ p_self = 0
+ p_other = 0
+ numbers = []
+ while p_self <= self.number_of_steps() and p_other <= other.number_of_steps():
+ i_self = self.index_of(p_self)
+ i_other = other.index_of(p_other)
+ if i_self < i_other:
+ numbers.append(i_self)
+ p_self += 1
+ elif i_self > i_other:
+ numbers.append(i_other)
+ p_other += 1
+ else:
+ numbers.append(i_self)
+ p_self += 1
+ p_other += 1
+ while p_self < self.number_of_steps():
+ numbers.append(self.index_of(p_self))
+ p_self += 1
+ else:
+ while p_other < other.number_of_steps():
+ numbers.append(other.index_of(p_other))
+ p_other += 1
+ rows = [[]]
+ row_ends = [0]
+ for number in numbers:
+ local_position = number - numbers[0]
+ row_index = 0
+ while row_index < len(rows) and row_ends[row_index] > local_position:
+ row_index += 1
+ if row_index == len(rows):
+ rows.append([])
+ row_ends.append(0)
+ rows[row_index].append(number)
+ row_ends[row_index] = local_position + len(str(number)) + 1
+ row_strings = []
+ for items in rows:
+ row_string = ''
+ self_string_end = 0
+ for number in items:
+ local_position = number - numbers[0]
+ row_string += ' ' * (local_position - self_string_end)
+ row_string += str(number)
+ self_string_end = local_position + len(str(number))
+ row_strings.append(row_string)
+ output = ''
+ for row_index in range(len(row_strings)):
+ output += row_strings[len(row_strings) - row_index - 1] + '\n'
+ self_string = ' ' * (self.index_of(0) - numbers[0])
+ self_string_end = self.index_of(0) - numbers[0]
+ for index in self.all_indices():
+ local_position = index - numbers[0]
+ self_string += ' ' * (local_position - self_string_end)
+ self_string += '|'
+ self_string_end = local_position + 1
+ output += self_string + '\n'
+ other_string = ' ' * (other.index_of(0) - numbers[0])
+ other_string_end = other.index_of(0) - numbers[0]
+ for index in other.all_indices():
+ local_position = index - numbers[0]
+ other_string += ' ' * (local_position - other_string_end)
+ other_string += '|'
+ other_string_end = local_position + 1
+ output += other_string + '\n'
+ return output
# a dynamic defined by the length of each time step
class TrivialDynamic(Dynamic):
# d_step: the length all time_steps
- def __init__(self, d_steps):
+ def __init__(self, d_steps, global_dynamic):
self.d_steps = d_steps
+ self.global_dynamic = global_dynamic
def root(self):
return self
@@ -134,6 +286,9 @@ class TrivialDynamic(Dynamic):
def time_steps(self):
return range(len(self.d_steps))
+ def all_indices(self):
+ return range(len(self.d_steps) + 1)
+
def has_index(self, index):
return 0 <= index and index <= len(self.d_steps)
@@ -159,25 +314,26 @@ class TrivialDynamic(Dynamic):
return self.d_steps
def sub_dynamic_p(self, positions):
- return BackedDynamic(self, positions)
+ return self.global_dynamic.sub_dynamic(self, positions)
def sub_dynamic(self, indices):
- return BackedDynamic(self, indices)
+ return self.global_dynamic.sub_dynamic(self, indices)
def partial_dynamic_p(self, p_start, p_end):
- return PartialDynamic(self, p_start, p_end)
+ return self.global_dynamic.partial_dynamic(self, p_start, p_end)
def partial_dynamic(self, i_start, i_end):
- return PartialDynamic(self, i_start, i_end)
+ return self.global_dynamic.partial_dynamic(self, i_start, i_end)
# a dynamic definied by taking certain time steps form a reference dynamic
class BackedDynamic(Dynamic):
# reference: the dynamic that backs this dynamic
# indices: the indicies of the time steps contained in this dynamic with the last index representing the end of the last time step
# indices are relative to the reference dynamic
- def __init__(self, reference, indices):
+ def __init__(self, reference, indices, global_dynamic):
self.reference = reference
self.indices = indices
+ self.global_dynamic = global_dynamic
def root(self):
return self.reference.root()
@@ -191,6 +347,9 @@ class BackedDynamic(Dynamic):
def time_steps(self):
return self.indices[:-1]
+ def all_indices(self):
+ return self.indices
+
def has_index(self, index):
return index in self.indices
@@ -230,22 +389,22 @@ class BackedDynamic(Dynamic):
def sub_dynamic_p(self, positions):
if any(position < 0 or len(self.indices) <= position for position in positions):
raise IndexError("The dynamic does not have all requested indices!")
- return BackedDynamic(self.reference, [self.indices[position] for position in positions])
+ return self.global_dynamic.sub_dynamic(self, [self.indices[position] for position in positions])
def sub_dynamic(self, indices):
if any(index not in self.indices for index in indices):
raise IndexError("The dynamic does not have all requested indices for the sub dynamic!")
- return BackedDynamic(self.reference, indices)
+ return self.global_dynamic.sub_dynamic(self, indices)
def partial_dynamic_p(self, p_start, p_end):
if p_start < 0 or len(self.indices) <= p_start or p_end < 0 or len(self.indices) <= p_end:
raise IndexError("The dynamic does not have all requested positions for the sub dynamic!")
- return PartialDynamic(self, p_start, p_end)
+ return self.global_dynamic.partial_dynamic(self, p_start, p_end)
def partial_dynamic(self, i_start, i_end):
if i_start not in self.indices or i_end not in self.indices:
raise IndexError("The dynamic does not have all requested indices for the sub dynamic!")
- return PartialDynamic(self, self.indices.index(i_start), self.indices.index(i_end))
+ return self.global_dynamic.partial_dynamic(self, self.indices.index(i_start), self.indices.index(i_end))
# a dynamic defined by taking a continuus intervall of time steps from a reference dynamic
class PartialDynamic(Dynamic):
@@ -253,10 +412,11 @@ class PartialDynamic(Dynamic):
# start: the position in the reference dynamic of the first time step contained in this dynamic
# end: the position in the reference dynamic of the end of the last time step contained in this dynamic
# start/end are relative to the reference dynamic
- def __init__(self, reference, start, end):
+ def __init__(self, reference, start, end, global_dynamic):
self.reference = reference
self.start = start
self.end = end
+ self.global_dynamic = global_dynamic
def root(self):
return self.reference.root()
@@ -270,6 +430,9 @@ class PartialDynamic(Dynamic):
def time_steps(self):
return self.reference.time_steps()[self.start:self.end]
+ def all_indices(self):
+ return self.reference.all_indices()[self.start:self.end + 1]
+
def has_index(self, index):
if self.reference.has_index(index):
reference_position = self.reference.position_of(index)
@@ -316,7 +479,7 @@ class PartialDynamic(Dynamic):
def sub_dynamic_p(self, positions):
if any(position < 0 or self.end - self.start < position for position in positions):
raise IndexError("The dynamic does not have all requested positions for the sub dynamic!")
- return self.reference.sub_dynamic_p([self.start + position for position in positions])
+ return self.global_dynamic.sub_dynamic(self, [self.index_of(position) for position in positions])
def sub_dynamic(self, indices):
def filter(index):
@@ -324,19 +487,19 @@ class PartialDynamic(Dynamic):
return reference_position < self.start or self.end < reference_position
if any(filter(index) for index in indices):
raise IndexError("The does not have all requested indices for the sub dynamic!")
- return self.reference.sub_dynamic(indices)
+ return self.global_dynamic.sub_dynamic(self, indices)
def partial_dynamic_p(self, p_start, p_end):
if p_start < 0 or self.end - self.start < p_start or p_end < 0 or self.end - self.start < p_end:
raise IndexError("The dynamic does not have all requested positions for the sub dynamic!")
- return PartialDynamic(self.reference, self.start + p_start, self.start + p_end)
+ return self.global_dynamic.partial_dynamic(self, p_start, p_end)
def partial_dynamic(self, i_start, i_end):
reference_start = self.reference.position_of(i_start)
reference_end = self.reference.position_of(i_end)
if reference_start < self.start or self.end < reference_start or reference_end < self.start or self.end < reference_end:
raise IndexError("The dynamic does not have all requested indices for the sub dynamic!")
- return PartialDynamic(self.reference, reference_start, reference_end)
+ return self.global_dynamic.partial_dynamic(self, self.position_of(i_start), self.position_of(i_end))
# only works if both dynamics share the same root dynamic
def resample(values, dynamic, target_dynamic):
@@ -702,28 +865,39 @@ def test_single_resampling(dynamic, target_dynamic):
print('hey')
def test_resampling():
- dynamics = {}
- dynamics['d'] = TrivialDynamic([1 for i in range(36)])
- dynamics['d_1'] = PartialDynamic(dynamics['d'], 0, 36)
- dynamics['d_2'] = PartialDynamic(dynamics['d'], 1, 35)
- dynamics['d_3'] = BackedDynamic(dynamics['d'], [0, 2, 4, 6, 8, 10, 12, 14, 16, 18, 20, 22, 24, 26, 28, 30, 32, 34, 36])
- dynamics['d_3_1'] = PartialDynamic(dynamics['d_3'], 0, 18)
- dynamics['d_3_2'] = PartialDynamic(dynamics['d_3'], 1, 17)
- dynamics['d_3_3'] = BackedDynamic(dynamics['d_3'], [0, 4, 8, 12, 16, 20, 24, 28, 32, 36])
- dynamics['d_3_3_1'] = PartialDynamic(dynamics['d_3_3'], 0, 9)
- dynamics['d_3_3_2'] = PartialDynamic(dynamics['d_3_3'], 1, 8)
- dynamics['d_3_4'] = BackedDynamic(dynamics['d_3'], [2, 6, 10, 14, 18, 22, 26, 30, 34])
- dynamics['d_3_4_1'] = PartialDynamic(dynamics['d_3_4'], 0, 8)
- dynamics['d_3_4_2'] = PartialDynamic(dynamics['d_3_4'], 1, 7)
- dynamics['d_4'] = BackedDynamic(dynamics['d'], [3, 6, 9, 12, 15, 18, 21, 24, 27, 30, 33])
- dynamics['d_4_1'] = PartialDynamic(dynamics['d_4'], 0, 10)
- dynamics['d_4_2'] = PartialDynamic(dynamics['d_4'], 1, 9)
- dynamics['d_4_3'] = BackedDynamic(dynamics['d_4'], [3, 9, 15, 21, 27, 33])
- dynamics['d_4_3_1'] = PartialDynamic(dynamics['d_4_3'], 0, 5)
- dynamics['d_4_3_2'] = PartialDynamic(dynamics['d_4_3'], 1, 4)
- dynamics['d_4_4'] = BackedDynamic(dynamics['d_4'], [6, 12, 18, 24, 30])
- dynamics['d_4_4_1'] = PartialDynamic(dynamics['d_4_4'], 0, 4)
- dynamics['d_4_4_2'] = PartialDynamic(dynamics['d_4_4'], 1, 3)
- dynamics['d_5'] = BackedDynamic(dynamics['d'], [0, 3, 6, 9, 12, 15, 18, 21, 24, 27, 30, 33, 36])
-
- test_single_resampling(dynamics['d'], dynamics['d'])
+ import random
+ random.seed(0)
+ global_dynamic = GlobalDynamic([1 for i in range(100)])
+ dynamics = []
+ dynamics.append(global_dynamic.root())
+ for i in range(100):
+ dynamic_number = random.randint(0, len(dynamics) - 1)
+ dynamic = dynamics[dynamic_number]
+ if random.random() < 0.75:
+ original_indices = list(dynamic.all_indices())
+ number = random.randint(2, len(original_indices))
+ indices = []
+ for i in range(number):
+ choice = random.choice(original_indices)
+ original_indices.remove(choice)
+ indices.append(choice)
+ indices.sort()
+ sub_dynamic = dynamic.sub_dynamic(indices)
+ if sub_dynamic not in dynamics:
+ dynamics.append(sub_dynamic)
+ else:
+ original_positions = list(range(dynamic.number_of_steps() + 1))
+ positions = []
+ for i in range(2):
+ choice = random.choice(original_positions)
+ original_positions.remove(choice)
+ positions.append(choice)
+ positions.sort()
+ p_start = positions[0]
+ p_end = positions[1]
+ partial_dynamic = dynamic.partial_dynamic_p(p_start, p_end)
+ if partial_dynamic not in dynamics:
+ dynamics.append(partial_dynamic)
+ print(global_dynamic.display(), end='')
+ test_single_resampling(dynamics[0], dynamics[0])
+ print(dynamics[0].display_alignment(dynamics[0]), end='')
--
GitLab