Integrating an RF modem interface with fprime-gds #2639
Replies: 2 comments 27 replies
-
|
Is this interfacing with the modem on the FSW or the GDS side of things? |
Beta Was this translation helpful? Give feedback.
-
|
Yeah, plugin support for your modem communication adapter would be the ideal way to go so that you don't have to modify the GDS code itself. Communication adapters are supported by the plugin architecture, but it looks like there aren't any communication plugin examples yet. However, the built-in IP and UART adapters are packaged as plugins and could be a good reference If you haven't found it yet, there's a plugin how-to guide here I suspect you could probably even use the ThreadedUDPServer class as part of your plugin if that's helpful |
Beta Was this translation helpful? Give feedback.
-
|
@Joshua-Anderson is spot-on. I definitely recommend using plugins. A few helpful tips as plugins are new functionality:
|
Beta Was this translation helpful? Give feedback.
-
|
Oh, and any feedback is appreciated! |
Beta Was this translation helpful? Give feedback.
-
|
looks great ! I'll update here as I'm sure I'll have questions as I get going (If that seems appropriate). |
Beta Was this translation helpful? Give feedback.
-
|
So I finally got back to this and I think I'm missing some info about how I can package this with fprime-gds (I see some mention in the docs but it's a bit unclear to me) I have a modemadapter like this: #!/usr/bin/env python3
import abc
import socket
from typing import NamedTuple, Tuple, Type
import threading
from fprime_gds.common.communication.framing import FramerDeframer
from fprime_gds.plugin.definitions import gds_plugin_implementation
from fprime_gds.common.communication.adapters import base as fprime_comms_base
from fprime_gds.common.communication.adapters.ip import IpAdapter
from modem_interface import ModemInterface
from dataclasses import dataclass, field
from datetime import datetime
from time import time
@dataclass
class ModemAdapterCfg:
"""
A class for the emulator's status.
"""
modem_ip_address: Tuple[str, int]
fsw_ip_address: Tuple[str, int]
poll_hz: int = 10
class ModemAdapter(fprime_comms_base):
'''This provides an fprime-gds adapter plugin such that downlink is facilitated via rf
(through the use of the modem interface) and uplink is meant to be done using the usual tcp interface.
'''
def __init__(self, cfg: ModemAdapterCfg):
self.cfg = cfg
self.poll = threading.Event()
self.shutdown = threading.Event()
self.modem = ModemInterface(cfg.modem_ip_address[0], cfg.modem_ip_address[1])
self.ip = IpAdapter(cfg.fsw_ip_address[0], cfg.fsw_ip_address[1])
def __poll_for_datagrams(self):
while not self.shutdown.is_set()
if self.poll.is_set():
self.modem.decodeIncomingData()
time.sleep(1 / self.cfg.poll_hz)
def __del__(self):
self.close()
self.shutdown.set()
def open(self):
self.poll.set()
def close(self):
self.poll.clear()
@abc.abstractmethod
def read(self, timeout=0.500):
return self.modem.popDatagram()
@abc.abstractmethod
def write(self, frame):
"""
Write to the interface. Must be overridden by the child adapter. Throw no fatal errors, reconnect instead.
:param frame: framed data to uplink
:return: True if data sent through adapter, False otherwise
"""
self.ip.write(frame)
@classmethod
@gds_plugin_specification
def register_communication_plugin(cls) -> Type["ModemAdapter"]:
return clsWhere I want to use rf for the downlink and ip for the uplink (ideally I'd actually be able to get downlink via ip simulataneously for comparison but that for a later step) Currently I'm attempting to test the tlm with just Ip with the following: class TelemLink(object):
"""
Class for managing telemetry link.
"""
def __init__(self, cfg: TelemLinkCfg, logger: Logger) -> None:
"""
Initializes a TelemLink object with the provided configuration
and logger objects.
"""
self.fprime_api, self.fprime_pipeline = self._establish_connection(cfg, logger)
self.cfg = cfg
self.logger = logger
def __del__(self):
self.fprime_api.teardown()
self.fprime_pipeline.disconnect()
@staticmethod
def _establish_connection(cfg: TelemLinkCfg, logger: Logger) -> Tuple[IntegrationTestAPI, StandardPipeline]:
"""
Establishes connection with the api interface by setting up
the pipeline using the provided configuration and logger.
"""
# Instantiate a pipeline and config manager for managing the api interface
pipeline = StandardPipeline()
config_manager = ConfigManager()
# Launch tcp server
run_deployment.launch_tts(50000, "0.0.0.0", cfg.tmp_path)
# Setup pipeline
pipeline.setup(config_manager, cfg.dict_path, cfg.tmp_path)
api_connect_str = f"tcp://127.0.0.1:50000"
api = IntegrationTestAPI(pipeline, cfg.log_path)
logger.info(f"Attempting to connect to {api_connect_str}")
pipeline.connect(api_connect_str, RoutingTag.FSW)
return api, pipeline
def get_data_mock_stats(self) -> DataMockStats:
"""
Acquires telemetry from the link, logs the result and asserts
conditions based on the received telemetry.
Displays telemetry based on the preconfigured export type
"""
result = self.fprime_api.await_telemetry("DataMockStatsTlm")
self.logger.info(result)
if result is None:
return None
else:
return DataMockStats.from_dict(result.get_val())
def set_flight_downlink_profiling_enabled(self, is_enabled: bool):
"""
Sends the `SetIsFlightDownlinkProfilingEnabled` command to the Profiler component.
"""
result = self.fprime_api.send_command("profiler.SetIsFlightDownlinkProfilingEnabled", [is_enabled])
self.logger.info(result)
return resultHow can I pass through the modem config at this level and "select" it as the adapter of choice ? |
Beta Was this translation helpful? Give feedback.
-
|
I've expanded on my initial implementation a bit but I'll leave the reply comment up for posterity. #!/usr/bin/env python3
import abc
import socket
from typing import NamedTuple, Tuple, Type
import threading
from fprime_gds.common.communication.framing import FramerDeframer
from fprime_gds.common.communication.ground import TCPGround
from fprime_gds.common.communication.updown import Downlinker, Uplinker
from fprime_gds.plugin.definitions import gds_plugin_implementation
from fprime_gds.common.communication.adapters.base import BaseAdapter
from fprime_gds.common.communication.adapters.ip import IpAdapter
from fprime_gds.plugin.system import FpFramerDeframer
from modem_interface import ModemInterface
from dataclasses import dataclass, field
from datetime import datetime
from time import time
import sys
@dataclass
class ModemAdapterCfg:
modem_ip_address: Tuple[str, int]
fsw_ip_address: Tuple[str, int]
poll_hz: int = 10
poll_timeout_s = 5000
@dataclass
class ModemStatus:
bytes_decoded: int = 0
start: datetime = datetime.now()
isPolling: bool = False
def __str__(self):
"""
Returns a formatted string representation of the status.
"""
time_since_start = 0
if self.isPolling:
time_since_start = (datetime.now() - self.start).total_seconds()
return (
f"Status:\n"
f" Time since start (S): {time_since_start} ms\n"
f" Bytes Decoded: {self.bytes_decoded}\n"
)
class ModemAdapter(BaseAdapter):
'''This provides an fprime-gds adapter plugin such that downlink is facilitated via rf
(through the use of the modem interface) and uplink is meant to be done using the usual tcp interface.
'''
def __init__(self, cfg: ModemAdapterCfg):
self.cfg = cfg
self.poll = threading.Event()
self.shutdown = threading.Event()
self.modem = ModemInterface(cfg.modem_ip_address[0], cfg.modem_ip_address[1])
self.ip = IpAdapter(cfg.fsw_ip_address[0], cfg.fsw_ip_address[1])
self.poll_thread = threading.Thread(target=self._poll_for_datagrams, daemon=True)
self._status = ModemStatus()
@classmethod
def get_name(cls) -> str:
return "ModemAdapter"
@classmethod
def get_arguments(cls):
""" Arguments to request from the CLI """
return {}
@classmethod
def check_arguments(cls):
""" Check arguments from the CLI """
pass
@property
def status(self):
"""
Retrieves and logs the status of the modem adapter.
"""
return self._status
def _poll_for_datagrams(self):
while not self.shutdown.is_set():
# Blocks thread on max timeout
try:
self.poll.wait(self.cfg.poll_timeout_s)
self.modem.decodeIncomingData()
time.sleep(1 / self.cfg.poll_hz)
except RuntimeError:
pass
def __del__(self):
self.close()
self.shutdown.set()
def open(self):
self._status.start = datetime.now()
self.poll.set()
def close(self):
self.poll.clear()
def read(self, timeout=0.500):
"""
Function that checks for available datagrams and pops the first one
when it becomes available. If no datagram becomes available within
'timeout' seconds, a RuntimeError is raised indicating the timeout.
Parameters:
timeout (float): Maximum time in seconds to wait for a datagram.
Default is 0.500 seconds.
Raises:
RuntimeError: When the function times out without a datagram being
available.
Returns:
datagram: The first datagram that becomes available.
"""
start_time = time.time() # recording the start time
datagram = None
while time.time() - start_time < timeout:
if self.modem.datagramsAvailable():
datagram = self.modem.popDatagram()
return datagram
# Sleep for a small duration to avoid busy-waiting
time.sleep(0.01)
end_time = time.time() # recording the end time
raise RuntimeError(f"Modem timed out after {end_time - start_time} seconds!")
def write(self, frame):
self.ip.write(frame)
@classmethod
@gds_plugin_implementation
def register_communication_plugin(cls) -> Type["ModemAdapter"]:
return cls
def launch_modem(modem: ModemAdapter):
adapter = modem
ground = TCPGround()
checksum_type='crc32',
framer = FpFramerDeframer(checksum_type)
downlinker = Downlinker(modem, ground, framer)
uplinker = Uplinker(adapter, ground, framer, downlinker)
# Open resources for the handlers on either side, this prepares the resources needed for reading/writing data
ground.open()
adapter.open()
# Finally start the processing of uplink and downlink
downlinker.start()
uplinker.start()
uplinker.join()
downlinker.join()Does this essentially capture the approach that the plugin framework implies ? |
Beta Was this translation helpful? Give feedback.
-
|
Hmm, I see what you're saying. imho a package manager is a bit overkill here though. I can say in my own project I have a structure like this: where downlink-client, flightcomputer-software, embedded-nodes-firmware is how the 3 main projects are stored (i.e. our ground system software, low level device and flight computer software respectively) Note that each of those are submodules/sub-repos. Then within flightcomputer-software and downlink-client I have fprime and fprime-gds as submodules. Since there's no release guide or lts I kinda find myself in a situation where I really don't want to change the fprime submodule to something other than the first thing that I got to work. However on the other side since there''s lot of plugin development happening I find myself updating the version of fprime-gds. However since fprime-gds's documentation is in fprime you end up needing to checkout a branch of fprime that is newer than what you will use in the flightcomputer software just to read the documentation that contains references to features used in fprime-gds. I get that we need to be able to make a relative reference to the fprime-gds docs from fprime for the purposes of generating the website however this could still be accomplished if fprime-gds is a submodule of fprime and it has its own docs folder. Wrt to the python example I would say that the manner in which I manage python dependencies depends a bit on how I'm using them. If it's something simple then I'd probably track the version in a pyproject.toml or something and then just pip install it. In some senses this actually points out that we already have package management in the form of cmakelists.txt and/or pyproject.toml. Thats just my two cents though if you think the way I'm doing it is out of left field I'm happy to hear of a better approach. |
Beta Was this translation helpful? Give feedback.
-
|
@ReggieMarr I think your methodology is valid. As someone who does devops as part of my job responsibilities, I prefer methods that emphasize tracking release versions as artifacts rather than source trees. That is orthogonal to the need to review foreign (not in-house development) code. Some differences that crop up here: when developing a solution, interoperability beyond the specific versions you develop against is irrelevant. This means convenience, simplification, or consistency in process can take precedence with respect to dependency management. For a library or framework developer (like @LeStarch ), the end user/developer setup is not constrained and adoption may depend on dependency version flexibility. My point here is that the F Prime project solution to dependency management may not be the same as yours, @ReggieMarr, nor need it be. Your method looks perfectly valid to me, especially in the context of a solution/mission developer. |
Beta Was this translation helpful? Give feedback.
-
|
I see, yeah my usecase is just one of many. I don't mean to convey that having fprime and fprime-gds as submodules in my "3rd party project" is a result tracking fprime-gds as a submodule of fprime. My point is rather I don't see the need for a package manager here. Imho pip and cmake provide a convenient and familiar way of installing specified versions. When it comes to selecting the version we would ideally have a detailed changelog, release notes, and release schedule to inform what versions should interop with one another. Given it would probably be non-trivial to produce those docs I was suggesting that tracking fprime-gds as a submodule to fprime would be an easy way of conveying at least some level of interoperability, this also wouldn't necessarily be a solution "in-tree" since by default cloning fprime wouldn't include the submodule, just a reference to it. And then we'd also have the benefits of discoverability when desired but not including irrelevant docs when not (i.e if someone doesn't care about fprime-gds they won't find references to it in fprime, but if they do they'll found in a manner consistent with fprime). |
Beta Was this translation helpful? Give feedback.
-
|
Yes, I think that an emerging issue in F Prime is that tracking version compatibility is challenging. I do agree that your suggestion does solve it as it currently itches! I have advocated for separating out all of the components in the framework with versioning information built around functionality and port definitions (so it can be automated to some degree) but thats not something that is currently on the F Prime roadmap to my knowledge. Perhaps that helps you see where I am coming from, and that I am not implying that the extreme solution I suggested is an immediate need. |
Beta Was this translation helpful? Give feedback.
-
|
I see yeah I can see how that sort of separation would benefit from some automation and could help for more expansive projects. |
Beta Was this translation helpful? Give feedback.
-
I'm currently working on integrating a previously developed interface to a Newtec MDM6000 modem which serves as the hardware interface between the flight software and the ground station (via RF).
I'd like to leverage the test integration api for deframing/decoding data received from the modem.
Since I'm interfacing with the modem via a udp socket connection it seems to me that I essentially want to provide a wrapper around my modem interface that can use the ThreadedUDPServer as a base class however I've been looking into the use of plugins lately and I wonder if instead I should be providing the modem interface via a plugin.
Any input on which seems to be the more "canonical" way is appreciated.
Beta Was this translation helpful? Give feedback.
All reactions