Basic mission vessel tunnel
This commit is contained in:
16
connector.py
Normal file
16
connector.py
Normal file
@@ -0,0 +1,16 @@
|
||||
import krpc
|
||||
|
||||
|
||||
class Connector:
|
||||
connexion = None
|
||||
|
||||
@classmethod
|
||||
def get_connexion(cls):
|
||||
if cls.connexion is None:
|
||||
cls.connexion = krpc.connect()
|
||||
|
||||
return cls.connexion
|
||||
|
||||
|
||||
def get_connexion():
|
||||
return Connector.get_connexion()
|
||||
25
lib.py
Normal file
25
lib.py
Normal file
@@ -0,0 +1,25 @@
|
||||
from connector import get_connexion
|
||||
|
||||
|
||||
def get_contract(title):
|
||||
for c in get_connexion().space_center.contract_manager.active_contracts:
|
||||
if c.title == title:
|
||||
return c
|
||||
|
||||
raise LookupError('Contract "{}" not found'.format(title))
|
||||
|
||||
|
||||
def get_vessel(name):
|
||||
for v in get_connexion().space_center.vessels:
|
||||
if v.name == name:
|
||||
return v
|
||||
|
||||
raise LookupError("Vessel {} not found".format(name))
|
||||
|
||||
|
||||
def get_rescuee_vessel(rescuee_name):
|
||||
for v in get_connexion().space_center.vessels:
|
||||
if rescuee_name in v.name:
|
||||
return v
|
||||
|
||||
raise LookupError("Rescuee {} vessel not found".format(rescuee_name))
|
||||
168
main.py
168
main.py
@@ -1,16 +1,166 @@
|
||||
# This is a sample Python script.
|
||||
import sys
|
||||
import signal
|
||||
from enum import Enum
|
||||
|
||||
# Press Maj+F10 to execute it or replace it with your code.
|
||||
# Press Double Shift to search everywhere for classes, files, tool windows, actions, and settings.
|
||||
from maneuver_scheduler import ManeuverScheduler
|
||||
|
||||
from maneuvers.rendezvous import RendezvousManeuver
|
||||
from maneuvers.approach import ApproachManeuver
|
||||
from maneuvers.docking import DockingManeuver
|
||||
|
||||
from connector import get_connexion
|
||||
from lib import get_contract, get_vessel, get_rescuee_vessel
|
||||
|
||||
|
||||
def print_hi(name):
|
||||
# Use a breakpoint in the code line below to debug your script.
|
||||
print(f'Hi, {name}') # Press Ctrl+F8 to toggle the breakpoint.
|
||||
class MissionStatus(Enum):
|
||||
Backlog = 1
|
||||
Assigned = 2
|
||||
InProgress = 3
|
||||
Done = 4
|
||||
|
||||
|
||||
class MissionReport:
|
||||
mission_status = None
|
||||
new_vessel_id = None
|
||||
new_missions = []
|
||||
|
||||
def __int__(self):
|
||||
pass
|
||||
|
||||
|
||||
class Mission:
|
||||
pass
|
||||
|
||||
|
||||
class TransportMission(Mission):
|
||||
def __init__(self, kerbal_name, destination):
|
||||
self.kerbal_name = kerbal_name
|
||||
self.destination = destination
|
||||
|
||||
|
||||
class RescueMission(Mission):
|
||||
def __init__(self, contract_title):
|
||||
self.contract_title = contract_title
|
||||
self.contract = get_contract(contract_title)
|
||||
rescuee_first_name = contract_title.split()[1]
|
||||
self.rescuee_vessel = get_rescuee_vessel(rescuee_first_name)
|
||||
self.rescuee_vessel_name = self.rescuee_vessel.name
|
||||
self.rescuee_name = self.rescuee_vessel.crew[0].name
|
||||
|
||||
def execute(self, vessel):
|
||||
conn = get_connexion()
|
||||
sc = conn.space_center
|
||||
|
||||
if sc.active_vessel.name != vessel.name:
|
||||
sc.active_vessel = vessel
|
||||
|
||||
if sc.target_vessel is None or sc.target_vessel.name != self.rescuee_vessel.name:
|
||||
sc.target_vessel = self.rescuee_vessel
|
||||
|
||||
rendezvous_done = RendezvousManeuver(conn, vessel).execute()
|
||||
if rendezvous_done:
|
||||
reference_frame = sc.ReferenceFrame.create_relative(
|
||||
self.rescuee_vessel.reference_frame,
|
||||
rotation=(1., 0., 0., 0.)
|
||||
)
|
||||
approach_done = ApproachManeuver(conn, vessel, reference_frame).start()
|
||||
if approach_done:
|
||||
docking_part = vessel.parts.root.children[0].children[10].children[0].children[0].children[0].children[
|
||||
0]
|
||||
docking_done = DockingManeuver(conn, vessel, docking_part, reference_frame).start()
|
||||
# Find new creat member name
|
||||
# sc.transfer_crew(self.rescuee_name, target_part)
|
||||
# Release Grapple
|
||||
# Destroy Capsule
|
||||
MissionReport.mission_status = MissionStatus.Done
|
||||
MissionReport.new_missions.append(
|
||||
TransportMission(self.rescuee_name, 'kerbin_orbit')
|
||||
)
|
||||
|
||||
|
||||
class ShuttleKerbin:
|
||||
|
||||
def __init__(self, vessel_name):
|
||||
self.vessel_name = vessel_name
|
||||
self.vessel = get_vessel(vessel_name)
|
||||
|
||||
self.current_orbit_missions = []
|
||||
self.current_mission = None
|
||||
|
||||
def plan_mission(self, mission):
|
||||
self.current_mission = mission
|
||||
planning_duration = (5*60)
|
||||
ut = ManeuverScheduler.next_free_timeslot(duration=planning_duration)
|
||||
ManeuverScheduler.book_timeslot(ut, self.vessel)
|
||||
|
||||
def pick_missions(self, backlog):
|
||||
for i, mission in enumerate(backlog.kerbin_orbit_rescue):
|
||||
if self.current_mission is None:
|
||||
self.plan_mission(mission)
|
||||
else:
|
||||
self.current_orbit_missions.append(mission)
|
||||
del(backlog.kerbin_orbit_rescue[i])
|
||||
|
||||
def execute_mission(self, alarm):
|
||||
if self.current_mission is None:
|
||||
if self.current_orbit_missions:
|
||||
self.current_mission = self.current_orbit_missions.pop(0)
|
||||
else:
|
||||
return
|
||||
self.current_mission.execute(self.vessel)
|
||||
|
||||
|
||||
class Backlog:
|
||||
kerbin_orbit_rescue = []
|
||||
kerbin_to_orbit = []
|
||||
kerbin_to_ground = []
|
||||
|
||||
|
||||
def signal_handler(signal, frame):
|
||||
print("\nBye!")
|
||||
sys.exit(0)
|
||||
|
||||
|
||||
# Press the green button in the gutter to run the script.
|
||||
if __name__ == '__main__':
|
||||
print_hi('PyCharm')
|
||||
signal.signal(signal.SIGINT, signal_handler)
|
||||
ships = []
|
||||
|
||||
m = RescueMission('Rescue Rossby from orbit of Kerbin.')
|
||||
Backlog.kerbin_orbit_rescue.append(m)
|
||||
|
||||
ships.append(ShuttleKerbin('KKS Gagarin'))
|
||||
|
||||
conn = get_connexion()
|
||||
# conn.space_center.GameMode
|
||||
space_center = conn.space_center
|
||||
|
||||
for s in ships:
|
||||
s.pick_missions(Backlog)
|
||||
|
||||
while True:
|
||||
for s in ships:
|
||||
s.pick_missions(Backlog)
|
||||
|
||||
space_center.rails_warp_factor = 7
|
||||
with conn.stream(getattr, space_center, 'rails_warp_factor') as rails_warp_factor:
|
||||
with rails_warp_factor.condition:
|
||||
while rails_warp_factor() > 0:
|
||||
rails_warp_factor.wait()
|
||||
|
||||
alarm = ManeuverScheduler.get_last_alarm()
|
||||
vessel = alarm.vessel
|
||||
|
||||
current_ship = None
|
||||
for s in ships:
|
||||
if s.vessel.name == vessel.name:
|
||||
current_ship = s
|
||||
break
|
||||
|
||||
if current_ship is None:
|
||||
raise LookupError("Current ship {} not found".format(vessel.name))
|
||||
current_ship.execute_mission(alarm)
|
||||
alarm.remove()
|
||||
print("turn")
|
||||
|
||||
|
||||
|
||||
# See PyCharm help at https://www.jetbrains.com/help/pycharm/
|
||||
|
||||
@@ -1,3 +1,10 @@
|
||||
from connector import get_connexion
|
||||
import json
|
||||
import math
|
||||
|
||||
from maneuvers import ManeuverAlarmType
|
||||
|
||||
|
||||
class Timeslot:
|
||||
def __init__(self, ut_start, duration):
|
||||
self.ut_start = ut_start
|
||||
@@ -9,26 +16,122 @@ class Timeslot:
|
||||
|
||||
@ut_end.setter
|
||||
def ut_end(self, value):
|
||||
self.duration = self.value - self.start
|
||||
self.duration = value - self.start
|
||||
|
||||
|
||||
class Calendar:
|
||||
def create_reservation(self, ut_start, duration, maneuver):
|
||||
if not self.timeslot_is_free(ut_start, duration):
|
||||
class ManeuverScheduler:
|
||||
|
||||
# alarm_manager = get_connexion().space_center.alarm_manager
|
||||
alarm_manager = get_connexion().kerbal_alarm_clock
|
||||
node_offsets = 60.
|
||||
|
||||
@classmethod
|
||||
def get_last_alarm(cls):
|
||||
return cls.get_ordered_alarms()[0]
|
||||
|
||||
@classmethod
|
||||
def get_ordered_alarms(cls):
|
||||
return sorted(cls.alarm_manager.alarms, key=lambda el: el.time)
|
||||
|
||||
@classmethod
|
||||
def book_timeslot_for_node(cls, vessel, node, maneuver, duration=None):
|
||||
time_required = (node.delta_v * vessel.mass) / vessel.available_thrust
|
||||
if duration is None:
|
||||
duration = math.floor(2 * cls.node_offsets + time_required)
|
||||
|
||||
if not cls.timeslot_is_free(node.ut, duration):
|
||||
raise
|
||||
|
||||
description = {
|
||||
'duration': duration,
|
||||
'vessel_name': vessel.name
|
||||
}
|
||||
|
||||
# arg_dict = {
|
||||
# 'title': "{}' Maneuver: {}".format(vessel.name, maneuver.name),
|
||||
# 'description': json.dumps(description),
|
||||
# 'offset': cls.node_offsets
|
||||
# }
|
||||
|
||||
# cls.alarm_manager.add_maneuver_node_alarm(
|
||||
# vessel,
|
||||
# vessel.control.nodes[0],
|
||||
# **arg_dict)
|
||||
alarm_start = node.ut - (duration / 2 + cls.node_offsets)
|
||||
alarm = cls.alarm_manager.create_alarm(
|
||||
cls.alarm_manager.AlarmType.maneuver,
|
||||
"{}' Maneuver: {}".format(vessel.name, maneuver.name),
|
||||
alarm_start
|
||||
)
|
||||
alarm.vessel = vessel
|
||||
# alarm.margin = cls.node_offsets
|
||||
alarm.notes = json.dumps(description)
|
||||
alarm.action = cls.alarm_manager.AlarmAction.kill_warp_only
|
||||
|
||||
@classmethod
|
||||
def book_timeslot(cls, ut, vessel, duration=None):
|
||||
if duration is None:
|
||||
duration = 5 * 60
|
||||
|
||||
if not cls.timeslot_is_free(ut, duration):
|
||||
raise
|
||||
|
||||
description = {
|
||||
'duration': duration,
|
||||
'vessel_name': vessel.name
|
||||
}
|
||||
|
||||
alarm = cls.alarm_manager.create_alarm(
|
||||
cls.alarm_manager.AlarmType.raw,
|
||||
"{}' Timeslot".format(vessel.name),
|
||||
ut
|
||||
)
|
||||
alarm.vessel = vessel
|
||||
alarm.margin = cls.node_offsets
|
||||
alarm.notes = json.dumps(description)
|
||||
alarm.action = cls.alarm_manager.AlarmAction.kill_warp_only
|
||||
|
||||
|
||||
|
||||
@classmethod
|
||||
def create_reservation(cls, ut_start, duration, maneuver):
|
||||
if not cls.timeslot_is_free(ut_start, duration):
|
||||
raise
|
||||
|
||||
arg_dict = {
|
||||
"title": "{}' Maneuver: {}".format(maneuver.vessel.name, maneuver.name),
|
||||
"description": maneuver.dumps_json()
|
||||
}
|
||||
if maneuver.alarm_type == ManeuverAlarmType.ManeuverNode:
|
||||
cls.alarm_manager.add_maneuver_node_alarm(
|
||||
maneuver.vessel,
|
||||
maneuver.vessel.control.nodes[0],
|
||||
**arg_dict)
|
||||
elif maneuver.alarm_type == ManeuverAlarmType.SOI:
|
||||
cls.alarm_manager.add_soi_alarm(
|
||||
maneuver.vessel,
|
||||
**arg_dict)
|
||||
|
||||
@classmethod
|
||||
def timeslot_is_free(cls, ut_start: int, duration: int) -> bool:
|
||||
return True
|
||||
|
||||
@classmethod
|
||||
def next_free_timeslot(cls, from_ut=None, duration=None) -> int:
|
||||
if from_ut is None:
|
||||
from_ut = get_connexion().space_center.ut
|
||||
if duration is None:
|
||||
duration = 5 * 60
|
||||
|
||||
return from_ut + duration
|
||||
|
||||
@classmethod
|
||||
def get_reservation(cls, ut_at) -> Timeslot:
|
||||
pass
|
||||
|
||||
def timeslot_is_free(self, ut_start: int, duration: int) -> bool:
|
||||
pass
|
||||
|
||||
def next_free_timeslot(self, from_ut, duration=None) -> int:
|
||||
pass
|
||||
|
||||
def get_reservation(self, ut_at) -> Timeslot:
|
||||
pass
|
||||
|
||||
def delete_reservation(self, ut_at, priority):
|
||||
reservation = self.get_re(ut_at)
|
||||
@classmethod
|
||||
def delete_reservation(cls, ut_at, priority):
|
||||
reservation = cls.get_reservation(ut_at)
|
||||
if priority <= reservation.priority:
|
||||
raise
|
||||
|
||||
|
||||
@@ -0,0 +1,50 @@
|
||||
from enum import Enum
|
||||
|
||||
|
||||
class ManeuverAlarmType(Enum):
|
||||
ManeuverNode = 1,
|
||||
SOI = 2
|
||||
|
||||
|
||||
class Maneuver:
|
||||
def __init__(self, conn, vessel):
|
||||
self.vessel = vessel
|
||||
self.conn = conn
|
||||
|
||||
def plan_next_maneuver(self, conn):
|
||||
pass
|
||||
|
||||
|
||||
class NodeManeuver(Maneuver):
|
||||
alarm_type = ManeuverAlarmType.ManeuverNode
|
||||
|
||||
def __init__(self, conn, vessel):
|
||||
super().__init__(conn, vessel)
|
||||
self.mech_jeb = conn.mech_jeb
|
||||
self.node_executor = self.mech_jeb.node_executor
|
||||
|
||||
def execute(self) -> bool:
|
||||
sc = self.conn.space_center
|
||||
if sc.active_vessel.name != self.vessel.name:
|
||||
sc.active_vessel = self.vessel
|
||||
|
||||
while self.vessel.control.nodes:
|
||||
self._execute_node()
|
||||
|
||||
return self.plan_next_maneuver()
|
||||
|
||||
def _execute_node(self):
|
||||
self.node_executor.execute_all_nodes()
|
||||
|
||||
with self.conn.stream(getattr, self.node_executor, "enabled") as enabled:
|
||||
enabled.rate = 1
|
||||
with enabled.condition:
|
||||
while enabled():
|
||||
enabled.wait()
|
||||
|
||||
|
||||
class MechJebManeuver(NodeManeuver):
|
||||
def __init__(self, conn, vessel):
|
||||
super().__init__(conn, vessel)
|
||||
self.maneuver_planner = self.mech_jeb.maneuver_planner
|
||||
|
||||
|
||||
@@ -3,11 +3,53 @@ from krpc.services.spacecenter import SASMode
|
||||
import numpy as np
|
||||
from time import time, sleep
|
||||
|
||||
from utils import magnitude, kill_relative_velocity, correct_course
|
||||
from .utils import magnitude, unitary, kill_relative_velocity, correct_course
|
||||
|
||||
from . import Maneuver
|
||||
|
||||
|
||||
def unitary(vector):
|
||||
return np.array(vector) / magnitude(vector)
|
||||
class ApproachManeuver(Maneuver):
|
||||
|
||||
def __init__(self, conn, vessel_id, reference_frame):
|
||||
super().__init__(conn, vessel_id)
|
||||
self.reference_frame = reference_frame
|
||||
|
||||
def start(self):
|
||||
sc = self.conn.space_center
|
||||
vessel = sc.active_vessel
|
||||
target = sc.target_vessel
|
||||
|
||||
kill_relative_velocity(self.conn, vessel, self.reference_frame)
|
||||
|
||||
self.conn.drawing.add_direction((0, 1, 0), self.reference_frame)
|
||||
|
||||
vessel.control.rcs = False
|
||||
|
||||
pv = vessel.position(self.reference_frame)
|
||||
|
||||
safety_radius = get_safety_radius(vessel) + get_safety_radius(target) + SAFETY_RADIUS_MARGIN
|
||||
|
||||
# if under and inside safety cylinder's circle
|
||||
if pv[1] < safety_radius and pow(pv[0], 2) + pow(pv[2], 2) <= pow(safety_radius, 2):
|
||||
print("We're under the target and inside the safety cylinder, getting out")
|
||||
# get out of the cylinder
|
||||
plane_move_vector = unitary(tuple((pv[0], pv[2]))) * (safety_radius - magnitude(tuple((pv[0], pv[2]))))
|
||||
|
||||
pv = vessel.position(self.reference_frame)
|
||||
move_vector = np.array((plane_move_vector[0], 0, plane_move_vector[1]))
|
||||
move_to_waypoint(self.conn, vessel, pv + move_vector, self.reference_frame)
|
||||
|
||||
print("We're outside of the safety cylinder, setting vertical distance")
|
||||
pv = vessel.position(self.reference_frame)
|
||||
move_to_waypoint(self.conn, vessel, (pv[0], safety_radius, pv[2]), self.reference_frame)
|
||||
|
||||
# should be above and outside => get inside
|
||||
print("We're at the right vertical distance to the target, setting horizontal position")
|
||||
move_to_waypoint(self.conn, vessel, (0, safety_radius, 0), self.reference_frame)
|
||||
|
||||
point_toward_direction(vessel, - np.array(vessel.position(self.reference_frame)), self.reference_frame)
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def get_safety_radius(vessel):
|
||||
@@ -106,8 +148,6 @@ def move_to_waypoint(conn, vessel, waypoint, reference_frame):
|
||||
thrust(vessel, -velocity, reference_frame)
|
||||
print("Ship decelerated")
|
||||
|
||||
#do positition correction
|
||||
|
||||
sa.autopilot_mode = mj.SmartASSAutopilotMode.off
|
||||
sa.update(True)
|
||||
|
||||
@@ -118,6 +158,7 @@ def move_to_waypoint(conn, vessel, waypoint, reference_frame):
|
||||
SAFETY_RADIUS_MARGIN = 10
|
||||
|
||||
|
||||
# DEPRECATED
|
||||
def maneuver_to_approach(conn, reference_frame):
|
||||
print("Handling approach")
|
||||
sc = conn.space_center
|
||||
@@ -158,6 +199,7 @@ def maneuver_to_approach(conn, reference_frame):
|
||||
TARGET_VELOCITY = 2
|
||||
|
||||
|
||||
# DEPRECATED
|
||||
def move_with_vector(conn, vessel, vector, reference_frame):
|
||||
mj = conn.mech_jeb
|
||||
sa = mj.smart_ass
|
||||
|
||||
@@ -1,6 +1,42 @@
|
||||
from time import sleep
|
||||
|
||||
from utils import kill_relative_velocity, correct_course, magnitude
|
||||
from .utils import kill_relative_velocity, correct_course, magnitude
|
||||
|
||||
from . import Maneuver
|
||||
|
||||
|
||||
class DockingManeuver(Maneuver):
|
||||
def __init__(self, conn, vessel_id, docking_part, reference_frame):
|
||||
super().__init__(conn, vessel_id)
|
||||
self.docking_part = docking_part
|
||||
self.reference_frame = reference_frame
|
||||
|
||||
def start(self):
|
||||
vessel = self.conn.space_center.active_vessel
|
||||
self.conn.drawing.add_direction((0, 1, 0), self.reference_frame)
|
||||
self.conn.drawing.add_direction((1, 0, 0), self.reference_frame)
|
||||
vessel.parts.controlling = self.docking_part
|
||||
|
||||
kill_relative_velocity(self.conn, vessel, self.reference_frame)
|
||||
set_attitude_and_roll(self.conn, vessel, self.reference_frame)
|
||||
align_horizontally(self.conn, vessel, self.reference_frame)
|
||||
|
||||
print("Starting docking procedure")
|
||||
vessel.control.set_action_group(0, True)
|
||||
rcs_push(vessel, {"y": 1}, .5)
|
||||
|
||||
vessel.control.rcs = True
|
||||
try:
|
||||
while vessel.position(self.reference_frame)[1] > 0:
|
||||
print(vessel.position(self.reference_frame)[1])
|
||||
correct_course(self.conn, vessel, (0, 0, 0), self.reference_frame)
|
||||
sleep(1)
|
||||
except ValueError as e:
|
||||
vessel = self.conn.space_center.active_vessel
|
||||
finally:
|
||||
vessel.control.rcs = False
|
||||
|
||||
return True
|
||||
|
||||
|
||||
def set_attitude_and_roll(conn, vessel, reference_frame):
|
||||
@@ -127,30 +163,3 @@ def align_horizontally(conn, vessel, reference_frame):
|
||||
|
||||
kill_rcs_velocity(vessel, reference_frame)
|
||||
print("Vertical alignment done!")
|
||||
|
||||
|
||||
def dock_ship(conn, vessel, docking_part, reference_frame):
|
||||
conn.drawing.add_direction((0, 1, 0), reference_frame)
|
||||
conn.drawing.add_direction((1, 0, 0), reference_frame)
|
||||
vessel.parts.controlling = docking_part
|
||||
|
||||
kill_relative_velocity(conn, vessel, reference_frame)
|
||||
|
||||
set_attitude_and_roll(conn, vessel, reference_frame)
|
||||
|
||||
align_horizontally(conn, vessel, reference_frame)
|
||||
|
||||
print("Starting docking procedure")
|
||||
vessel.control.set_action_group(0, True)
|
||||
rcs_push(vessel, {"y": 1}, .5)
|
||||
|
||||
vessel.control.rcs = True
|
||||
try:
|
||||
while vessel.position(reference_frame)[1] > 0:
|
||||
print(vessel.position(reference_frame)[1])
|
||||
correct_course(conn, vessel, (0, 0, 0), reference_frame)
|
||||
sleep(1)
|
||||
except ValueError as e:
|
||||
vessel = conn.space_center.active_vessel
|
||||
finally:
|
||||
vessel.control.rcs = False
|
||||
|
||||
@@ -1,103 +1,87 @@
|
||||
import numpy as np
|
||||
|
||||
from utils import execute_node, magnitude
|
||||
from .utils import magnitude
|
||||
|
||||
from maneuver_scheduler import ManeuverScheduler
|
||||
|
||||
from . import MechJebManeuver
|
||||
|
||||
|
||||
def align_orbit_planes(conn):
|
||||
print("Aligning planes")
|
||||
mj = conn.mech_jeb
|
||||
mp = mj.maneuver_planner
|
||||
oi = mp.operation_inclination
|
||||
oi.time_selector.time_reference = mj.TimeReference.eq_nearest_ad
|
||||
nodes = oi.make_nodes()
|
||||
class RendezvousManeuver(MechJebManeuver):
|
||||
def start(self):
|
||||
self.plan_next_maneuver()
|
||||
|
||||
# kac = conn.kerbal_alarm_clock
|
||||
# kac.create_alarm(
|
||||
# kac.AlarmType.maneuver,
|
||||
# "{}'s Orbital transfer".format(v.name),
|
||||
# nodes[0].ut
|
||||
# )
|
||||
def plan_next_maneuver(self):
|
||||
sc = self.conn.space_center
|
||||
vessel = sc.active_vessel
|
||||
target = sc.target_vessel
|
||||
|
||||
execute_node(conn)
|
||||
if vessel.orbit.distance_at_closest_approach(target.orbit) > 1000:
|
||||
if vessel.orbit.relative_inclination(target.orbit) > 0.0001:
|
||||
AlignOrbitPlaneWithTarget(self.conn, vessel).prepare_maneuver()
|
||||
|
||||
print("Planes aligned!")
|
||||
elif vessel.orbit.distance_at_closest_approach(target.orbit) > 10000:
|
||||
InterceptTargetOrbit(self.conn, vessel).prepare_maneuver()
|
||||
|
||||
else:
|
||||
TuneClosestApproach(self.conn, vessel).prepare_maneuver()
|
||||
return False
|
||||
|
||||
elif vessel.orbit.distance_at_closest_approach(target.orbit) <= 1000 < magnitude(
|
||||
np.array(vessel.position(vessel.reference_frame)) - np.array(target.position(vessel.reference_frame))):
|
||||
MatchVelocityWithTarget(self.conn, vessel).prepare_maneuver()
|
||||
return False
|
||||
else:
|
||||
return True
|
||||
|
||||
|
||||
def intercepting_target_orbit(conn):
|
||||
print("Intercepting target orbit")
|
||||
sc = conn.space_center
|
||||
v = sc.active_vessel
|
||||
mj = conn.mech_jeb
|
||||
mp = mj.maneuver_planner
|
||||
ot = mp.operation_transfer
|
||||
ot.time_selector.time_reference = mj.TimeReference.computed
|
||||
nodes = ot.make_nodes()
|
||||
nodes[0].ut = nodes[0].ut + 0.1
|
||||
class AlignOrbitPlaneWithTarget(RendezvousManeuver):
|
||||
name = "Align orbit plane with target's"
|
||||
|
||||
# kac = conn.kerbal_alarm_clock
|
||||
# kac.create_alarm(
|
||||
# kac.AlarmType.maneuver,
|
||||
# "{}'s Orbital transfer".format(v.name),
|
||||
# nodes[0].ut
|
||||
# )
|
||||
def prepare_maneuver(self):
|
||||
oi = self.maneuver_planner.operation_inclination
|
||||
oi.time_selector.time_reference = self.mech_jeb.TimeReference.eq_nearest_ad
|
||||
nodes = oi.make_nodes()
|
||||
|
||||
execute_node(conn)
|
||||
node = nodes[0]
|
||||
|
||||
print("Target orbit intercepted!")
|
||||
ManeuverScheduler.book_timeslot_for_node(self.vessel, node, self)
|
||||
|
||||
|
||||
def tune_closest_approach(conn):
|
||||
print("Tuning closest approach")
|
||||
sc = conn.space_center
|
||||
v = sc.active_vessel
|
||||
mj = conn.mech_jeb
|
||||
mp = mj.maneuver_planner
|
||||
occ = mp.operation_course_correction
|
||||
nodes = occ.make_nodes()
|
||||
class InterceptTargetOrbit(RendezvousManeuver):
|
||||
name = "Intercept target's orbit"
|
||||
|
||||
# kac = conn.kerbal_alarm_clock
|
||||
# kac.create_alarm(
|
||||
# kac.AlarmType.maneuver,
|
||||
# "{}'s Orbital transfer".format(v.name),
|
||||
# nodes[0].ut
|
||||
# )
|
||||
def prepare_maneuver(self):
|
||||
ot = self.maneuver_planner.operation_transfer
|
||||
ot.time_selector.time_reference = self.mech_jeb.TimeReference.computed
|
||||
nodes = ot.make_nodes()
|
||||
|
||||
execute_node(conn)
|
||||
node = nodes[0]
|
||||
node.ut = node.ut + 1
|
||||
|
||||
print("Closest approach tuned!")
|
||||
ManeuverScheduler.book_timeslot_for_node(self.vessel, node, self)
|
||||
|
||||
|
||||
def match_velocities(conn):
|
||||
print("Matching velocities")
|
||||
sc = conn.space_center
|
||||
v = sc.active_vessel
|
||||
mj = conn.mech_jeb
|
||||
mp = mj.maneuver_planner
|
||||
okrv = mp.operation_kill_rel_vel
|
||||
nodes = okrv.make_nodes()
|
||||
class TuneClosestApproach(RendezvousManeuver):
|
||||
name = "Tune closest approach with target"
|
||||
|
||||
# kac = conn.kerbal_alarm_clock
|
||||
# kac.create_alarm(
|
||||
# kac.AlarmType.maneuver,
|
||||
# "{}'s Orbital transfer".format(v.name),
|
||||
# nodes[0].ut
|
||||
# )
|
||||
def prepare_maneuver(self):
|
||||
occ = self.maneuver_planner.operation_course_correction
|
||||
nodes = occ.make_nodes()
|
||||
|
||||
execute_node(conn)
|
||||
node = nodes[0]
|
||||
|
||||
print("Velocities matched!")
|
||||
ManeuverScheduler.book_timeslot_for_node(self.vessel, node, self)
|
||||
|
||||
|
||||
def maneuver_to_rendezvous(conn, vessel, target):
|
||||
if vessel.orbit.distance_at_closest_approach(target.orbit) > 1000:
|
||||
if vessel.orbit.relative_inclination(target.orbit) > 0.0001:
|
||||
align_orbit_planes(conn)
|
||||
class MatchVelocityWithTarget(RendezvousManeuver):
|
||||
name = "Math velocity with target's"
|
||||
duration = 10 * 60
|
||||
|
||||
if vessel.orbit.distance_at_closest_approach(target.orbit) > 10000:
|
||||
intercepting_target_orbit(conn)
|
||||
def prepare_maneuver(self):
|
||||
okrv = self.maneuver_planner.operation_kill_rel_vel
|
||||
nodes = okrv.make_nodes()
|
||||
|
||||
tune_closest_approach(conn)
|
||||
node = nodes[0]
|
||||
|
||||
if vessel.orbit.distance_at_closest_approach(target.orbit) <= 1000 < magnitude(
|
||||
np.array(vessel.position(vessel.reference_frame)) - np.array(target.position(vessel.reference_frame))):
|
||||
match_velocities(conn)
|
||||
ManeuverScheduler.book_timeslot_for_node(self.vessel, node, self, duration=self.duration)
|
||||
|
||||
@@ -3,21 +3,18 @@ from time import time, sleep
|
||||
import numpy as np
|
||||
|
||||
|
||||
def execute_node(conn):
|
||||
ne = conn.mech_jeb.node_executor
|
||||
ne.execute_all_nodes()
|
||||
|
||||
with conn.stream(getattr, ne, "enabled") as enabled:
|
||||
enabled.rate = 1
|
||||
with enabled.condition:
|
||||
while enabled():
|
||||
enabled.wait()
|
||||
|
||||
|
||||
def magnitude(vector):
|
||||
return np.linalg.norm(vector)
|
||||
|
||||
|
||||
def node_thrust_time(vessel, node):
|
||||
return (node.delta_v * vessel.mass) / vessel.available_thrust
|
||||
|
||||
|
||||
def unitary(vector):
|
||||
return np.array(vector) / magnitude(vector)
|
||||
|
||||
|
||||
THROTTLE = .1
|
||||
|
||||
|
||||
|
||||
49
map.py
Normal file
49
map.py
Normal file
@@ -0,0 +1,49 @@
|
||||
import rustworkx as rx
|
||||
|
||||
graph_dict = {
|
||||
'kerbol': {
|
||||
'moho': {},
|
||||
'eve': {'gilly': {}},
|
||||
'kerbin': {'mun': {}, 'minmus': {}},
|
||||
'duna': {'ike': {}},
|
||||
'dres': {},
|
||||
'jool': {'laythe': {}, 'vall': {}, 'tylo': {}, 'bop': {}, 'pol': {}},
|
||||
'eeloo': {},
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
def create_graph_node(graph_map, name, graph_dict):
|
||||
ground_name = "{}_ground".format(name)
|
||||
ground = graph_map.add_node(ground_name)
|
||||
orbit_name = "{}_orbit".format(name)
|
||||
orbit = graph_map.add_node(orbit_name)
|
||||
|
||||
graph_map.add_edge(orbit, ground, 1)
|
||||
|
||||
for body_name, body_dict in graph_dict.items():
|
||||
child_orbit = create_graph_node(graph_map, body_name, body_dict)
|
||||
graph_map.add_edge(orbit, child_orbit, 1)
|
||||
|
||||
return orbit
|
||||
|
||||
|
||||
class Map:
|
||||
instance = None
|
||||
|
||||
@classmethod
|
||||
def get_map(cls):
|
||||
if cls.instance is None:
|
||||
cls.instance = Map()
|
||||
return cls.instance
|
||||
|
||||
def __init__(self):
|
||||
self.graph = rx.PyGraph()
|
||||
self.index_dict = {}
|
||||
create_graph_node(self.graph, 'kerbol', graph_dict['kerbol'])
|
||||
self.reverse_index = self.graph.nodes()
|
||||
self.index_dict = {value: key for key, value in enumerate(self.reverse_index)}
|
||||
|
||||
def get_shortest_path(self, body_a, body_b):
|
||||
result = rx.dijkstra_shortest_paths(self.graph, self.index_dict[body_a], self.index_dict[body_b])
|
||||
return [self.reverse_index[body_index] for body_index in result[self.index_dict[body_b]]]
|
||||
Reference in New Issue
Block a user