jacob.eva/openCom-Companion
1
0
Fork 0
mirror of https://github.com/liberatedsystems/openCom-Companion.git synced 2024-10-26 02:50:37 +02:00
Code Issues Actions Packages Projects Releases Wiki Activity

Added telemetry plugin functionality

Browse Source
This commit is contained in:
Mark Qvist 2024-03-26 00:29:49 +01:00
parent d5630d72de
commit c93fe2ce0d
5 changed files with 371 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

34
docs/example_plugins/telemetry.py Normal file
View File

@ -0,0 +1,34 @@
import RNS
class BasicTelemetryPlugin(SidebandTelemetryPlugin):
plugin_name = "telemetry_example"
def start(self):
# Do any initialisation work here
RNS.log("Basic telemetry plugin example starting...")
# And finally call start on superclass
super().start()
def stop(self):
# Do any teardown work here
pass
# And finally call stop on superclass
super().stop()
def update_telemetry(self, telemeter):
if telemeter != None:
RNS.log("Updating power sensors")
telemeter.synthesize("power_consumption")
telemeter.sensors["power_consumption"].update_consumer(2163.15, type_label="Heater consumption")
telemeter.sensors["power_consumption"].update_consumer(12.7/1e6, type_label="Receiver consumption")
telemeter.sensors["power_consumption"].update_consumer(0.055, type_label="LED consumption")
telemeter.sensors["power_consumption"].update_consumer(982.22*1e9, type_label="Smelter consumption")
telemeter.synthesize("power_production")
telemeter.sensors["power_production"].update_producer(5732.15, type_label="Solar production")
# Finally, tell Sideband what class in this
# file is the actual plugin class.
plugin_class = BasicTelemetryPlugin

32
sbapp/sideband/core.py
View File

@ -17,7 +17,7 @@ import multiprocessing.connection
from threading import Lock
from .res import sideband_fb_data
from .sense import Telemeter, Commands
from .plugins import SidebandCommandPlugin, SidebandServicePlugin
from .plugins import SidebandCommandPlugin, SidebandServicePlugin, SidebandTelemetryPlugin
if RNS.vendor.platformutils.get_platform() == "android":
from jnius import autoclass, cast
@ -247,6 +247,7 @@ class SidebandCore():
self.active_command_plugins = {}
self.active_service_plugins = {}
self.active_telemetry_plugins = {}
if self.is_service or self.is_standalone:
self.__load_plugins()
@ -684,6 +685,7 @@ class SidebandCore():
plugin_globals = {}
plugin_globals["SidebandServicePlugin"] = SidebandServicePlugin
plugin_globals["SidebandCommandPlugin"] = SidebandCommandPlugin
plugin_globals["SidebandTelemetryPlugin"] = SidebandTelemetryPlugin
RNS.log("Loading plugin \""+str(file)+"\"", RNS.LOG_NOTICE)
plugin_path = os.path.join(plugins_path, file)
exec(open(plugin_path).read(), plugin_globals)
@ -715,8 +717,27 @@ class SidebandCore():
pass
del plugin
elif issubclass(type(plugin), SidebandTelemetryPlugin):
plugin_name = plugin.plugin_name
if not plugin_name in self.active_telemetry_plugins:
self.active_telemetry_plugins[plugin_name] = plugin
RNS.log("Registered "+str(plugin)+" as telemetry plugin \""+str(plugin_name)+"\"", RNS.LOG_NOTICE)
else:
RNS.log("Could not register "+str(plugin)+" as telemetry plugin \""+str(plugin_name)+"\". Telemetry type was already registered", RNS.LOG_ERROR)
try:
plugin.stop()
except Exception as e:
pass
del plugin
else:
RNS.log("Unknown plugin type was loaded, ignoring it.", RNS.LOG_ERROR)
try:
plugin.stop()
except Exception as e:
pass
del plugin
else:
RNS.log("Plugin "+str(plugin)+" failed to start, ignoring it.", RNS.LOG_ERROR)
del plugin
@ -2421,6 +2442,15 @@ class SidebandCore():
else:
self.telemeter.disable(sensor)
for telemetry_plugin in self.active_telemetry_plugins:
try:
plugin = self.active_telemetry_plugins[telemetry_plugin]
plugin.update_telemetry(self.telemeter)
except Exception as e:
RNS.log("An error occurred while "+str(telemetry_plugin)+" was handling telemetry. The contained exception was: "+str(e), RNS.LOG_ERROR)
RNS.trace_exception(e)
if self.config["telemetry_s_fixed_location"]:
self.telemeter.synthesize("location")
self.telemeter.sensors["location"].latitude = self.config["telemetry_s_fixed_latlon"][0]

