Added sensor a sensor module to work out GET/SET/EXE/NOT api
This commit is contained in:
Laika
parent
a9a7e1eead
commit
652e122d30
23
satsim/module.py
Normal file
23
satsim/module.py
Normal file
|
|
@ -0,0 +1,23 @@
|
|||
|
||||
class module:
|
||||
id_c = 0
|
||||
def __init__(self, name):
|
||||
self.fields = {}
|
||||
self.fields["name"] = "PRIMITIVE"
|
||||
self.id = module.id_c
|
||||
module.id_c += 1
|
||||
|
||||
def mod_get(self):
|
||||
print("[!] GET CALLED ON PRIMITIVE MODULE")
|
||||
|
||||
def mod_set(self):
|
||||
print("[!] SET CALLED ON PRIMITIVE MODULE")
|
||||
|
||||
def mod_exe(self):
|
||||
print("[!] EXE CALLED ON PRIMITIVE MODULE")
|
||||
|
||||
def mod_not(self):
|
||||
print("[!] NOT CALLED ON PRIMITIVE MODULE")
|
||||
|
||||
def mod_update(self):
|
||||
pass
|
||||
137
satsim/satsim.py
137
satsim/satsim.py
|
|
@ -1,135 +1,66 @@
|
|||
import numpy as np
|
||||
import threading
|
||||
from sensor_module import gamma_sensor
|
||||
from time import sleep
|
||||
from astropy.coordinates import SkyCoord as sc
|
||||
from os import system
|
||||
|
||||
#debugg to rebuild the file every test
|
||||
system('cp command_q test_q')
|
||||
|
||||
ONSAT_FILESTORE = "DATA"
|
||||
DOWNLINK_FILESTORE = "../store/data"
|
||||
|
||||
class Sat:
|
||||
class sat:
|
||||
def __init__(self):
|
||||
self.location = self.get_loc()
|
||||
self.attitude = sc(0,0, unit='deg', frame='icrs')
|
||||
self.battery = 10000
|
||||
self.battery = 100.00
|
||||
self.MTE = 0
|
||||
self.kill = False
|
||||
self.system_modes = {
|
||||
"opt_read": False,
|
||||
"opt_cool": False,
|
||||
"rad_mode": False,
|
||||
"grv_mode": False,
|
||||
"HGA_mode": "off", #off/rx/tx
|
||||
"LGA_mode": "idle", #idle/dup/tx, for power these modes will be automatic
|
||||
"torquers": False,
|
||||
"gyro_del": False,
|
||||
}
|
||||
self.maneuver_q = []
|
||||
self.maneuver_TTC = 0 #time to completion
|
||||
self.gyro_saturation = 0
|
||||
|
||||
self.lens_temp =
|
||||
self.modules = [gamma_sensor(),gamma_sensor()]
|
||||
|
||||
def get_loc(self):
|
||||
"""
|
||||
retrieve location from orbit sim
|
||||
"""
|
||||
pass
|
||||
return (10,10,10)
|
||||
|
||||
def get_status(self, command="null"):
|
||||
print("BoSLOO STATUS:")
|
||||
print("Feelin Fine <3")
|
||||
|
||||
def inst_optical(self, command="null"):
|
||||
"""
|
||||
Main Telescope
|
||||
"""
|
||||
pass
|
||||
|
||||
def inst_ultraviolet(self, command="null"):
|
||||
"""
|
||||
UV band Telescope
|
||||
"""
|
||||
pass
|
||||
|
||||
def inst_hydrogen_line(self, command="null"):
|
||||
"""
|
||||
hydrogen line radio telescope
|
||||
"""
|
||||
|
||||
|
||||
def inst_radiation(self, command="null"):
|
||||
"""
|
||||
beta/gamma detector
|
||||
"""
|
||||
pass
|
||||
|
||||
def inst_gravitometer(self, command="null"):
|
||||
"""
|
||||
local gravity and potential gravity waves (rare)
|
||||
"""
|
||||
pass
|
||||
|
||||
def inst_hg_antenna(self, command="null"):
|
||||
"""
|
||||
tight beam antenna for high speed up/downlink
|
||||
has to point at your ground station.
|
||||
"""
|
||||
pass
|
||||
status_string = ""
|
||||
|
||||
def check_command_q(self):
|
||||
with open('test_q', 'r+') as q:
|
||||
commands = q.read().splitlines()
|
||||
remainder = ""
|
||||
for com in commands:
|
||||
cur_batch = []
|
||||
with open("test_q","r") as q:
|
||||
com_q = q.read().splitlines()
|
||||
for com in com_q:
|
||||
if int(com.split(":")[0]) == self.MTE:
|
||||
print("got com -> ", com)
|
||||
cur_batch.append(com)
|
||||
return cur_batch
|
||||
|
||||
com_type = com.split(":")[1]
|
||||
if com_type.strip() == "status":
|
||||
self.get_status(com)
|
||||
elif com_type.strip() == "point":
|
||||
self.set_target(com)
|
||||
elif com_type.strip() == "uload":
|
||||
self.upload(com)
|
||||
elif com_type.strip() == "dload":
|
||||
self.download()
|
||||
elif com_type.strip() == "list":
|
||||
self.list_drive()
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
elif com_type.strip() == "status":
|
||||
|
||||
else:
|
||||
remainder += com + '\n'
|
||||
q.truncate(0)
|
||||
q.seek(0)
|
||||
q.write(remainder)
|
||||
|
||||
def update(self):
|
||||
self.check_command_q()
|
||||
if not self.kill:
|
||||
threading.Timer(1, self.update).start()
|
||||
else:
|
||||
print("heading to bed")
|
||||
self.MTE += 1
|
||||
def spin(self):
|
||||
while True:
|
||||
for m in self.modules:
|
||||
m.mod_update()
|
||||
print(m.mod_get("detections"))
|
||||
if self.MTE % 10 == 0:
|
||||
m.mod_set("flush", 1)
|
||||
batch = self.check_command_q()
|
||||
print("MTE:", self.MTE)
|
||||
print("COMS:", batch)
|
||||
if self.MTE == 5:
|
||||
self.kill = False
|
||||
if self.kill:
|
||||
print("heading to bed")
|
||||
return
|
||||
self.MTE += 1
|
||||
sleep(1)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
BoSLOO = Sat()
|
||||
BoSLOO.update()
|
||||
#while True:
|
||||
# try:
|
||||
# pass
|
||||
# except KeyboardInterrupt:
|
||||
# BoSLOO.kill = True
|
||||
# print("Good night!")
|
||||
# break
|
||||
BoSLOO = sat()
|
||||
BoSLOO.spin()
|
||||
|
||||
|
|
|
|||
45
satsim/sensor_module.py
Normal file
45
satsim/sensor_module.py
Normal file
|
|
@ -0,0 +1,45 @@
|
|||
from module import module
|
||||
from random import randint
|
||||
class gamma_sensor(module):
|
||||
def __init__(self):
|
||||
super().__init__("gamma spectrometer")
|
||||
self.fields["name"] = "gammaSensor"
|
||||
self.fields["detections"] = 0
|
||||
self.fields["detections_str"] = []
|
||||
self.fields["threshold"] = 90
|
||||
self.fields["flush"] = 0
|
||||
self.writable = ["threshold","flush"]
|
||||
print("id is:", self.id)
|
||||
|
||||
def mod_get(self, field="none"):
|
||||
if field not in self.fields:
|
||||
return (-1, "GET FATAL: field '" + field + "'' does not exist in module: " + self.fields["name"])
|
||||
return (self.fields[field], "GET OK")
|
||||
|
||||
def mod_set(self, field="none", value="none"):
|
||||
if field not in self.fields:
|
||||
return (-1, "SET FATAL: field '" + field + "'' does not exist in module: " + self.fields["name"])
|
||||
if field not in self.writable:
|
||||
return (-1, "SET FATAL: field '" + field + "' is not writable in module: " + self.fields["name"])
|
||||
try:
|
||||
self.fields[field] = int(value)
|
||||
return (0, "SET OK")
|
||||
except:
|
||||
return (-1, "SET FATAL: field '" + field + "' in module: " + self.fields["name"] + " takes int")
|
||||
|
||||
def mod_exe(self):
|
||||
pass
|
||||
|
||||
def mod_not(self):
|
||||
pass
|
||||
|
||||
def mod_update(self):
|
||||
if self.fields["flush"] != 0:
|
||||
self.fields["detections"] = 0
|
||||
self.fields["detections_str"] = []
|
||||
self.fields["flush"] = 0
|
||||
for i in range(0,randint(4,100)):
|
||||
strength = randint(0,100)
|
||||
if strength > self.fields["threshold"]:
|
||||
self.fields["detections"] += 1
|
||||
self.fields["detections_str"].append(strength)
|
||||
Loading…
Reference in a new issue