Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Add EDL commands to XTCE generator #37

Open
wants to merge 27 commits into
base: master
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from 24 commits
Commits
Show all changes
27 commits
Select commit Hold shift + click to select a range
74fe562
add edl commands yaml and initial xtce gen script update
ryanpdx Jul 13, 2024
be8d436
add support for variable str and byte edl cmd arguments
ryanpdx Jul 13, 2024
8cb3ac8
fix xtce command issues found xmllint
ryanpdx Jul 14, 2024
c133696
add support edl bytes data type with a dynamic length
ryanpdx Jul 16, 2024
8f9dca7
add the start of edl command responses to the xtce generator
ryanpdx Jul 19, 2024
895a2a3
refactor xtce generator into multiple functions
ryanpdx Jul 21, 2024
141a020
add encode and decode methods for edl request and response
ryanpdx Jul 25, 2024
30e6008
xtce generator setups an edl response root container for yamcs
ryanpdx Jul 27, 2024
40897ac
edl encode/decode no longer expect uid as part of fields
ryanpdx Jul 27, 2024
c9e38db
add missing edl code parameter and data type to xtce generator
ryanpdx Jul 28, 2024
da67a61
xtce generator does not add empty descriptions anymore
ryanpdx Jul 28, 2024
0d2c6ee
add unit tests for edl yaml config
ryanpdx Jul 28, 2024
1d55b6a
update edl command unit tests to test for max size of string fields
ryanpdx Jul 28, 2024
a5aa6d7
fix missing beacon fields bug with xtce generator
ryanpdx Jul 28, 2024
3dc342a
removed node_id and opd_addr types and encode/decode fixes for edl co…
ryanpdx Jul 29, 2024
54633d7
rename EdlCommandField to SubpacketField
ryanpdx Aug 1, 2024
208f5cc
replaced fields for bytes size with prefix attribute
ryanpdx Aug 2, 2024
0d7da29
fix issue with pascal string size and opd_addr args
ryanpdx Aug 11, 2024
f46961c
add uslp fields to edl reponses
ryanpdx Aug 5, 2024
056d97c
rename edl command definition things
ryanpdx Aug 12, 2024
45b26a8
change edl encode to support non-tuples
ryanpdx Aug 12, 2024
37eea6f
minor generate xtce script fixes
ryanpdx Aug 12, 2024
ed9835b
cleanup and fixes
ryanpdx Aug 12, 2024
b64a422
change yamls long strings and remove edl time sync response
ryanpdx Aug 14, 2024
90ef374
add hmac and uslp fecf to uslp base container
ryanpdx Aug 17, 2024
5ba1562
add missing int encode for bool/enum params; fix loc in container
ryanpdx Aug 17, 2024
272cf4c
edl field attribute size_prefix is now a data_type value
ryanpdx Aug 17, 2024
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 3 additions & 0 deletions .gitignore
Original file line number Diff line number Diff line change
Expand Up @@ -186,3 +186,6 @@ OD.*

# setuptools-scm
*/_version.py

# xml
*.xsd
19 changes: 18 additions & 1 deletion oresat_configs/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -11,7 +11,7 @@
"pyyaml missing/installed without libyaml bindings. See oresat-configs README.md for more"
) from e

from dataclasses import dataclass
import os
from typing import Union