21
sbapp/sideband/plugins.py
View File

@ -39,3 +39,24 @@ class SidebandServicePlugin(SidebandPlugin):
def get_sideband(self):
return self.__sideband
class SidebandTelemetryPlugin(SidebandPlugin):
def __init__(self, sideband_core):
self.__sideband = sideband_core
self.__started = False
self.plugin_name = type(self).plugin_name
def start(self):
self.__started = True
def stop(self):
self.__started = False
def is_running(self):
return self.__started == True
def get_sideband(self):
return self.__sideband
def update_telemetry(self, telemeter):
raise NotImplementedError

257
sbapp/sideband/sense.py
View File

@ -43,38 +43,28 @@ class Telemeter():
def __init__(self, from_packed=False, android_context=None, service=False, location_provider=None):
self.sids = {
Sensor.SID_TIME: Time,
Sensor.SID_RECEIVED: Received,
Sensor.SID_INFORMATION: Information,
Sensor.SID_BATTERY: Battery,
Sensor.SID_PRESSURE: Pressure,
Sensor.SID_LOCATION: Location,
Sensor.SID_PHYSICAL_LINK: PhysicalLink,
Sensor.SID_TEMPERATURE: Temperature,
Sensor.SID_HUMIDITY: Humidity,
Sensor.SID_MAGNETIC_FIELD: MagneticField,
Sensor.SID_AMBIENT_LIGHT: AmbientLight,
Sensor.SID_GRAVITY: Gravity,
Sensor.SID_ANGULAR_VELOCITY: AngularVelocity,
Sensor.SID_ACCELERATION: Acceleration,
Sensor.SID_PROXIMITY: Proximity,
Sensor.SID_TIME: Time, Sensor.SID_RECEIVED: Received,
Sensor.SID_INFORMATION: Information, Sensor.SID_BATTERY: Battery,
Sensor.SID_PRESSURE: Pressure, Sensor.SID_LOCATION: Location,
Sensor.SID_PHYSICAL_LINK: PhysicalLink, Sensor.SID_TEMPERATURE: Temperature,
Sensor.SID_HUMIDITY: Humidity, Sensor.SID_MAGNETIC_FIELD: MagneticField,
Sensor.SID_AMBIENT_LIGHT: AmbientLight, Sensor.SID_GRAVITY: Gravity,
Sensor.SID_ANGULAR_VELOCITY: AngularVelocity, Sensor.SID_ACCELERATION: Acceleration,
Sensor.SID_PROXIMITY: Proximity, Sensor.SID_POWER_CONSUMPTION: PowerConsumption,
Sensor.SID_POWER_PRODUCTION: PowerProduction, Sensor.SID_PROCESSOR: Processor,
Sensor.SID_RAM: RandomAccessMemory, Sensor.SID_NVM: NonVolatileMemory,
}
self.available = {
"time": Sensor.SID_TIME,
"information": Sensor.SID_INFORMATION,
"received": Sensor.SID_RECEIVED,
"battery": Sensor.SID_BATTERY,
"pressure": Sensor.SID_PRESSURE,
"location": Sensor.SID_LOCATION,
"physical_link": Sensor.SID_PHYSICAL_LINK,
"temperature": Sensor.SID_TEMPERATURE,
"humidity": Sensor.SID_HUMIDITY,
"magnetic_field": Sensor.SID_MAGNETIC_FIELD,
"ambient_light": Sensor.SID_AMBIENT_LIGHT,
"gravity": Sensor.SID_GRAVITY,
"angular_velocity": Sensor.SID_ANGULAR_VELOCITY,
"acceleration": Sensor.SID_ACCELERATION,
"proximity": Sensor.SID_PROXIMITY,
"information": Sensor.SID_INFORMATION, "received": Sensor.SID_RECEIVED,
"battery": Sensor.SID_BATTERY, "pressure": Sensor.SID_PRESSURE,
"location": Sensor.SID_LOCATION, "physical_link": Sensor.SID_PHYSICAL_LINK,
"temperature": Sensor.SID_TEMPERATURE, "humidity": Sensor.SID_HUMIDITY,
"magnetic_field": Sensor.SID_MAGNETIC_FIELD, "ambient_light": Sensor.SID_AMBIENT_LIGHT,
"gravity": Sensor.SID_GRAVITY, "angular_velocity": Sensor.SID_ANGULAR_VELOCITY,
"acceleration": Sensor.SID_ACCELERATION, "proximity": Sensor.SID_PROXIMITY,
"power_consumption": Sensor.SID_POWER_CONSUMPTION, "power_production": Sensor.SID_POWER_PRODUCTION,
"processor": Sensor.SID_PROCESSOR, "ram": Sensor.SID_RAM, "nvm": Sensor.SID_NVM,
}
self.from_packed = from_packed
self.sensors = {}
@ -180,22 +170,27 @@ class Telemeter():
class Sensor():
SID_NONE = 0x00
SID_TIME = 0x01
SID_LOCATION = 0x02
SID_PRESSURE = 0x03
SID_BATTERY = 0x04
SID_PHYSICAL_LINK = 0x05
SID_ACCELERATION = 0x06
SID_TEMPERATURE = 0x07
SID_HUMIDITY = 0x08
SID_MAGNETIC_FIELD = 0x09
SID_AMBIENT_LIGHT = 0x0A
SID_GRAVITY = 0x0B
SID_ANGULAR_VELOCITY = 0x0C
SID_PROXIMITY = 0x0E
SID_INFORMATION = 0x0F
SID_RECEIVED = 0x10
SID_NONE = 0x00
SID_TIME = 0x01
SID_LOCATION = 0x02
SID_PRESSURE = 0x03
SID_BATTERY = 0x04
SID_PHYSICAL_LINK = 0x05
SID_ACCELERATION = 0x06
SID_TEMPERATURE = 0x07
SID_HUMIDITY = 0x08
SID_MAGNETIC_FIELD = 0x09
SID_AMBIENT_LIGHT = 0x0A
SID_GRAVITY = 0x0B
SID_ANGULAR_VELOCITY = 0x0C
SID_PROXIMITY = 0x0E
SID_INFORMATION = 0x0F
SID_RECEIVED = 0x10
SID_POWER_CONSUMPTION = 0x11
SID_POWER_PRODUCTION = 0x12
SID_PROCESSOR = 0x13
SID_RAM = 0x14
SID_NVM = 0x15
def __init__(self, sid = None, stale_time = None):
self._telemeter = None
@ -1312,3 +1307,175 @@ class Proximity(Sensor):
return packed
except:
return None
class PowerConsumption(Sensor):
SID = Sensor.SID_POWER_CONSUMPTION
STALE_TIME = 5
def __init__(self):
super().__init__(type(self).SID, type(self).STALE_TIME)
def setup_sensor(self):
self.update_data()
def teardown_sensor(self):
self.data = None
def update_consumer(self, power, type_label=None):
if type_label == None:
type_label = 0x00
elif type(type_label) != str:
return False
if self.data == None:
self.data = {}
self.data[type_label] = power
return True
def remove_consumer(self, type_label=None):
if type_label == None:
type_label = 0x00
if type_label in self.data:
self.data.pop(type_label)
return True
return False
def update_data(self):
pass
def pack(self):
d = self.data
if d == None:
return None
else:
packed = []
for type_label in self.data:
packed.append([type_label, self.data[type_label]])
return packed
def unpack(self, packed):
try:
if packed == None:
return None
else:
unpacked = {}
for entry in packed:
unpacked[entry[0]] = entry[1]
return unpacked
except:
return None
def render(self, relative_to=None):
if self.data == None:
return None
consumers = []
for type_label in self.data:
if type_label == 0x00:
label = "Power consumption"
else:
label = type_label
consumers.append({"label": label, "w": self.data[type_label]})
rendered = {
"icon": "power-plug-outline",
"name": "Power Consumption",
"values": consumers,
}
return rendered
class PowerProduction(Sensor):
SID = Sensor.SID_POWER_PRODUCTION
STALE_TIME = 5
def __init__(self):
super().__init__(type(self).SID, type(self).STALE_TIME)
def setup_sensor(self):
self.update_data()
def teardown_sensor(self):
self.data = None
def update_producer(self, power, type_label=None):
if type_label == None:
type_label = 0x00
elif type(type_label) != str:
return False
if self.data == None:
self.data = {}
self.data[type_label] = power
return True
def remove_producer(self, type_label=None):
if type_label == None:
type_label = 0x00
if type_label in self.data:
self.data.pop(type_label)
return True
return False
def update_data(self):
pass
def pack(self):
d = self.data
if d == None:
return None
else:
packed = []
for type_label in self.data:
packed.append([type_label, self.data[type_label]])
return packed
def unpack(self, packed):
try:
if packed == None:
return None
else:
unpacked = {}
for entry in packed:
unpacked[entry[0]] = entry[1]
return unpacked
except:
return None
def render(self, relative_to=None):
if self.data == None:
return None
producers = []
for type_label in self.data:
if type_label == 0x00:
label = "Power Production"
else:
label = type_label
producers.append({"label": label, "w": self.data[type_label]})
rendered = {
"icon": "lightning-bolt",
"name": "Power Production",
"values": producers,
}
return rendered
# TODO: Implement
class Processor(Sensor):
pass
class RandomAccessMemory(Sensor):
pass
class NonVolatileMemory(Sensor):
pass

