Integrating an RF modem interface with fprime-gds

[ReggieMarr]

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.

[Joshua-Anderson]

Is this interfacing with the modem on the FSW or the GDS side of things?

[Joshua-Anderson]

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

[LeStarch]

@Joshua-Anderson is spot-on. I definitely recommend using plugins. A few helpful tips as plugins are new functionality:

1. You'll want to clone https://github.com/fprime-community/fprime-gds.git and install it with pip install /path/to/clone.
2. Use the "communication" plugin category to setup a custom reader of UDP
3. Use the "framing" plugin category if you need to frame/deframe non-standard frames (not using the fprime frame format)
4. Ask questions! These plugins are for your purpose....but you are an initial user. We are here to help!

[LeStarch]

Oh, and any feedback is appreciated!

[ReggieMarr]

looks great ! I'll update here as I'm sure I'll have questions as I get going (If that seems appropriate).
I actually started off thinking I'd have to maintain a forked version of fprime-gds so this def seems like a better way to go

[ReggieMarr]

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 cls

Where 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 result

How can I pass through the modem config at this level and "select" it as the adapter of choice ?

[ReggieMarr]

I've expanded on my initial implementation a bit but I'll leave the reply comment up for posterity.
This is what I have now:

#!/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 ?

[ReggieMarr]

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:

├── Dockerfile
├── downlink-client
│   ├── docker-compose.yml
│   ├── Dockerfile
│   ├── gds
│   │   ├── custom-deps
│   │   │   └── fprime-gds
│   ├── pyrightconfig.json
│   └── README.md
├── embedded-nodes-firmware
│   ├── common
│   ├── libs
│   ├── nodes
│   └── README.md
├── flightcomputer-software
│   ├── docker-compose.yml
│   ├── Dockerfile
│   ├── FlightComputer
│   ├── fprime
│   ├── pyrightconfig.json
│   └── README.md
├── README.md
└── tools
    └── generate_release_description.py

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.
However if its a dependency that it more wide ranging that I find myself reviewing the source code for or potentially making changes to it then I would track it as a submodule.

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.

[SterlingPeet]

@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.

[ReggieMarr]

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).

[SterlingPeet]

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.

[ReggieMarr]

I see yeah I can see how that sort of separation would benefit from some automation and could help for more expansive projects.