from ._yaml_to_od import (
Expand All @@ -25,6 +25,7 @@
from .beacon_config import BeaconConfig
from .card_info import Card, cards_from_csv
from .constants import Consts, NodeId, OreSatId, __version__
from .edl_cmd_defs import EdlCommandDefinition, EdlCommandDefinitions, EdlCommandField

__all__ = ["Card", "Consts", "NodeId", "OreSatId", "__version__"]

Expand Down Expand Up @@ -57,3 +58,19 @@ def __init__(self, mission: Union[OreSatId, Consts, str]):
self.beacon_def = _gen_c3_beacon_defs(c3_od, beacon_config)
self.fram_def = _gen_c3_fram_defs(c3_od, self.configs["c3"])
self.fw_base_od = _gen_fw_base_od(mission, FW_COMMON_CONFIG_PATH)

# edl commands definitions
node_ids = {}
opd_addrs = {}
for name in self.configs:
card = self.cards[name]
if card.node_id != 0:
node_ids[name] = card.node_id
if card.opd_address != 0:
opd_addrs[name] = card.opd_address
custom_enums = {
"node_id": node_ids,
"opd_addr": opd_addrs,
}
edl_file_path = f"{os.path.dirname(os.path.abspath(__file__))}/edl_cmd_defs.yaml"
self.edl_cmd_defs = EdlCommandDefinitions(edl_file_path, custom_enums)
2 changes: 1 addition & 1 deletion oresat_configs/base/c3.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -118,7 +118,7 @@ objects:
- subindex: 0x8
name: good_test_threshold
data_type: uint16
description: |
description: >-
when testing the antennas, any voltage above this threshold is considered good
access_type: const
default: 100
Expand Down
7 changes: 3 additions & 4 deletions oresat_configs/base/solar.yaml
Original file line number Diff line number Diff line change
Expand Up @@ -127,10 +127,9 @@ objects:
- index: 0x4003
name: mppt_alg
data_type: uint8
description: |
mppt (maximum power point tracking) algorithm
- 0: perturb and observse
access_type: rw
description: mppt (maximum power point tracking) algorithm
value_descriptions:
perturb_and_observse: 0

- index: 0x4004
name: lt1618_iadj
Expand Down
226 changes: 226 additions & 0 deletions oresat_configs/edl_cmd_defs.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,226 @@
"""Used to parse edl_cmd_defs.yaml that defines all EDL command requests and responses."""

import struct
from dataclasses import dataclass, field
from typing import Any, Union

from dacite import from_dict
from yaml import CLoader, load

_COMMAND_DATA_FMT = {
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Similar to "magic numbers", are we using magic letters here? Or is this tied to a spec?

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

These are struct format characters. You can see them being used down below in struct.pack() and struct.unpack() for format strings.

"bool": "?",
"int8": "b",
"int16": "h",
"int32": "i",
"int64": "q",
"uint8": "B",
"uint16": "H",
"uint32": "I",
"uint64": "Q",
"float32": "f",
"float64": "d",
}


@dataclass
class EdlCommandField:
"""A field in EDL command request or response packet."""

name: str
"""
str: Unique name (scope of the fields in the command, not all fields in all commands) for the
EDL command field.
"""
data_type: str
"""
str: Data type of field.

Can be:
- signed integers: `"int8"`, `"int16"`, `"int32"`, `"int64"`
- unsigned integers: `"uint8"`, `"uint16"`, `"uint32"`, `"uint64"`
- floats: `"float32"`, `"float64"`
- string: `"str"` (NOTE: `fix_size` or `max_size` must be set.)
- binary: `"bytes"` (NOTE: `fix_size` or `size_prefix` must be set.)
Comment on lines +42 to +43
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It took a bit for me to grok this, I think it might make more sense to have different unique types for the fixed and dynamic length versions of str and bytes. I usually see the dynamic str here referred to as c-style strings so maybe cstring or cstr and the dynamic length bytes as length-value or LV so maybe lvbytes. If you squint you could probably also get away with bytearray as the python type vaguely has the same connotation.

This way all str must and only must have fixed_size set and cstr must only have max_size set. Similarly with bytes and fixed_size, and lvbytes and size_prefix. Each type now has a unique behavior instead of conditional overlapping behaviors.

"""
description: str = ""
"""str: A short description of the EDL command field."""
enums: dict[str, int] = field(default_factory=dict)
"""dict[str, int]: Enum values for "uintX" or "bool" types."""
max_size: int = 0
"""
int: Max size in bytes for variable "str" data types. String must end with a '\0'.
This takes precedence over fix_size.
"""
size_prefix: int = 0
"""
int: Number of leading prefix bytes used to determind the size of a "bytes" field.
This takes precedence over fix_size.
"""
ryanpdx marked this conversation as resolved.
Show resolved Hide resolved
fixed_size: int = 0
"""
int: Fixed size in bytes for "bytes" or "str" data types. Value that are not the full size
will be filled with "\0" at the end as padding.
"""
unit: str = ""
"str: Optional unit for the field"


@dataclass
class EdlCommandDefinition:
"""A EDL command."""

uid: int
"""int: Unique id to identify the EDL command."""
name: str
"""str: A unique snake_case name for the EDL command."""
description: str = ""
"""str: A short description of the EDL command."""
request: list[EdlCommandField] = field(default_factory=list)
"""list[EdlCommandDefinition]: List of request fields for the EDL command."""
response: list[EdlCommandField] = field(default_factory=list)
"""list[EdlCommandDefinition]: List of response fields for the EDL command."""

def _dynamic_len(self, fields: list[EdlCommandField]) -> bool:
return True in [f.size_prefix != 0 for f in fields]

def _decode(self, raw: bytes, fields: list[EdlCommandField]) -> tuple[Any]:

# fixed size packet - quick decode
if not self._dynamic_len(fields):
fmt = "".join([_COMMAND_DATA_FMT[f.data_type] for f in fields])
return struct.unpack(fmt, raw)

# dynamic size packet - slower decode
data: dict[str, Any] = {}
offset = 0
for f in fields:
if f.data_type in _COMMAND_DATA_FMT:
data_type_size = struct.calcsize(_COMMAND_DATA_FMT[f.data_type])
tmp = raw[offset : offset + data_type_size]
fmt = _COMMAND_DATA_FMT[f.data_type]
data[f.name] = struct.unpack(fmt, tmp)[0]
elif f.data_type == "bytes":
if f.size_prefix != 0: # dynamic length in bits
data_type_size_raw = raw[offset : offset + f.size_prefix]
data_type_size = int.from_bytes(data_type_size_raw, "little") // 8
offset += f.size_prefix
else: # fix_size
data_type_size = f.fixed_size
data[f.name] = raw[offset : offset + data_type_size]
elif f.data_type == "str":
if f.max_size != "": # dynamic length that ends with "\0"
data_type_size = raw[offset:].find(b"\0")
else: # fix_size
data_type_size = f.fixed_size
data[f.name] = raw[offset : offset + data_type_size].decode()
else:
raise ValueError(f"invalid edl field {f.name}")
offset += data_type_size

return tuple(data.values())

def _encode(self, values: tuple[Any], fields: list[EdlCommandField]) -> bytes:

if not isinstance(values, (tuple, list)):
values = (values,)

if len(values) != len(fields):
raise ValueError(
f"invalid number of values for packet: got {len(values)} expected {len(fields)}"
)

# fixed size packet - quick encode
if not self._dynamic_len(fields):
fmt = "".join([_COMMAND_DATA_FMT[f.data_type] for f in fields])
return struct.pack(fmt, *values)

# dynamic size packet - slower encode
raw = b""
for f, v in zip(fields, values):
if f.data_type in _COMMAND_DATA_FMT:
fmt = _COMMAND_DATA_FMT[f.data_type]
raw += struct.pack(fmt, v)
elif f.data_type == "bytes":
value = v
if f.size_prefix != 0: # dynamic length in bits
fmt = _COMMAND_DATA_FMT[f"uint{f.size_prefix * 8}"]
raw += struct.pack(fmt, len(v) * 8)
else: # fixed length
value += b"\x00" * (f.fixed_size - len(value))
raw += value
elif f.data_type == "str":
value = v.encode()
if f.max_size != "": # dynamic length that ends with "\0"
value += b"\0"
else: # fixed length
value += b"\0" * (f.fixed_size - len(value))
raw += value
else:
raise ValueError(f"invalid data type {f.data_type} for edl field {f.name}")
return raw

def decode_request(self, raw: bytes) -> tuple[Any]:
"""Decode a EDL request payload."""
return self._decode(raw, self.request)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm having a hard time grokking why self.request is embedded into the object but values are not.

Is this object intended to keep-state on anything?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nope, it is just the definition


def encode_request(self, values: tuple[Any]) -> bytes:
"""Encode a EDL request payload."""
return self._encode(values, self.request)

def decode_response(self, raw: bytes) -> tuple[Any]:
"""Decode a EDL response payload."""
return self._decode(raw, self.response)

def encode_response(self, values: tuple[Any]) -> bytes:
"""Encode a EDL reponse payload."""
return self._encode(values, self.response)
Comment on lines +163 to +177
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

In non-test code there's no single user that will use both encode_request() and decode_request() at the same time (similarly with the response()s). This presents an API that can be misused, I'd like to see it presented in a way that would be harder to "you're holding it wrong" with.

I'd do it with a separate class for the sender and the receiver, or client/server, source/dest, etc. Something like:

class EdlCommandSource:
    def __init__(self, request, response):
        self.decode_fmt = response
        self.encode_fmt = request

    def decode(self, raw: bytes) -> tuple[Any]:
        ...

    def encode(self, values: tuple[Any]) -> bytes:
        ...

class EdlCommandDest(EdlCommandSource):
    def __init__(self, request, response):
        super().__init__(response, request)

With this the ground scripts only use EdlCommandSource and EdlService only uses EdlCommandDest and correct use is ensured via the type system.

As an added bonus most of the struct fmt strings can be generated once in __init__ instead of each time .encode()/decode() is called.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I don't see the need for this.



class EdlCommandDefinitions:
"""
A custom dictionary-like class to store EDL commands that can use the EDL command uid and EDL
command name as keys.
"""

def __init__(self, file_path: str, custom_enums: dict[str, dict[str, int]] = {}):
self._names: dict[str, EdlCommandDefinition] = {}
self._uids: dict[int, EdlCommandDefinition] = {}

_raw = {}
with open(file_path, "r") as f:
_raw = load(f, Loader=CLoader)

for command_raw in _raw.get("commands", []):
command = from_dict(data_class=EdlCommandDefinition, data=command_raw)
command.description = command.description.replace("\n", "")
for req in command.request:
req.description = req.description.replace("\n", "")
if req.name in custom_enums:
req.enums = custom_enums[req.name]
for res in command.response:
res.description = res.description.replace("\n", "")
if res.name in custom_enums:
res.enums = custom_enums[res.name]
self._uids[command.uid] = command
self._names[command.name] = command

def __getitem__(self, value: Union[int, str]) -> EdlCommandDefinition:
return self._uids.get(value) or self._names.get(value) # type: ignore

def __len__(self) -> int:
return len(self._uids)

def __iter__(self):
return iter(self._uids)

def values(self):
"""Get dictionary values."""
return self._uids.values()

def names(self):
"""Get command names."""
return self._names.keys()

def uid(self):
"""Get command unique ids."""
return self._uids.keys()
Loading