72
sbapp/ui/objectdetails.py
View File

@ -415,6 +415,78 @@ class RVDetails(MDRecycleView):
if q != None or rssi != None: snr_str = ", "+snr_str
if q_str or rssi_str or snr_str:
formatted_values = q_str+rssi_str+snr_str
elif name == "Power Consumption":
cs = s["values"]
if cs != None:
for c in cs:
label = c["label"]
watts = c["w"]
prefix = ""
if watts < 1/1e6:
watts *= 1e9
prefix = "n"
elif watts < 1/1e3:
watts *= 1e6
prefix = "µ"
elif watts < 1:
watts *= 1e3
prefix = "m"
elif watts >= 1e15:
watts /= 1e15
prefix = "E"
elif watts >= 1e12:
watts /= 1e12
prefix = "T"
elif watts >= 1e9:
watts /= 1e9
prefix = "G"
elif watts >= 1e6:
watts /= 1e6
prefix = "M"
elif watts >= 1e3:
watts /= 1e3
prefix = "K"
watts = round(watts, 2)
p_text = f"{label} [b]{watts} {prefix}W[/b]"
extra_entries.append({"icon": s["icon"], "text": p_text})
elif name == "Power Production":
cs = s["values"]
if cs != None:
for c in cs:
label = c["label"]
watts = c["w"]
prefix = ""
if watts < 1/1e6:
watts *= 1e9
prefix = "n"
elif watts < 1/1e3:
watts *= 1e6
prefix = "µ"
elif watts < 1:
watts *= 1e3
prefix = "m"
elif watts >= 1e15:
watts /= 1e15
prefix = "E"
elif watts >= 1e12:
watts /= 1e12
prefix = "T"
elif watts >= 1e9:
watts /= 1e9
prefix = "G"
elif watts >= 1e6:
watts /= 1e6
prefix = "M"
elif watts >= 1e3:
watts /= 1e3
prefix = "K"
watts = round(watts, 2)
p_text = f"{label} [b]{watts} {prefix}W[/b]"
extra_entries.append({"icon": s["icon"], "text": p_text})
elif name == "Location":
lat = s["values"]["latitude"]
lon = s["values"]["longitude"]