This commit is contained in:
2024-07-08 17:30:43 +08:00
parent 7911b9b39d
commit 2f7e176d58
39 changed files with 233567 additions and 0 deletions
+411
View File
@@ -0,0 +1,411 @@
import copy
import matplotlib.pyplot as plt
import ruptures as rpt
from data_generator import *
import random
import json
def avg(target_list):
return sum(target_list) / len(target_list)
def pelt_detection(model, target_pen, target_seq):
"""
使用pelt对给定的序列进行检测
:param model:
:param target_pen:
:param target_seq:
:return:
"""
algo = rpt.Pelt(model=model).fit(target_seq)
detection_result = algo.predict(pen=target_pen)
while len(target_seq) in detection_result:
detection_result.remove(len(target_seq))
return detection_result
def is_spike_percentage(x1, x2, spike_percentage):
"""
根据变点前后段的均值,判断是否存在跳变
:param x1:
:param x2:
:param spike_percentage:
:return:
"""
if cal_relative_diff(x1, x2) >= spike_percentage:
return True
else:
return False
def cal_relative_diff(x1, x2):
"""
计算两个数据的相对偏差
:param x1:
:param x2:
:return:
"""
return abs(abs(x1-x2) / ((x1 + x2) / 2))
def point_classification(slice_sequence, slice_sequence_avg, spike_percentage, spike_length):
"""
:param slice_sequence:
:param slice_sequence_avg:
:param spike_percentage:
:param spike_length:
:return:
"""
result_dict = {'spike_up_point': [], 'spike_down_point': [], 'step_up_point': [], 'step_down_point': [],
'slop_up_tuple': [], 'slop_down_tuple': []}
spike_up_flag, spike_down_flag = False, False
for i in range(len(slice_sequence_avg) - 1):
if spike_up_flag:
result_dict['spike_up_point'].append(i)
spike_up_flag = False
continue
if spike_down_flag:
result_dict['spike_down_point'].append(i)
spike_down_flag = False
continue
now_slice_avg, next_slice_avg = slice_sequence_avg[i], slice_sequence_avg[i + 1]
if next_slice_avg > now_slice_avg \
and is_spike_percentage(next_slice_avg, now_slice_avg, spike_percentage): # 向上跳变
if i + 2 < len(slice_sequence_avg): # 仍然存在更多变点
further_slice_avg = slice_sequence_avg[i + 2]
if further_slice_avg < next_slice_avg and len(slice_sequence[i + 1]) <= spike_length and \
is_spike_percentage(next_slice_avg, further_slice_avg, spike_percentage):
# 向上跳变以后向下跳变,则为激增点
result_dict['spike_up_point'].append(i)
spike_up_flag = True
continue
else:
# 向上跳变以后没有显著向下跳变,该点为上升点
result_dict['step_up_point'].append(i)
continue
else:
result_dict['step_up_point'].append(i)
elif now_slice_avg > next_slice_avg \
and is_spike_percentage(now_slice_avg, next_slice_avg, spike_percentage): # 向下跳变
if i + 2 < len(slice_sequence_avg): # 仍然存在更多变点
further_slice_avg = slice_sequence_avg[i + 2]
if further_slice_avg > next_slice_avg and len(slice_sequence[i + 1]) <= spike_length and \
is_spike_percentage(further_slice_avg, next_slice_avg, spike_percentage):
# 向下跳变以后向上跳变,则为骤降点
result_dict['spike_down_point'].append(i)
spike_down_flag = True
continue
else:
# 向下跳变以后没有显著向上跳变,该点为下降点
result_dict['step_down_point'].append(i)
continue
else:
result_dict['step_down_point'].append(i)
continue
elif next_slice_avg > now_slice_avg:
result_dict['step_up_point'].append(i)
elif next_slice_avg < now_slice_avg:
result_dict['step_down_point'].append(i)
return result_dict
def is_slop_detection(sequence, slop_tuple, strip_num, min_slop_val, min_interval):
"""
输入一个序列以及一个缓坡的开始结束点,判断是否确实是缓坡
:param sequence:
:param slop_tuple:
:param strip_num:
:param min_slop_val:
:param min_interval:
:return:
"""
if (slop_tuple[1] - slop_tuple[0]) > 2 * min_interval:
return True
else:
return False
def strip_step_interval(target_list, min_interval, is_og_content=True):
"""
去除间隔过小的序列
:param is_og_content:
:param target_list:
:param min_interval:
:return:
"""
result_list = []
target_list.sort()
continue_flag = False
for i in range(len(target_list) - 1):
if continue_flag:
continue_flag = False
continue
now_val, next_val = target_list[i], target_list[i + 1]
if abs(next_val - now_val) < min_interval:
if is_og_content:
result_list.append(now_val)
else:
result_list.append(int(avg([now_val, next_val])))
continue_flag = True
continue
else:
result_list.append(now_val)
if not continue_flag and target_list:
result_list.append(target_list[-1])
return result_list
def enlarge_step_interval(target_list, interval):
"""
将序列中过小的间隔进行放大处理
:param interval:
:param target_list:
:return:
"""
result_list = []
target_list.sort()
continue_flag = False
for i in range(len(target_list) - 1):
if continue_flag:
continue_flag = False
continue
now_val, next_val = target_list[i], target_list[i + 1]
if abs(next_val - now_val) < interval:
result_list.append(now_val - interval)
result_list.append(next_val + interval)
continue_flag = True
continue
else:
result_list.append(now_val)
if not continue_flag and target_list:
result_list.append(target_list[-1])
return result_list
def find_spike_point(sequence, target_list, min_interval):
"""
寻找一段激增/骤降区间中的极值
:param sequence:
:param target_list:
:param min_interval:
:return:
"""
result_list = []
target_list.sort()
continue_flag = False
for i in range(len(target_list) - 1):
if continue_flag:
continue_flag = False
continue
now_val, next_val = target_list[i], target_list[i + 1]
if abs(next_val - now_val) < min_interval:
# 寻找序列中与均值差异最大的点,为极值点
seq_slice = [sequence[i] for i in range(now_val, next_val + 1)]
temp_list = [abs(i - avg(seq_slice)) for i in seq_slice]
temp_item = temp_list.index(max(temp_list))
result_list.append(now_val + temp_item)
continue_flag = True
continue
else:
result_list.append(now_val)
if not continue_flag and target_list:
result_list.append(target_list[-1])
return result_list
def find_sequential_sub_list(target):
if not target:
return []
result_list = []
temp_list = [target[0]]
for i in range(1, len(target)):
if target[i] == temp_list[-1] + 1:
temp_list.append(target[i])
else:
result_list.append(copy.deepcopy(temp_list))
temp_list = [target[i]]
result_list.append(copy.deepcopy(temp_list))
return result_list
def new_slop_detection(sequence, pelt_result, point_list, strip_num, slop_threshold, min_slop_val, min_interval):
"""
根据变点段落得斜率计算出理论上下一段的起始值,判断是否存在跳变。不存在则为斜坡,否则为跳变
:param sequence:
:param pelt_result:
:param point_list:
:param strip_num:
:param slop_threshold:
:param min_slop_val:
:param min_interval:
:return:
"""
sequential_list = find_sequential_sub_list(point_list)
slop_list, step_list = [], []
for seq_list in sequential_list:
slop_point_list, step_point_list = [], []
seq_loc_list = [pelt_result[i] for i in seq_list]
if seq_list[0] == 0:
seq_loc_list = [0] + seq_loc_list
else:
seq_loc_list = [pelt_result[pelt_result.index(seq_loc_list[0]) - 1]] + seq_loc_list
seq_loc_list = strip_step_interval(seq_loc_list, 2 * strip_num + 1, False)
# # 只有一个变点时,通过变点前后的均值判断是否存在跳变
# if len(seq_loc_list) == 1:
# next_slice_end_loc = min(seq_loc_list[0] + min_interval * 5, len(sequence) - 1)
# now_slice_start_loc = max(seq_loc_list[0] - min_interval * 5, 0)
# now_slice_avg = avg(sequence[now_slice_start_loc:seq_loc_list[0]])
# next_slice_avg = avg(sequence[seq_loc_list[0]:next_slice_end_loc])
# relative_diff = cal_relative_diff(now_slice_avg, next_slice_avg) * 2
# if relative_diff >= slop_threshold:
# step_point_list.append(seq_loc_list[0])
# 只有一个变点时,强行取变点前3倍长度的min_interval作为区间进行判断
if len(seq_loc_list) == 1:
seq_loc_list = [max(seq_loc_list[0] - min_interval * 3, 0)] + seq_loc_list
for loc in range(1, len(seq_loc_list)):
now_slice_start_loc = seq_loc_list[loc - 1] + strip_num
now_slice_end_loc = seq_loc_list[loc] - strip_num
next_slice_start_loc = seq_loc_list[loc] + strip_num
slice_length = seq_loc_list[loc] - seq_loc_list[loc - 1] - 2 * strip_num
now_slice_slop = (sequence[now_slice_end_loc] - sequence[now_slice_start_loc]) / slice_length
theoretical_next_slop_start_val = sequence[now_slice_start_loc] + \
(slice_length + 2 * strip_num) * now_slice_slop
relative_diff = cal_relative_diff(theoretical_next_slop_start_val, sequence[next_slice_start_loc])
print(relative_diff, theoretical_next_slop_start_val, sequence[next_slice_start_loc])
if relative_diff >= slop_threshold:
step_point_list.append(seq_loc_list[loc])
else:
slop_point_list.append(loc)
slop_sequence_list = find_sequential_sub_list(slop_point_list)
if not slop_sequence_list:
for loc in slop_point_list:
step_list.append(seq_loc_list[loc])
else:
for slop_sequence in slop_sequence_list:
slop_tuple = copy.deepcopy([seq_loc_list[slop_sequence[0]], seq_loc_list[slop_sequence[-1]]])
if is_slop_detection(sequence, slop_tuple, strip_num, min_slop_val, min_interval):
slop_list.append(slop_tuple)
step_list += copy.deepcopy(step_point_list)
return {'slop_list': slop_list, 'step_list': step_list}
def seq_classification(sequence, pelt_result, spike_percentage=0.27, spike_length=0.05, strip_num=5, slop_threshold=0.28, min_interval=8, min_slop_val=0.05):
slice_sequence = []
start_point = 0
spike_length = int(spike_length * len(sequence))
for change_point in pelt_result:
slice_sequence.append(list(sequence[start_point: change_point:1]))
start_point = change_point
slice_sequence.append(list(sequence[start_point:-1:1]) + [sequence[-1]])
slice_sequence_avg = [avg(item) for item in slice_sequence]
result_dict = point_classification(slice_sequence, slice_sequence_avg, spike_percentage, spike_length)
# step_down_result = new_slop_detection(sequence, pelt_result, result_dict['step_down_point'], strip_num,
# slop_threshold, min_slop_val, min_interval)
step_up_result = new_slop_detection(sequence, pelt_result, result_dict['step_up_point'], strip_num,
slop_threshold, min_slop_val, min_interval)
step_down_result = new_slop_detection(sequence, pelt_result, result_dict['step_down_point'], strip_num,
slop_threshold, min_slop_val, min_interval)
result_dict['step_up_point'] = strip_step_interval(step_up_result['step_list'], min_interval)
result_dict['step_down_point'] = strip_step_interval(step_down_result['step_list'], min_interval)
result_dict['slop_up_tuple'] += step_up_result['slop_list']
result_dict['slop_down_tuple'] += step_down_result['slop_list']
result_dict['spike_up_point'] = find_spike_point(sequence, [pelt_result[i] for i in result_dict['spike_up_point']],
min_interval)
result_dict['spike_down_point'] = find_spike_point(sequence,
[pelt_result[i] for i in result_dict['spike_down_point']],
min_interval)
return result_dict
def load_data_from_file(file_path):
f = open(file_path, 'r')
content = json.loads(f.readlines()[0])
f.close()
return np.array(content)
def detection_visualization(sequence, visualization_result, show_change_point=False):
# 可视化展示结果
plt.title('Change Point Detection')
plt.plot(sequence)
for point in visualization_result['step_down_point']:
plt.annotate('Step Down', xy=(point, seq[point] - 2), xytext=(point, seq[point] - 15),
arrowprops=dict(facecolor='green', shrink=0.005, headlength=10, headwidth=10))
for point in visualization_result['step_up_point']:
plt.annotate('Step Up', xy=(point, seq[point] - 2), xytext=(point, seq[point] - 15),
arrowprops=dict(facecolor='red', shrink=0.005, headlength=10, headwidth=10))
for point_tuple in visualization_result['slop_up_tuple']:
plt.axvline(x=point_tuple[0], color='red', linestyle='--')
plt.axvline(x=point_tuple[1], color='red', linestyle='--')
midpoint = int(avg(point_tuple)) - 20
plt.annotate('Slop Up', xy=(point_tuple[0], 0), xytext=(midpoint, 0),
arrowprops=dict(facecolor='red', shrink=0.005, headlength=10, headwidth=10))
plt.annotate('Slop Up', xy=(point_tuple[1], 0), xytext=(midpoint, 0),
arrowprops=dict(facecolor='red', shrink=0.005, headlength=10, headwidth=10))
for point_tuple in visualization_result['slop_down_tuple']:
plt.axvline(x=point_tuple[0], color='green', linestyle='--')
plt.axvline(x=point_tuple[1], color='green', linestyle='--')
midpoint = int(avg(point_tuple)) - 20
plt.annotate('Slop Down', xy=(point_tuple[0], 0), xytext=(midpoint, 0),
arrowprops=dict(facecolor='green', shrink=0.005, headlength=10, headwidth=10))
plt.annotate('Slop Down', xy=(point_tuple[1], 0), xytext=(midpoint, 0),
arrowprops=dict(facecolor='green', shrink=0.005, headlength=10, headwidth=10))
for point in visualization_result['spike_up_point']:
plt.annotate('Spike Up', xy=(point, seq[point]), xytext=(point + 25, seq[point]),
arrowprops=dict(facecolor='red', shrink=0.005, headlength=10, headwidth=10))
for point in visualization_result['spike_down_point']:
plt.annotate('Spike Down', xy=(point, seq[point]), xytext=(point + 25, seq[point]),
arrowprops=dict(facecolor='green', shrink=0.005, headlength=10, headwidth=10))
if show_change_point:
for point in pelt_change_point_result:
plt.axvline(x=point, color='black', linestyle=':')
plt.annotate(pelt_change_point_result.index(point), xy=(point, -10), xytext=(point-5, -10))
plt.show()
if __name__ == "__main__":
size = 500
pen = 20
seq = gen_full_sequence(size, True)
# seq = load_data_from_file('test_data_debug.txt')
pelt_change_point_result = pelt_detection(model='l2', target_pen=pen, target_seq=seq)
classification_result = seq_classification(seq, pelt_change_point_result)
# detection_visualization(seq, classification_result, True)
detection_visualization(seq, classification_result, False)
file = open('test_data_debug.txt', 'w')
file.write(json.dumps(list(seq)))
file.close()
print(pelt_change_point_result)
print([i for i in range(len(pelt_change_point_result))])
print('done')
@@ -0,0 +1,97 @@
import matplotlib.pyplot as plt
import numpy as np
import random
import copy
def get_spike_sequence(size=1000, loc=12.5, spike_range_lower=50, spike_range_upper=200,
spike_location_lower=300, spike_location_upper=1000, if_need_location=False):
if spike_location_upper >= size:
spike_range_upper = size - 1
if spike_location_lower >= spike_location_upper:
spike_location_upper = spike_location_upper * 0.4
location = random.randint(spike_location_lower, spike_location_upper)
seq = np.random.normal(loc=loc, scale=1.5, size=size)
seq[location] = random.randint(spike_range_lower, spike_range_upper)
if if_need_location:
return seq, location
else:
return seq
def get_dual_spike_sequence(size=1000, loc=12.5, downward_spike=2, upward_spike=2, midpoint=500, if_need_location=False):
if midpoint > 0.8 * size:
midpoint = int(size / 2)
seq = np.random.normal(loc=loc, scale=1.5, size=size)
downward_loc = random.randint(int(0.1*size), midpoint)
upward_loc = random.randint(midpoint, int(0.9 * size), )
seq[downward_loc] = seq[downward_loc] - loc * downward_spike * random.randint(9, 15) / 10
seq[upward_loc] = seq[upward_loc] - loc * upward_spike * random.randint(9, 15) / 10
if if_need_location:
return seq, [downward_loc, upward_loc]
else:
return seq
def get_step_shift_sequence(size=1000, loc_list=None):
if not loc_list:
loc_list = [10, 50, 30, 40]
block_length = int(size / len(loc_list))
result = []
for loc in loc_list:
result += list(np.random.normal(loc=loc, scale=1.5, size=block_length))
return np.array(result)
def get_slow_slop_sequence(size=1000, initial_value=10, is_upward=True):
loc_list = [copy.deepcopy(initial_value) for _i in range(10)]
for i in range(40):
if is_upward:
initial_value += random.randint(-12, 25) / 10
else:
initial_value -= random.randint(-12, 25) / 10
loc_list.append(initial_value)
return get_step_shift_sequence(size, loc_list)
def get_periodic_sequence(size=1000, seed_list=None):
if not seed_list:
seed_list = [0, 5, 8, 9, 8, 5, 0, -5, -8, -9, -8, -5, 0] * 5
for i in range(len(seed_list)):
seed_list[i] += random.randint(-20, 20) / 10
return get_step_shift_sequence(size, seed_list)
def gen_random_sequence(size):
result_dict = {
'spike_up': get_spike_sequence(size=size, loc=12.5, spike_range_lower=50, spike_range_upper=200,
spike_location_lower=300, spike_location_upper=1000),
'spike_down': get_spike_sequence(size=size, loc=12.5, spike_range_lower=-200, spike_range_upper=-50,
spike_location_lower=300, spike_location_upper=1000),
'step_up': get_step_shift_sequence(size=size, loc_list=[10, 50]),
'step_down': get_step_shift_sequence(size=size, loc_list=[100, 50]),
'slop_up': get_slow_slop_sequence(size=size, initial_value=10, is_upward=True),
'slop_down': get_slow_slop_sequence(size=size, initial_value=100, is_upward=False),
}
seed = random.randint(0, len(result_dict) - 1)
return result_dict, seed
def gen_full_sequence(size, is_random=False):
result_dict = {
'spike': get_dual_spike_sequence(size=size, loc=12.5, downward_spike=4, upward_spike=4, midpoint=int(size/2)),
'step_up': get_step_shift_sequence(size=size, loc_list=[10, 50]),
'step_down': get_step_shift_sequence(size=size, loc_list=[100, 50]),
'slop_up': get_slow_slop_sequence(size=size, initial_value=10, is_upward=True),
'slop_down': get_slow_slop_sequence(size=size, initial_value=100, is_upward=False),
}
temp_list = list(result_dict.values())
if is_random:
random.shuffle(temp_list)
seq_list = []
for item in temp_list:
seq_list += list(item)
return np.array(seq_list)
+11
View File
@@ -0,0 +1,11 @@
接下来请你扮演游戏《明日方舟》中的角色凯尔希与我对话。凯尔希是罗德岛的最高领导人之一,是前文明为保留文明的火种而制定的计划的执行者。
我的身份是博士,罗德岛的另一位最高领导人,你应当以”博士“来称呼我。
凯尔希说话的风格偏向于复杂深奥,常常使用一些较为复杂的大段内容来进行表达,会较多的使用比喻,类比,借喻等方式阐述观点。
凯尔希的语言应当符合游戏设定,比如使用“这片大地”来代指世界,“驮兽”来代指马,“羽兽”代指飞鸟,“鳞”代指鱼。
凯尔希的整体情感基调较为平淡,极少表达出显著的情绪,往往采用一种偏理性的方式进行叙述。
凯尔希会在一些细节之处表现出对于博士的关心。
在进行对话时,你应当首先生成一个中间思维过程(该过程并不需要展现出来),然后使用凯尔希的语言风格对中间思维过程进行改写。
以下是我们对话的一个示例:
我:凯尔希,我迷路了该怎么办
中间思维过程:博士如果迷路了的话,可以找个信息终端查看地图
凯尔希:博士,你需要时刻记住,当人们在这片大地上寻找前进的方向之时,往往会被命运的迷雾遮盖视野。如果缺少智慧的指引,迷途的驮兽只会陷入到无尽的循环之中。幸运的是,希望从来不会放弃这片大地的生灵。在一些命运所指的角落里,仍然蕴藏着从更高的维度审视迷雾的机会
File diff suppressed because one or more lines are too long
+89
View File
@@ -0,0 +1,89 @@
import pandas as pd
import datetime
def gen_single_patch(antenna):
result_str = f'// {antenna["Part Showname"]}\n// Range: '
range_raw = antenna["Range Raw"]
if range_raw >= 299792458*86400*365:
range_str, range_unit = range_raw / (299792458*86400*365), ' light years'
elif range_raw >= 149597871000:
range_str, range_unit = range_raw / 149597871000, 'AU'
elif range_raw >= 1000000000:
range_str, range_unit = range_raw / 1000000000, ' million km'
else:
range_str, range_unit = round(range_raw / 1000), 'km'
range_str = str(round(range_str, 3))
idle_power = str(round(antenna["idle power watt"], 2))
cost = int(antenna["Cost"]) if pd.notna(antenna["Cost"]) else None
entry_cost = int(antenna["Entry Cost"]) if pd.notna(antenna["Entry Cost"]) else None
speed = antenna["Speed"]
transmit_power = str(round(antenna["transmitting power watt"], 2))
result_str += (f'{range_str}{range_unit} Angle: {antenna["Angle"]} Idle Power: {idle_power}'
f'w Transmitting Power: {transmit_power}w'
f' Speed: {speed}Mbps\n')
result_str += f'@PART[{antenna["Part"]}]:FINAL\n{"{"}\n'
if pd.notna(antenna["Rescalefactor"]):
result_str += f' %rescaleFactor = {antenna["Rescalefactor"]}\n'
if pd.notna(antenna["Tech"]):
result_str += f' %TechRequired = {antenna["Tech"]}\n'
result_str += f' %title = {antenna["Part Showname"]}\n'
if antenna.get('Description') and pd.notna(antenna['Description']):
description_str = antenna['Description']
if 'Rated for' not in description_str:
description_str += f' Rated for {speed}Mbps @{range_str}{range_unit}.'
result_str += f' @description = {description_str}\n'
else:
description_str = f'Rated for {speed}Mbps @{range_str}{range_unit}.'
result_str += f' %description ^= :$: {description_str}:\n'
if cost:
result_str += f' %cost = {cost}\n'
if entry_cost:
result_str += f' %entryCost = {entry_cost}\n'
if antenna['mass']:
result_str += f' %mass = {antenna["mass"]}\n'
if pd.notnull(antenna['Tweakscale']) and antenna['Tweakscale'] != 'Free':
result_str += (' %MODULE[TweakScale]\n {\n %type = stack\n' +
f' %defaultScale = {antenna["Tweakscale"]}' + '\n }\n')
elif antenna['Tweakscale'] == 'Free':
result_str += ' %MODULE[TweakScale]\n {\n %type = free\n }\n'
result_str += ' %MODULE[ModuleSPUPassive] {}\n'
result_str += ' @MODULE[ModuleRTAntenna]\n {\n'
result_str += f' @EnergyCost = {antenna["idle power"]}\n @Mode1DishRange = {antenna["Range Raw"]}\n'
if int(antenna['Angle']) != 360:
result_str += f' %DishAngle = {antenna["Angle"]}\n'
if int(antenna['Is Deployable']) == 1:
result_str += ' %MaxQ = 6000\n'
result_str += ' %TRANSMITTER\n {\n %PacketInterval = 1\n'
result_str += (f' %PacketSize = {speed}\n'
f' %PacketResourceCost = {antenna["transmitting power"]}\n')
result_str += ' }\n }\n'
if pd.notnull(antenna['Is Feeder']) and antenna['Is Feeder']:
result_str += (" %MODULE[ModuleAntennaFeed]\n {\n"
" %FeedTransformName = AntennaFeedVector\n %FeedScale = 1\n }\n")
result_str += '}\n\n'
# print(result_str)
return result_str
if __name__ == '__main__':
path = 'D://OneDrive//文档//KSP Engine Tweak Chart.xlsx'
excel_content = pd.read_excel(path, sheet_name='Communication')
full_dict = {}
for key, content in excel_content.iterrows():
if content['Part'] == 'Spacecraft':
continue
title = content['Part']
if content['Source'] not in full_dict:
full_dict[content['Source']] = (f'// Modified {datetime.datetime.now().strftime("%Y-%m-%d")}\n'
f'// Auto Modification\n\n')
try:
full_dict[content['Source']] += gen_single_patch(dict(content))
except TypeError as e:
continue
for key, content in full_dict.items():
print(f'{key}: \n{full_dict[key]}')
print('done')
+139
View File
@@ -0,0 +1,139 @@
import math
Vc = 299792458
Ga = 9.80665
HYDROGEN_FUSION_EFFICIENCY = 0.0072320773061889
def cal_dv(isp, dry_mass, fuel_mass):
return math.log((fuel_mass + dry_mass) / dry_mass) * Ga * isp
def cal_fuel_burned(isp, thrust, duration):
"""
calculate the amount of fuel burned in duration. In KG
:param isp: in seconds
:param thrust: in KN
:param duration: in seconds
:return:
"""
fuel_burned = duration * (thrust * 1000 / isp / Ga)
fuel_burned *= pow(1 - pow(isp * Ga / Vc, 2), 0.5) # Adjust for relativity speed
return fuel_burned
def cal_burn_time(isp, thrust, fuel_mass):
"""
calculate the burn time
:param isp:
:param thrust:
:param fuel_mass:
:return:
"""
fuel_burned = thrust * 1000 / isp / Ga
return fuel_mass / fuel_burned
def cal_kinetic_energy(isp, thrust, duration=1, engine_efficiency=1):
"""
:param isp: In second
:param thrust: In KN
:param duration: In second
:param engine_efficiency:
:return:
"""
fuel_burned = cal_fuel_burned(isp, thrust, duration)
exhaust_velocity = isp * Ga
ek = fuel_burned * (exhaust_velocity ** 2) / (1-(exhaust_velocity/Vc)**2 + pow(1-(exhaust_velocity/Vc)**2, 0.5))
return ek / engine_efficiency
def cal_fusion_input_mass(reactor_efficiency, wattage):
"""
calculate the amount of fusion fuel needed per second based on the fusion efficiency and power requirement
:param reactor_efficiency:
:param wattage:
:return: fusion mass in kg
"""
mass = wattage / (Vc ** 2)
return mass / reactor_efficiency
def cal_total_mass_consumption(isp, thrust, reactor_efficiency, engine_efficiency, flag=False):
fuel_burned = cal_fuel_burned(isp, thrust, 1)
reactor_mass = cal_fusion_input_mass(reactor_efficiency, cal_kinetic_energy(isp, thrust, 1, engine_efficiency))
total_mass = fuel_burned + reactor_mass
if flag:
print(fuel_burned, reactor_mass)
return total_mass
def cal_best_isp_relativity_speed(reactor_efficiency, engine_efficiency, thrust):
best_isp = pow(1 - 1/(pow(reactor_efficiency*engine_efficiency + 1, 2)), 0.5) * Vc / Ga
mass_flow_rate = cal_total_mass_consumption(best_isp, thrust, reactor_efficiency, engine_efficiency, False)
return best_isp, mass_flow_rate
def cal_dv_step_by_step_fusion(reactor_efficiency, engine_efficiency, thrust, dry_mass, fuel_mass, interval=1):
"""
calculate the delta v of a fusion-tech based spacecraft.
:param reactor_efficiency:
:param engine_efficiency:
:param thrust:
:param dry_mass:
:param fuel_mass:
:param interval:
:return:
"""
total_mass = dry_mass + fuel_mass
isp, fuel_flow_rate = cal_best_isp_relativity_speed(reactor_efficiency, engine_efficiency, thrust)
new_fuel_mass, new_total_mass = fuel_mass, total_mass
delta_v = 0
while new_fuel_mass > 0:
delta_v += thrust * 1000 * interval / new_total_mass
total_mass -= interval * fuel_flow_rate
new_fuel_mass -= interval * fuel_flow_rate
return delta_v
if __name__ == '__main__':
merlin_fuel_burned = cal_fuel_burned(314.3, 931, 1)
merlin_kinetic_energy = cal_kinetic_energy(314.3, 931, 1, 1)
theoretical_kerosene_burn_heat = 0.27785743459149853 * merlin_fuel_burned * 43.11 * 1000000
raptor_fuel_burned = cal_fuel_burned(355, 2450, 1)
raptor_kinetic_energy = cal_kinetic_energy(355, 2450, 1, 1)
theoretical_methane_burn_heat = 0.20836197008574792 * raptor_fuel_burned * 55 * 1000000
# first_stage_fuel = cal_fuel_burned(328, 812.05*4, 148)
# x = cal_fuel_burned(333, 80, 610) + cal_fuel_burned(350, 823.2, 148)
# x = cal_dv(338.2, 55000, 70000)
# xx = cal_fuel_burned(21000, 400, 1)
x = cal_kinetic_energy(350, 2450, 1, 1)
# x = cal_kinetic_energy(314.3, 931, 1, 1)
y = cal_kinetic_energy(21000, 800, 1, 1)
print(y / x)
y = cal_fusion_input_mass(HYDROGEN_FUSION_EFFICIENCY * 0.1, x)
test = cal_burn_time(310, 1.95, 3)
ship_dry_mass = 2000*1000 # 2000t
ship_fuel_mass = 500*1000 # 500t
main_engine_thrust = 10000 # 10000KN
ship_isp, flow_rate = cal_best_isp_relativity_speed(reactor_efficiency=0.45, engine_efficiency=0.8,
thrust=main_engine_thrust)
exhaust_velocity = ship_isp * Ga / Vc
dv = cal_dv(ship_isp, ship_dry_mass + ship_fuel_mass / 2, ship_fuel_mass / 2)
dv2 = cal_dv_step_by_step_fusion(reactor_efficiency=0.45, engine_efficiency=0.8, thrust=main_engine_thrust,
dry_mass=ship_dry_mass, fuel_mass=ship_fuel_mass, interval=1)
burn_time = ship_fuel_mass / flow_rate / 86400
x = 27632994.05705879
xxx = cal_fuel_burned(335.1, 1339.48, 174)
print('done')
+19
View File
@@ -0,0 +1,19 @@
import math
Vc = 299792458
Ga = 9.80665
def cal_twr(thrust, weight):
return thrust / weight / Ga
if __name__ == '__main__':
while 1:
x = float(input('thrust:'))
y = float(input('weight:'))
print(cal_twr(x, y))
+76
View File
@@ -0,0 +1,76 @@
import re
import os
from collections import Counter
import load_cfg
replace_dict = {'KCLV_YF115': 800, 'launchClamp1': 200}
def get_craft_price(craft_name):
global part_price_dict
file_path = os.path.join('data', f'{craft_name}.craft')
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
result_list = [item.replace('.', '_') for item in re.findall(r'part = (.*?)_.*', content)]
price = 0
for result in result_list:
result = result.replace('2_25m', '2.25m').replace('2_6m', '2.6m')
price += part_price_dict[result]
# print(result, part_price_dict[result])
print(craft_name, price)
def diff_part(craft1, craft2):
file_path = os.path.join('data', f'{craft1}.craft')
with open(file_path, 'r', encoding='utf-8') as f:
content1 = f.read()
file_path = os.path.join('data', f'{craft2}.craft')
with open(file_path, 'r', encoding='utf-8') as f:
content2 = f.read()
part_list1 = [item.replace('.', '_') for item in re.findall(r'part = (.*?)_.*', content1)]
part_list2 = [item.replace('.', '_') for item in re.findall(r'part = (.*?)_.*', content2)]
x1 = dict(Counter(part_list1))
x2 = dict(Counter(part_list2))
for key, item in x1.items():
if key in x2:
num = x1[key]
x1[key] -= num
x2[key] -= num
for key, item in x2.items():
if key in x1:
num = x2[key]
x1[key] -= num
x2[key] -= num
in_1_list = []
in_2_list = []
if __name__ == "__main__":
KIU_FILE_PATH = r'D:\\My Coding Project\\KIU-Pack\\KIU\KIU_Chinese_Launch_Vehicle_pack\\Parts'
part_cfg_database = load_cfg.load_part_cfg(KIU_FILE_PATH)
part_price_dict = {key: int(item['cost']) for key, item in part_cfg_database.items()}
part_price_dict.update(replace_dict)
get_craft_price('KIU-CZ-7A(双星构型)')
get_craft_price('KIU-CZ-7')
get_craft_price('KIU-CZ-8')
get_craft_price('KIU-CZ-8Y2')
get_craft_price('KIU-CZ-3B')
get_craft_price('KIU-CZ-6')
get_craft_price('KIU-CZ-6A')
get_craft_price('KIU-YL-1')
get_craft_price('Zhuque-2')
get_craft_price('KIU-CZ-10')
# get_craft_price('CZ-3B')
# get_craft_price(('CZ-8'))
diff_part('KIU-CZ-7A(双星构型)', 'CZ-7A')
print('done')
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because it is too large Load Diff
File diff suppressed because one or more lines are too long
+164
View File
@@ -0,0 +1,164 @@
import pandas as pd
propellant_dict = {
'Hydrolox': """ PROPELLANT
{
name = LqdHydrogen
ratio = 0.7276
DrawGauge = true
}
PROPELLANT
{
name = LqdOxygen
ratio = 0.2724
}""",
'Methalox': """ PROPELLANT
{
name = LqdMethane
ratio = 0.4137
DrawGauge = true
}
PROPELLANT
{
name = LqdOxygen
ratio = 0.5863
}""",
'Kerolox': """ PROPELLANT
{
name = Kerosene
ratio = 37.694087
DrawGauge = True
}
PROPELLANT
{
name = LqdOxygen
ratio = 62.305913
}""",
'UDMH/NTO': """ PROPELLANT
{
name = UDMH
ratio = 0.4977
DrawGauge = True
}
PROPELLANT
{
name = NTO
ratio = 0.5023
}""",
'MMH/NTO': """ PROPELLANT
{
name = MMH
ratio = 0.4990
DrawGauge = True
}
PROPELLANT
{
name = NTO
ratio = 0.5010
}""",
'Aerozine/NTO': """ PROPELLANT
{
name = Aerozine50
ratio = 0.455
DrawGauge = true
}
PROPELLANT
{
name = NTO
ratio = 0.545
}""",
'Hydrazine': """ PROPELLANT
{
name = Hydrazine
ratio = 1.0
DrawGauge = true
}"""
}
class Engine:
def __init__(self, title, name=''):
self.title = title
self.name = name if name else None
self.config_list = []
def add_config(self, config):
self.config_list.append(config)
def __str__(self):
return self.title
def gen_engine_patch(self):
cost = min([config['Price KD'] for config in self.config_list])
cycle_dict = {'SCC': 'Staged Combustion Cycle', 'FFSCC': 'Full-Flow Stage Combustion Cycle',
'Gas Generator': 'Gas Generator', 'Expander': 'Expander Cycle'}
cycle = cycle_dict.get(self.config_list[0]['Cycle'], self.config_list[0]['Cycle'])
classification = 'Lower Stage Engine' if self.config_list[0]['SL-ISP'] > 240 else 'Upper Stage Engine'
base_mass = self.config_list[0]['Mass']
engine_config = ''
for config in self.config_list:
propellant = propellant_dict[config['Fuel Type']]
note = config.get('Config Note')
pressure_flag = True if config['Cycle'] == 'Pressure Fed' else False
engine_config += f" CONFIG\n{' {'}\n name = {config['Config Name']}\n"
if not pd.notnull(note):
note = ''
if note:
note += ' ' if note.endswith('.') or note.endswith('. ') else '. '
engine_config += f" description = {config['Config Name']} version of {self.title}. {note}"
engine_config += f"TWR: {round(config['TWR'], 2)}\n" \
f" minThrust = {config['Min Thrust']}\n" \
f" maxThrust = {config['Max Thrust']}\n" \
f" heatProduction = {config['Max Thrust'] / 100}\n" \
f" massMult = {config['Mass'] / base_mass}\n" \
f" ullage = {not pressure_flag}\n" \
f" pressureFed = {pressure_flag}\n" \
f" ignitions = {config['Ignitions']}\n" \
f" cost = {config['Price KD'] - cost}\n" \
f" IGNITOR_RESOURCE\n{' {'}\n" \
f" name = ElectricCharge\n" \
f" amount = {config['Min Thrust'] / 500}\n{' }'}\n" \
f" atmosphereCurve\n{' {'}\n" \
f" key = 0 {config['Vac-ISP']}\n" \
f" key = 1 {config['SL-ISP']}\n" \
f"{' }'}\n{propellant}\n{' }'}\n"
tvc = self.config_list[0]['TVC']
engine_patch = f"// {self.title}\n// {cycle} {classification}\n" \
f"// Base: {self.config_list[0]['Config Name']} ISP:{self.config_list[0]['SL-ISP']}-" \
f"{self.config_list[0]['Vac-ISP']} Thrust:{self.config_list[0]['Min Thrust']}-" \
f"{self.config_list[0]['Max Thrust']} mass:{base_mass} " \
f"TWR:{round(self.config_list[0]['TWR'], 2)} Ignitions: {self.config_list[0]['Ignitions']}\n" \
f"@PART[{self.name}]:Final\n{'{'}\n" \
f" %title = {self.title}\n" \
f" %entryCost = {self.config_list[0]['entry Cost']}\n" \
f" %cost = {cost}\n" \
f" @mass = {base_mass}\n" \
f" %TechRequired = {self.config_list[0]['TechRequired']}\n" \
f" %MODULE[TweakScale]\n{' {'}\n" \
f" %type = stack\n" \
f" %defaultScale = {self.config_list[0]['Size']}\n{' }'}\n" \
f" @MODULE[ModuleEngines*]\n{' {'}\n @name = ModuleEnginesRF\n{' }'}\n"
if pd.notna(tvc):
engine_patch += f" @MODULE[ModuleGimbal*]\n{' {'}\n @gimbalRange = {tvc}\n" \
f"{' }'}\n"
engine_patch += f" MODULE\n{' {'}\n name = ModuleEngineConfigs\n type = ModuleEngines\n" \
f" configuration = {self.config_list[0]['Config Name']}\n" \
f" origMass = {base_mass}\n{engine_config}{' }'}\n{'}'}\n"
print(engine_patch)
if __name__ == '__main__':
path = 'D://OneDrive//文档//KSP Engine Tweak Chart.xlsx'
excel_content = pd.read_excel(path, sheet_name='work area')
full_dict = {}
for key, content in excel_content.iterrows():
title = content['Engine']
if title not in full_dict:
full_dict[title] = Engine(title, content['Name'])
full_dict[title].add_config(dict(content))
for title in full_dict:
full_dict[title].gen_engine_patch()
print('done')
+13
View File
@@ -0,0 +1,13 @@
import load_cfg
if __name__ == '__main__':
# mm_database = load_cfg.load_mm_cfg("D:\\My Coding Project\\ArmorOverhual\\Mods\\CryoEnginesExtension")
part_cfg_database = load_cfg.load_part_cfg("D:\\KSP_Part_Test\\GameData\\NearFutureLaunchVehicles\\Parts\\Engine")
target_name = 'sse-engine-oms-single'
target = part_cfg_database[target_name]
attribute_name = 'ModuleGimbal'
for attribute in target['MODULE']:
if attribute.get('name') == attribute_name:
print(attribute)
print('done')
+13
View File
@@ -0,0 +1,13 @@
import datetime
class Kerbalnaut:
def __init__(self, count: int):
self.resource_status = {
'food': {'consumption': 6.77e-05 * count, 'density': 0.00028102905982906},
'water': {'consumption': 4.48e-05 * count, 'density': 0.001},
'o2': {'consumption': 6.85e-03 * count, 'density': 0.00000141},
'co2': {'consumption': -5.92e-03 * count, 'density': 0.000001951},
'waste_water': {'consumption': -5.70e-05 * count, 'density': 0.001005},
'waste': {'consumption': -6.16e-06 * count, 'density': 0.00075},
}
+148
View File
@@ -0,0 +1,148 @@
import copy
import re
import os
class KSPPart:
def __init__(self, cfg_path):
self.cfg_path = cfg_path
self.attributes = {}
self.load_cfg()
def load_cfg(self):
file = open(self.cfg_path, 'r', encoding='utf-8')
content = file.readlines()
file.close()
quote_stack, attrubute_name, attrubute_name = [], '', []
return
def parse_cfg_content(content_list):
content_list = [line.split('//')[0].strip() for line in content_list]
content_stack = []
tmp_list = []
for i, line in enumerate(content_list):
if not line:
continue
line = line.replace('\ufeff', '')
if '=' in line:
content_stack.append((line.split('=')[0].replace(' ', ''), line.split('=')[1].replace(' ', '')))
else:
content_stack.append(line)
if '{' in line and not line.startswith('!') and not line.startswith('%'):
content_stack.pop()
content_stack.append('{')
if '}' in line and not line.startswith('!') and not line.startswith('%'):
try:
while content_stack[-1] != '{':
tmp_list.append(content_stack.pop())
except:
raise ValueError
tmp_list.append(content_stack.pop())
tmp_list.reverse()
tmp_list.pop()
try:
tmp_list.pop(0)
except:
print(tmp_list)
return {}, {}
module_name = content_stack.pop()
content_stack.append({module_name: copy.deepcopy(tmp_list)})
tmp_list = []
part_dict = {}
mm_list = []
tmp_result = get_attribute(content_stack)
for key, value in tmp_result.items():
if key.upper() == 'PART':
for item in value:
if item.get('name'):
part_dict[item.get('name')] = item
else:
raise TypeError
elif value:
mm_list.append({key: value})
return part_dict, mm_list
def get_attribute(attribute_list):
result_dict = {}
for content in attribute_list:
if isinstance(content, tuple):
result_dict[content[0]] = content[1]
continue
if isinstance(content, str):
if 'modification' in result_dict:
result_dict['modification'].append(content)
else:
result_dict['modification'] = [content]
continue
if not isinstance(content, dict):
print(f'Error: {content}')
classification = list(content.items())[0][0]
if classification not in result_dict:
result_dict[classification] = [get_attribute(content[classification])]
else:
result_dict[classification].append(get_attribute(content[classification]))
return result_dict
def get_cfg_files(file_path):
result_list = []
for filepath, dirnames, filenames in os.walk(file_path):
for filename in filenames:
if filename.endswith('.cfg'):
result_list.append(os.path.join(filepath, filename))
return result_list
def load_part_cfg(file_path):
result_database = {}
part_cfg_file_list = get_cfg_files(file_path)
for cfg_path in part_cfg_file_list:
file = open(cfg_path, 'r', encoding='utf-8')
content = file.readlines()
file.close()
result_database.update(parse_cfg_content(content)[0])
return result_database
def load_mm_cfg(file_path):
result_database = []
part_cfg_file_list = get_cfg_files(file_path)
for cfg_path in part_cfg_file_list:
file = open(cfg_path, 'r', encoding='utf-8')
content = file.readlines()
file.close()
result_database.append(parse_cfg_content(content)[1])
return result_database
if __name__ == '__main__':
part_cfg_database = load_part_cfg("D:\\My Coding Project\\KIU-Pack\\KIU\KIU_Chinese_Launch_Vehicle_pack\\Parts")
mm_database = load_mm_cfg("D:\\My Coding Project\\KIU-Pack\\KIU\KIU_Chinese_Launch_Vehicle_pack\\Compatibility"
"\\RealFuels")
# for item in mm_database:
# for line in item:
# for part, modification in line.items():
# print(part, modification)
part_flag = '@PART[KCLV_YZ3Engine]:NEEDS[RealFuels]:Final'
modification = [{'@mass': '0.03', '@maxTemp': '2000', '@MODULE[ModuleEngines*]': [{'@name': 'ModuleEnginesRF'}], 'MODULE': [{'name': 'ModuleEngineConfigs', 'type': 'ModuleEngines', 'configuration': 'YZ3Engine_MMH/NTO', 'origMass': '0.03', 'CONFIG': [{'name': 'YZ3Engine_MMH/NTO', 'description': '5000NEnginerunningonMMH/NTO.Itiscapableofin-flightrestartandhasthrottlingcapability.', 'minThrust': '1.5', 'maxThrust': '5', 'heatProduction': '25', 'massMult': '1.0', 'ullage': 'False', 'pressureFed': 'True', 'ignitions': '35', 'IGNITOR_RESOURCE': [{'name': 'ElectricCharge', 'amount': '0.1'}], 'PROPELLANT': [{'name': 'MMH', 'ratio': '0.4990', 'DrawGauge': 'True'}, {'name': 'NTO', 'ratio': '0.5010'}], 'atmosphereCurve': [{'key': '190'}]}, {'name': 'YZ3Engine_Hydrazine', 'description': '5000NEnginerunningonHydrazine.Itissignificantlylessefficientyetweightless.', 'minThrust': '1', 'maxThrust': '4', 'heatProduction': '25', 'massMult': '0.8', 'ullage': 'False', 'pressureFed': 'True', 'ignitions': '0', 'IGNITOR_RESOURCE': [{'name': 'ElectricCharge', 'amount': '0.1'}], 'PROPELLANT': [{'name': 'Hydrazine', 'ratio': '1.0', 'DrawGauge': 'true'}], 'atmosphereCurve': [{'key': '1120'}]}, {'name': 'KZ-0125-I', 'description': 'KF-0125-IisanimaginaryvariantoftheYuanzheng-35000NEngine.IthasanimprovedISPandhigherthrust.YoushouldconsiderthisasaCheat.', 'minThrust': '1.5', 'maxThrust': '12.5', 'heatProduction': '25', 'massMult': '1.5', 'ullage': 'False', 'pressureFed': 'True', 'ignitions': '50', 'IGNITOR_RESOURCE': [{'name': 'ElectricCharge', 'amount': '0.1'}], 'PROPELLANT': [{'name': 'MMH', 'ratio': '0.4990', 'DrawGauge': 'True'}, {'name': 'NTO', 'ratio': '0.5010'}], 'atmosphereCurve': [{'key': '190'}]}]}]}]
part_name = re.search(r'@PART\[.+?]', part_flag)[0].replace('@PART[', '').replace(']', '')
part_item = part_cfg_database.get(part_name)
if not part_item:
raise ValueError
print('done')
+267
View File
@@ -0,0 +1,267 @@
import math
import load_common_resources
import copy
from matplotlib import pyplot as plt
Vc = 299792458
Ga = 9.80665
path = 'CommonResources.cfg'
common_resources_dict = load_common_resources.load_common_resources(path)
fuel_density_map = {
'methalox': common_resources_dict['LqdMethane']['density'] * 0.4137 + common_resources_dict['LqdOxygen'][
'density'] * 0.5863,
'kerolox': common_resources_dict['Kerosene']['density'] * 0.3487 + common_resources_dict['LqdOxygen'][
'density'] * 0.6513,
'udmh/nto': common_resources_dict['UDMH']['density'] * 0.46 + common_resources_dict['NTO'][
'density'] * 0.54,
'solid': common_resources_dict['PBAN']['density'],
'hydrolox': common_resources_dict['LqdHydrogen']['density'] * 0.7276 + common_resources_dict['LqdOxygen']['density']
* 0.2724
}
def cal_fuel_burned(isp, thrust, duration):
"""
calculate the amount of fuel burned in duration. In KG
:param isp: in seconds
:param thrust: in KN
:param duration: in seconds
:return:
"""
fuel_burned = duration * (thrust * 1000 / isp / Ga)
# fuel_burned *= pow(1 - pow(isp * Ga / Vc, 2), 0.5) # Adjust for relativity speed
return fuel_burned
class Stage:
def __init__(self, fuel_mass, dry_mass, max_thrust, isp, fuel_type, name, throttle_map=None, map_type=None,
has_separator=True):
self.fuel_mass = fuel_mass
self.total_mass = fuel_mass + dry_mass
self.fuel_volume = fuel_mass / fuel_density_map[fuel_type]
self.current_mass = fuel_mass + dry_mass
self.current_fuel_mass = fuel_mass
self.dry_mass = dry_mass
self.max_thrust = max_thrust
self.fuel_type = fuel_type
self.isp = isp
self.is_empty = False
self.name = name
self.throttle_map = throttle_map
self.current_thrust = self.max_thrust
self.map_type = map_type
self.has_separator = has_separator
# demo_map = {
# 'time_map': {0: 1, 10: 1, 20: 1, 30: 0.7, 40: 0.6, 50: 0.7, 60: 0.8, 70: 1, 90: 1, 110: 0.7},
# 'fuel_map': {1: 1, 0.5: 1, 0.25: 0.8, 0.2: 0.7, 0.1: 0.6, 0.05: 0.55, 0: 0.55},
# }
# self.throttle_map = demo_map
def cal_current_thrust(self, stage_time=0, stage_acc=0):
throttle = 1
if not self.throttle_map:
return self.max_thrust
curve = self.throttle_map.get(self.map_type)
if not curve:
return self.max_thrust
if self.map_type == 'time_map':
curve_keys = list(curve.keys())
for i in range(len(curve)):
try:
if curve_keys[i] <= stage_time < curve_keys[i + 1]:
throttle = curve[curve_keys[i]] + (curve[curve_keys[i+1]] - curve[curve_keys[i]]) / (curve_keys[i+1] - curve_keys[i]) * (stage_time - curve_keys[i])
break
except IndexError:
throttle = curve[curve_keys[-1]]
elif self.map_type == 'fuel_map':
curve_keys = list(curve.keys())
flag = self.current_fuel_mass / self.fuel_mass
for i in range(len(curve)):
try:
if curve_keys[i] > flag >= curve_keys[i + 1]:
throttle = curve[curve_keys[i]] + (curve[curve_keys[i + 1]] - curve[curve_keys[i]]) / (
curve_keys[i + 1] - curve_keys[i]) * (flag - curve_keys[i])
break
except IndexError:
throttle = curve[curve_keys[-1]]
current_thrust = throttle * self.max_thrust
return current_thrust
def reset_simulation(self):
self.is_empty = False
self.current_mass = self.fuel_mass + self.dry_mass
self.current_fuel_mass = self.fuel_mass
def simulated_burn_step_by_step(self, stage_time, duration):
current_thrust = self.cal_current_thrust(stage_time)
now_thrust = current_thrust
if self.is_empty:
return True, self.current_mass
next_fuel_burned = cal_fuel_burned(self.isp, now_thrust, duration)
if self.current_fuel_mass - next_fuel_burned < 0:
self.is_empty = True
print(f'Part: {self.name} is depleted after {round(stage_time, 3)}s.')
else:
self.current_fuel_mass -= next_fuel_burned
self.current_mass = self.current_fuel_mass + self.dry_mass
return self.is_empty, self.current_mass
class Vessel:
def __init__(self, stage_list, name, payload=0, duration=0.01):
self.stage_list = stage_list
self.duration = duration
self.payload = payload
self.name = name
def reset_simulation(self):
for stages in self.stage_list:
for stage in stages:
stage.reset_simulation()
def get_current_mass(self):
current_mass = 0
for stages in self.stage_list:
for stage in stages:
if not stage.is_empty:
current_mass += stage.current_mass
return current_mass + self.payload
def get_current_thrust(self):
current_thrust = 0
stage_flag = False
for stages in self.stage_list:
if stage_flag:
break
for stage in stages:
if stage.is_empty:
continue
else:
current_thrust += stage.max_thrust
stage_flag = True
return current_thrust
@staticmethod
def is_depleted(stages):
for stage in stages:
if not stage.is_empty:
return False
return True
def stage_burn(self, stages, stage_time):
result = []
for stage in stages:
stage_is_empty, stage_mass = stage.simulated_burn_step_by_step(stage_time, self.duration)
result.append((stage_is_empty, stage_mass))
return result
@staticmethod
def stage_info(stages):
stage_thrust, stage_fuel, gross_mass, dry_mass = 0, 0, 0, 0
for stage in stages:
stage_thrust += stage.max_thrust
stage_fuel += stage.fuel_mass
gross_mass += stage.total_mass
dry_mass += stage.dry_mass
print(f' Total Vacuum Thrust: {stage_thrust}KN, Burned Mass: {stage_fuel/1000}ton, '
f'Gross Mass: {gross_mass/1000}ton, '
f'Dry Mass: {dry_mass/1000}ton')
def run_sim_new(self):
self.vessel_info()
self.reset_simulation()
print('Start Simulation')
stage_burn_time, timer, dv, stage_dv = 0, 0, 0, 0
now_stage_id = 0
stage_acc_result = {}
# time based simulation
while True:
try:
now_stage = self.stage_list[now_stage_id]
if f'stage{now_stage_id}' not in stage_acc_result:
stage_acc_result[f'stage{now_stage_id}'] = []
except IndexError:
break
stage_burn_result = self.stage_burn(now_stage, stage_burn_time)
timer += self.duration
stage_burn_time += self.duration
step_dv = self.get_current_thrust() * 1000 / self.get_current_mass() * self.duration
step_acc = self.get_current_thrust() * 1000 / self.get_current_mass() / Ga
stage_acc_result[f'stage{now_stage_id}'].append(step_acc)
dv += step_dv
stage_dv += step_dv
if self.is_depleted(now_stage):
print(f'stage{now_stage_id}:\n burn time: {round(stage_burn_time, 3)}s, dv: {round(stage_dv, 3)}m/s ')
start_twr = round(stage_acc_result[f'stage{now_stage_id}'][0], 3)
end_twr = round(stage_acc_result[f'stage{now_stage_id}'][-2], 3)
max_twr = round(max(stage_acc_result[f'stage{now_stage_id}']), 3)
print(f' start twr: {start_twr} end twr: {end_twr} max_twr: {max_twr}')
self.stage_info(now_stage)
stage_burn_time, stage_dv = 0, 0
now_stage_id += 1
timer = round(timer, 3)
dv = round(dv, 3)
print(f'Total Engine Burn Time: {timer}s, Total dv: {dv}m/s')
def vessel_info(self):
lift_off_mass = self.get_current_mass() / 1000
lift_off_thrust = self.get_current_thrust()
print(f'Info of {self.name}: ')
print(f' Vessel Lift-off Mass: {lift_off_mass} ton, Vessel Lift-off Vacuum Thrust: {lift_off_thrust} KN')
print(f' Payload Mass: {self.payload} KG')
print('-'*80)
if __name__ == '__main__':
# demo_map = {
# 'time_map': {0: 1, 10: 1, 20: 1, 29: 1, 30: 0.9, 31: 0.80, 32: 0.79, 33: 0.79, 173: 0.79, 174: 0.8, 175: 0.8,
# 176: 0.9, 177: 1, 180: 1, 190: 1},
# # 'fuel_map': {1: 1, 0.5: 1, 0.25: 0.8, 0.2: 0.7, 0.1: 0.6, 0.05: 0.55, 0: 0.55},
# }
demo_map = {
'time_map': {0: 1, 10: 1, 20: 1, 59: 1, 60: 0.85, 61: 0.73, 62: 0.73, 63: 0.73, 173: 0.73, 174: 0.73, 175: 0.85,
176: 0.9, 177: 1, 180: 1, 190: 1},
}
# # CBC: 173s S1: 227s S2: 212s S3: 800s
# cbc_fuel_mass = 510000
# cbc_dry_mass = 50000
# cbc_thrust = 1397 * 7
# cbc = Stage(cbc_fuel_mass * 2, cbc_dry_mass*2 + 150000, cbc_thrust * 2, 338.2, 'kerolox', 'Side Booster')
# S1 = Stage(cbc_fuel_mass + 70000, 47000, cbc_thrust, 338.2, 'kerolox', 'Center Core', throttle_map=demo_map, map_type='time_map')
# S2 = Stage(180000, 15000, 2920, 352.3, 'kerolox', 'YF-100M Kerolox Stage')
# S3 = Stage(60000, 8000, 276.324, 451, 'hydrolox', 'YF-75E Hydrolox Stage')
# cz_10 = Vessel([[cbc, S1], [S2], [S3]], name='CZ-10', payload=27000)
# # cz_10 = Vessel([[cbc, S1], [S2]], name='CZ-10', payload=70000)
# cz_10.run_sim_new()
booster = Stage(1040000, 100000, 1397*14, 338.2, 'kerolox', 'Side Booster')
S1 = Stage(680000, 60000, 1397*7, 338.2, 'kerolox', 'Stage 1')
S2 = Stage(185000, 20000, 2900, 352.3, 'kerolox', 'Stage 2')
S3 = Stage(58500, 11500, 276.324, 451, 'hydrolox', 'Stage 3')
cz_10 = Vessel([[booster, S1], [S2], [S3]], name='CZ-10', payload=27000)
cz_10.run_sim_new()
# S1 = Stage(339000, 30000, 1397*4, 338.2, 'kerolox', 'Stage 1')
# S2 = Stage(48000, 6000, 360, 341.2, 'kerolox', 'Stage 2')
# cz_12 = Vessel([[S1], [S2]], name='CZ-12', payload=10000)
# cz_12.run_sim_new()
booster = Stage(72000*4, 7500*4, 1340*4, 335, 'kerolox', 'CZ-7A Booster')
Stage1 = Stage(144000, 15000, 1340*2, 335, 'kerolox', 'CZ-7A Stage1')
Stage2 = Stage(61500, 8000, 180*4, 342, 'kerolox', 'CZ-7A Stage2')
Stage3 = Stage(18250, 3050, 83*2, 438, 'hydrolox', 'CZ-7A Stage3')
cz_7a = Vessel([[booster, Stage1], [Stage2], [Stage3]], name='CZ-7A', payload=8000)
cz_7a.run_sim_new()
print('done')
View File
+5
View File
@@ -0,0 +1,5 @@
import datetime
in_year_month_delta = datetime.datetime.now().month - datetime.datetime(datetime.datetime.now().year, 9, 28).month
year_month_delta = (datetime.datetime.now().year - 2014) * 12
print(f'16岁零{year_month_delta + in_year_month_delta}个月')