From 74a37e5e6011ba2699e3ca9f3c5349329da37c0c Mon Sep 17 00:00:00 2001 From: Alex Mariano Date: Thu, 9 Jul 2026 18:00:11 -0500 Subject: [PATCH] Added pop() feature --- src/fpy/SPEC.md | 1 + src/fpy/bytecode/SPEC.md | 27 +++ src/fpy/bytecode/directives.py | 14 ++ src/fpy/bytecode/grammar.lark | 4 +- src/fpy/codegen.py | 35 ++- src/fpy/macros.py | 11 + src/fpy/model.py | 16 +- src/fpy/semantics.py | 152 +++++++++++++ test/fpy/RefTopologyDictionary.json | 37 ++++ test/fpy/golden/pop.fpy | 20 ++ test/fpy/golden/pop.fpybc | 20 ++ test/fpy/test_dictionary.py | 2 +- test/fpy/test_pop.py | 322 ++++++++++++++++++++++++++++ 13 files changed, 653 insertions(+), 8 deletions(-) create mode 100644 test/fpy/golden/pop.fpy create mode 100644 test/fpy/golden/pop.fpybc create mode 100644 test/fpy/test_pop.py diff --git a/src/fpy/SPEC.md b/src/fpy/SPEC.md index 928ee43..b6e24b1 100644 --- a/src/fpy/SPEC.md +++ b/src/fpy/SPEC.md @@ -129,6 +129,7 @@ Available macros: * `iabs(value: I64) -> I64`: returns the absolute value of a signed 64-bit integer. * `fabs(value: F64) -> F64`: returns the absolute value of a 64-bit float. * `log(message: String, severity: Fw.LogSeverity = Fw.LogSeverity.ACTIVITY_HI)`: pushes the severity, message, and message size onto the stack, then emits a `PopEventDirective`. Both arguments must be compile-time constants. +* `pop(port, value)`: pushes `value` onto the stack and emits a `PopSerializableDirective` that pops those bytes and sends them out the sequencer's `serialOut[port]` port. `port` must be a compile-time constant (typically a `Svc.Fpy.SerialPortIndex` enum constant) validated at compile time to be in `[0, MAX_SERIAL_PORTS)`; `value` may be any constant-size expression and its serialized size becomes the directive's `size` field. ## Type constructors Structs, arrays, and `Fw.TimeValue` expose constructors whose callable name is the fully qualified type name. Their arguments correspond to the members in declaration order (struct fields by name, array elements as `e0`, `e1`, ..., and `Fw.TimeValue` with `timeBase: TimeBase`, `timeContext: U8`, `seconds: U32`, `useconds: U32`). A constructor call serializes the provided values into a new instance of that type. `Fw.Time` is an alias for `Fw.TimeValue`, and `Fw.TimeInterval` is an alias for `Fw.TimeIntervalValue`. diff --git a/src/fpy/bytecode/SPEC.md b/src/fpy/bytecode/SPEC.md index 337fae7..b6b8a21 100644 --- a/src/fpy/bytecode/SPEC.md +++ b/src/fpy/bytecode/SPEC.md @@ -1136,3 +1136,30 @@ If this is called without the seed being manually set beforehand, then the seed | ------------------|-------------| | U32 | The next pseudorandom 32-bit value from the sequencer's internal PRNG | +## POP_SERIALIZABLE (78) +Pops `size` bytes of serialized data from the stack and sends them to an external component via the sequencer's `serialOut` output port array. + +Exposed to sequences via the `pop(port, value)` builtin, where `port` is a compile-time constant (typically a `Svc.Fpy.SerialPortIndex` enum constant) and `value` is any constant-size expression. The compiler pushes `value` onto the stack and emits `POP_SERIALIZABLE` with `size` set to the serialized size of `value`'s type. + +**Preconditions:** +- `port_index < MAX_SERIAL_PORTS` (from the `Svc.Fpy.SerialPortIndex` enum in `FpySequencerCfg`) +- `serialOut[port_index]` is connected to a target component +- `len(stack) >= size` + +**Semantics:** +1. Validate the port index is within bounds. +2. Check that the specified port is connected. +3. Pop `size` bytes from the stack. +4. Send the popped bytes to the target component via `serialOut[port_index]`. + +**Error Conditions:** +- If `port_index >= MAX_SERIAL_PORTS`: `SERIAL_PORT_INVALID_INDEX` +- If `serialOut[port_index]` is not connected: `SERIAL_PORT_NOT_CONNECTED` +- If `len(stack) < size`: `STACK_UNDERFLOW` + +| Arg Name | Arg Type | Source | Description | +|------------|---------------|------------|-------------| +| port_index | FwIndexType | hardcoded | Index of the `serialOut` port array to use. | +| size | StackSizeType | hardcoded | Number of bytes to pop and send. | +| value | bytes | stack | Serialized data to send (popped from stack). | + diff --git a/src/fpy/bytecode/directives.py b/src/fpy/bytecode/directives.py index e604585..bda79c7 100644 --- a/src/fpy/bytecode/directives.py +++ b/src/fpy/bytecode/directives.py @@ -38,6 +38,7 @@ FwChanIdType = FpyType(TypeKind.U32, "U32") FwPrmIdType = FpyType(TypeKind.U32, "U32") FwOpcodeType = FpyType(TypeKind.U32, "U32") +FwIndexType = FpyType(TypeKind.I16, "I16") ArrayIndexType = I64 @@ -66,6 +67,7 @@ def update_configurable_types_from_dict(type_defs: dict[str, FpyType]) -> None: _update_configurable_type(FwChanIdType, type_defs, "FwChanIdType") _update_configurable_type(FwPrmIdType, type_defs, "FwPrmIdType") _update_configurable_type(FwOpcodeType, type_defs, "FwOpcodeType") + _update_configurable_type(FwIndexType, type_defs, "FwIndexType") _update_configurable_type(FwSizeStoreType, type_defs, "FwSizeStoreType") @@ -172,6 +174,7 @@ class DirectiveId(Enum): POP_EVENT = 75 SET_SEED = 76 PUSH_RAND = 77 + POP_SERIALIZABLE = 78 # ───────────────────────────────────────────────────────────────────────────── @@ -403,6 +406,17 @@ class PopEventDirective(Directive): _FIELD_TYPES: ClassVar[dict[str, FpyType]] = {} +@dataclass +class PopSerializableDirective(Directive): + opcode: ClassVar[DirectiveId] = DirectiveId.POP_SERIALIZABLE + portIndex: int + size: int + _FIELD_TYPES: ClassVar[dict[str, FpyType]] = { + "portIndex": FwIndexType, + "size": StackSizeType, + } + + # ───────────────────────────────────────────────────────────────────────────── # Arithmetic stack op directives (no fields) # ───────────────────────────────────────────────────────────────────────────── diff --git a/src/fpy/bytecode/grammar.lark b/src/fpy/bytecode/grammar.lark index 3c30ec4..5cffda2 100644 --- a/src/fpy/bytecode/grammar.lark +++ b/src/fpy/bytecode/grammar.lark @@ -22,6 +22,7 @@ goto_tag: NAME ":" | "store_abs_const_offset" _int _uint | "get_field" _uint _uint | "return" _uint _uint + | "pop_serializable" _int _uint !op_with_no_args: "no_op" | "siext_8_64" @@ -78,8 +79,9 @@ goto_tag: NAME ":" | "push_time" | "push_rand" | "set_seed" - | "call" + | "call" | "peek" + | "pop_event" _uint: [UNARY_IDENTITY] DEC_NUMBER _int: NEG_DEC_NUMBER _float: FLOAT_NUMBER diff --git a/src/fpy/codegen.py b/src/fpy/codegen.py index 6ffb843..d1b65af 100644 --- a/src/fpy/codegen.py +++ b/src/fpy/codegen.py @@ -81,6 +81,7 @@ IntegerZeroExtend8To64Directive, OrDirective, PeekDirective, + PopSerializableDirective, FloatMultiplyDirective, GetFieldDirective, IntAddDirective, @@ -1060,12 +1061,36 @@ def emit_AstFuncCall(self, node: AstFuncCall, state: CompileState): assert const_val is not None, f"const arg {i} of {func.name} should have been validated by semantics" const_arg_values[i] = const_val - # put non-const arg values on stack - for i, arg_node in enumerate(node_args): - if i not in func.const_arg_indices: - dirs.extend(self._emit_func_arg(arg_node, state)) + # Special case for pop: push value arg, then emit directive + if func.name == "pop": + value_arg = node_args[1] + dirs.extend(self._emit_func_arg(value_arg, state)) + value_type = state.contextual_types[value_arg] + + # Get the port index from const args + # Semantics has already validated this is int or enum + port_val = const_arg_values[0] + if isinstance(port_val.val, int): + port_int = port_val.val + elif isinstance(port_val.val, str) and port_val.type.kind == TypeKind.ENUM: + # Enum constant: look up the integer value + port_int = port_val.type.enum_dict[port_val.val] + else: + # This should never happen - semantics validates port type + assert False, f"Invariant violation: port should be int or enum after semantics validation, got {port_val}" + + # Get size from the value's type + # Semantics has validated this is a concrete type with computable max_size + size = value_type.max_size + + dirs.append(PopSerializableDirective(portIndex=port_int, size=size)) + else: + # put non-const arg values on stack + for i, arg_node in enumerate(node_args): + if i not in func.const_arg_indices: + dirs.extend(self._emit_func_arg(arg_node, state)) - dirs.extend(func.generate(node, const_arg_values)) + dirs.extend(func.generate(node, const_arg_values)) elif is_instance_compat(func, TypeCtorSymbol): # put arg values onto stack in correct order for serialization for arg_node in node_args: diff --git a/src/fpy/macros.py b/src/fpy/macros.py index f06e20a..ebbbe4c 100644 --- a/src/fpy/macros.py +++ b/src/fpy/macros.py @@ -4,6 +4,7 @@ ExitDirective, FloatLogDirective, PopEventDirective, + PopSerializableDirective, PushTimeDirective, SetSeedDirective, PushValDirective, @@ -211,4 +212,14 @@ def generate_randf(node: Ast, const_args: dict[int, FpyValue]) -> list[Directive ], const_arg_indices=frozenset({0, 1}), ), + # Serial pop builtin — compile-time port index, any constant-size value + # The generate function is a placeholder; codegen.py handles this specially + "pop": BuiltinFuncSymbol( + "pop", NOTHING, [ + ("port", None, None), # None means "any constant-size type" + ("value", None, None), # None means "any constant-size type" + ], + lambda n, c: [], # Placeholder; codegen.py emits the directive + const_arg_indices=frozenset({0}), # port must be compile-time constant + ), } diff --git a/src/fpy/model.py b/src/fpy/model.py index 49f07ae..c3ecd4e 100644 --- a/src/fpy/model.py +++ b/src/fpy/model.py @@ -13,6 +13,7 @@ ConstCmdDirective, Directive, PopEventDirective, + PopSerializableDirective, FwOpcodeType, PeekDirective, ExitDirective, @@ -136,6 +137,8 @@ class DirectiveErrorCode(Enum): STACK_UNDERFLOW = 15 INVALID_ARG = 16 CMD_FAIL = 17 + SERIAL_PORT_NOT_CONNECTED = 18 + SERIAL_PORT_INVALID_INDEX = 19 class FpySequencerModel: @@ -488,7 +491,7 @@ def handle_store_abs_const_offset(self, dir: StoreAbsConstOffsetDirective): return None def handle_push_val(self, dir: PushValDirective): - if len(self.stack) + 8 > self.max_stack_size: + if len(self.stack) + len(dir.val) > self.max_stack_size: return DirectiveErrorCode.STACK_OVERFLOW self.push(dir.val) return None @@ -1305,3 +1308,14 @@ def handle_pop_event(self, dir: PopEventDirective): if debug: print(f"POP_EVENT [{severity_name}]: {message.decode('utf-8')}") return None + + def handle_pop_serializable(self, dir: PopSerializableDirective): + # Check stack has enough bytes + if len(self.stack) < dir.size: + return DirectiveErrorCode.STACK_UNDERFLOW + # Pop the bytes (in model, we just discard them like pop_event does) + popped_bytes = self.pop(type=bytes, size=dir.size) + if debug: + print(f"POP_SERIALIZABLE port={dir.portIndex} size={dir.size} bytes={popped_bytes.hex()}") + # In the model we don't have actual serial ports, so we just discard the data + return None diff --git a/src/fpy/semantics.py b/src/fpy/semantics.py index 7fafdab..3414485 100644 --- a/src/fpy/semantics.py +++ b/src/fpy/semantics.py @@ -1943,6 +1943,11 @@ def resolve_args( if value_expr not in state.synthesized_types: continue + # Skip type check if arg_type is None (sentinel for "any type") + # Used by builtins like pop that accept any constant-size type + if arg_type is None: + continue + unconverted_type = state.synthesized_types[value_expr] if not self.can_coerce_type(unconverted_type, arg_type): return CompileError( @@ -2002,6 +2007,9 @@ def visit_AstFuncCall(self, node: AstFuncCall, state: CompileState): # These will be coerced when the function definition is visited. if value_expr not in state.synthesized_types: continue + # Skip coercion if target_type is None (sentinel for "any type") + if target_type is None: + continue assert self.coerce_expr_type(value_expr, target_type, state) state.synthesized_types[node] = func.return_type @@ -2529,6 +2537,150 @@ def visit_AstFuncCall(self, node: AstFuncCall, state: CompileState): ) return + # Special validation for pop after confirming args are const + if func.name == "pop": + port_val = arg_values[0] + value_arg = resolved_args[1] + + # Extract integer value from port_val + # Validate port value type - must be integer (reject float, Decimal, bool) + # Python bool is a subclass of int, so check bool first + if isinstance(port_val.val, bool): + state.errors.append( + CompileError( + "pop port argument must be an integer or enum constant, not bool", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + elif isinstance(port_val.val, (Decimal, float)): + state.errors.append( + CompileError( + f"pop port argument must be an integer or enum constant, not float/Decimal", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + elif isinstance(port_val.val, int): + # Integer value - OK (includes abstract INTEGER type from literals) + port_int = port_val.val + elif isinstance(port_val.val, str): + # Enum constant: look up the integer value from the enum dict + if port_val.type.kind == TypeKind.ENUM and hasattr(port_val.type, 'enum_dict'): + port_int = port_val.type.enum_dict.get(port_val.val) + if port_int is None: + state.errors.append( + CompileError( + f"Unknown enum value '{port_val.val}' in {port_val.type.name}", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + else: + state.errors.append( + CompileError( + "pop port argument must be an integer or enum constant", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + else: + # Reject any other type + state.errors.append( + CompileError( + f"pop port argument must be an integer or enum constant, got {type(port_val.val).__name__}", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + + # Get MAX_SERIAL_PORTS from the dictionary enum + serial_port_enum = state.type_defs.get("Svc.Fpy.SerialPortIndex") + if serial_port_enum is None: + state.errors.append( + CompileError( + "Svc.Fpy.SerialPortIndex enum not found in dictionary", + node, + ) + ) + return + + max_ports = serial_port_enum.enum_dict.get("MAX_SERIAL_PORTS") + if max_ports is None: + state.errors.append( + CompileError( + "MAX_SERIAL_PORTS not found in Svc.Fpy.SerialPortIndex enum", + node, + ) + ) + return + + if not (0 <= port_int < max_ports): + state.errors.append( + CompileError( + f"pop port index {port_int} out of range [0, {max_ports})", + resolved_args[0] if is_instance_compat(resolved_args[0], Ast) else node, + ) + ) + return + + # Validate value is constant-sized + if is_instance_compat(value_arg, FpyValue): + value_type = value_arg.type + else: + value_type = state.synthesized_types.get(value_arg) + if value_type is None: + state.errors.append( + CompileError( + "Cannot determine type of pop value argument", + value_arg, + ) + ) + return + + # Reject abstract INTEGER/FLOAT types (bare literals without a concrete type) + # These have no computable max_size and would crash codegen + if value_type.kind in (TypeKind.INTEGER, TypeKind.FLOAT): + state.errors.append( + CompileError( + f"pop value must have a concrete type (e.g., U32(42) not 42). " + f"Bare numeric literals are not allowed.", + value_arg if is_instance_compat(value_arg, Ast) else node, + ) + ) + return + + # Reject anonymous/void kinds that are not serializable and would + # crash codegen (anon struct/array literals, ranges, void values). + # INTERNAL_STRING is intentionally excluded here so that bare + # string literals fall through to the constant-size gate below and + # produce the "constant-sized" error message users expect. + if value_type.kind in ( + TypeKind.ANON_STRUCT, + TypeKind.ANON_ARRAY, + TypeKind.RANGE, + TypeKind.NOTHING, + ): + state.errors.append( + CompileError( + "pop value must be a concrete, serializable value " + "(a variable, a type constructor like Ref.SignalPair(...), " + "or a cast); anonymous literals, ranges, and void " + "expressions are not allowed.", + value_arg if is_instance_compat(value_arg, Ast) else node, + ) + ) + return + + if not is_type_constant_size(value_type): + state.errors.append( + CompileError( + f"pop value must be constant-sized (no strings), found {value_type.display_name}", + value_arg if is_instance_compat(value_arg, Ast) else node, + ) + ) + return + if unknown_value: # we will have to calculate this at runtime state.const_expr_values[node] = None diff --git a/test/fpy/RefTopologyDictionary.json b/test/fpy/RefTopologyDictionary.json index 5b0f21a..b2014e4 100644 --- a/test/fpy/RefTopologyDictionary.json +++ b/test/fpy/RefTopologyDictionary.json @@ -44,6 +44,43 @@ ], "default" : "Svc.PrmDb.Merge.MERGE" }, + { + "kind" : "enum", + "qualifiedName" : "Svc.Fpy.SerialPortIndex", + "representationType" : { + "name" : "U8", + "kind" : "integer", + "size" : 8, + "signed" : false + }, + "enumeratedConstants" : [ + { + "name" : "EXAMPLE_PORT_0", + "value" : 0 + }, + { + "name" : "EXAMPLE_PORT_1", + "value" : 1 + }, + { + "name" : "EXAMPLE_PORT_2", + "value" : 2 + }, + { + "name" : "EXAMPLE_PORT_3", + "value" : 3 + }, + { + "name" : "EXAMPLE_PORT_4", + "value" : 4 + }, + { + "name" : "MAX_SERIAL_PORTS", + "value" : 5 + } + ], + "default" : "Svc.Fpy.SerialPortIndex.EXAMPLE_PORT_0" + }, { "kind" : "alias", "qualifiedName" : "FwTlmPacketizeIdType", diff --git a/test/fpy/golden/pop.fpy b/test/fpy/golden/pop.fpy new file mode 100644 index 0000000..a23ce8e --- /dev/null +++ b/test/fpy/golden/pop.fpy @@ -0,0 +1,20 @@ +# Test pop feature with various types and port indices + +# Basic U32 pop to port 0 +value1: U32 = 42 +pop(0, value1) + +# U8 pop to port 1 +value2: U8 = 100 +pop(1, value2) + +# F64 pop to port 2 using enum constant +value3: F64 = 3.14159 +pop(Svc.Fpy.SerialPortIndex.EXAMPLE_PORT_2, value3) + +# Struct pop to port 3 +signal_pair: Ref.SignalPair = Ref.SignalPair(time=1.0, value=2.0) +pop(3, signal_pair) + +# Expression value to port 4 (must be wrapped in concrete type) +pop(4, U32(100 + 200)) diff --git a/test/fpy/golden/pop.fpybc b/test/fpy/golden/pop.fpybc new file mode 100644 index 0000000..422df2e --- /dev/null +++ b/test/fpy/golden/pop.fpybc @@ -0,0 +1,20 @@ +push_val 255 +allocate 21 +push_val 0 0 0 42 +store_rel_const_offset 1 4 +load_rel 1 4 +pop_serializable 0 4 +push_val 100 +store_rel_const_offset 5 1 +load_rel 5 1 +pop_serializable 1 1 +push_val 64 9 33 249 240 27 134 110 +store_rel_const_offset 6 8 +load_rel 6 8 +pop_serializable 2 8 +push_val 63 128 0 0 64 0 0 0 +store_rel_const_offset 14 8 +load_rel 14 8 +pop_serializable 3 8 +push_val 0 0 1 44 +pop_serializable 4 4 diff --git a/test/fpy/test_dictionary.py b/test/fpy/test_dictionary.py index 9409baf..c28c709 100644 --- a/test/fpy/test_dictionary.py +++ b/test/fpy/test_dictionary.py @@ -2395,7 +2395,7 @@ def test_loads_ref_dictionary(self): def test_type_counts(self): d = load_dictionary(REF_DICT_PATH) - assert len(d["type_defs"]) == 95 + assert len(d["type_defs"]) == 96 def test_command_counts(self): d = load_dictionary(REF_DICT_PATH) diff --git a/test/fpy/test_pop.py b/test/fpy/test_pop.py new file mode 100644 index 0000000..cfb9e47 --- /dev/null +++ b/test/fpy/test_pop.py @@ -0,0 +1,322 @@ +from fpy.bytecode.directives import Directive, PopSerializableDirective +from fpy.model import DirectiveErrorCode +from fpy.test_helpers import ( + assert_compile_failure, + assert_run_success, + compile_seq, +) + + +class TestPop: + + def test_basic_u32(self, fprime_test_api): + seq = ''' +value: U32 = 42 +pop(Svc.Fpy.SerialPortIndex.EXAMPLE_PORT_0, value) +''' + assert_run_success(fprime_test_api, seq) + + def test_basic_u8(self, fprime_test_api): + seq = ''' +value: U8 = 100 +pop(0, value) +''' + assert_run_success(fprime_test_api, seq) + + def test_emits_pop_serializable_directive(self, fprime_test_api): + seq = ''' +value: U32 = 42 +pop(0, value) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].portIndex == 0 + assert pop_dirs[0].size == 4 # U32 is 4 bytes + + def test_correct_size_for_different_types(self, fprime_test_api): + # size should match the byte width of the popped value + seq = ''' +v1: U8 = 1 +pop(0, v1) +v2: U32 = 2 +pop(1, v2) +v3: F64 = 3.0 +pop(2, v3) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 3 + assert pop_dirs[0].size == 1 # U8 + assert pop_dirs[1].size == 4 # U32 + assert pop_dirs[2].size == 8 # F64 + + def test_enum_port_constant(self, fprime_test_api): + # port index may be given as an enum constant + seq = ''' +value: U32 = 123 +pop(Svc.Fpy.SerialPortIndex.EXAMPLE_PORT_2, value) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].portIndex == 2 + + def test_max_port_index(self, fprime_test_api): + seq = ''' +value: U32 = 42 +pop(4, value) # MAX_SERIAL_PORTS is 5, so 4 is valid +''' + assert_run_success(fprime_test_api, seq) + + def test_port_out_of_bounds_high(self, fprime_test_api): + # port index past the last serial port is rejected at compile time + seq = ''' +value: U32 = 42 +pop(5, value) # MAX_SERIAL_PORTS is 5, so 5 is out of range +''' + assert_compile_failure(fprime_test_api, seq) + + def test_port_out_of_bounds_negative(self, fprime_test_api): + seq = ''' +value: U32 = 42 +pop(-1, value) +''' + assert_compile_failure(fprime_test_api, seq) + + def test_non_constant_port_rejected(self, fprime_test_api): + # the port index must be a compile-time constant + seq = ''' +port: U32 = 0 +value: U32 = 42 +pop(port, value) +''' + assert_compile_failure(fprime_test_api, seq) + + def test_non_constant_size_value_rejected(self, fprime_test_api): + # strings are not constant-sized, so they can't be popped + seq = ''' +pop(0, "test") +''' + assert_compile_failure(fprime_test_api, seq, match="constant-sized") + + def test_serialization_roundtrip(self, fprime_test_api): + # the directive survives a serialize/deserialize round trip + seq = ''' +value: U32 = 42 +pop(1, value) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + + original = pop_dirs[0] + serialized = original.serialize() + _, deserialized = Directive.deserialize(serialized, 0) + assert isinstance(deserialized, PopSerializableDirective) + assert deserialized.portIndex == 1 + assert deserialized.size == 4 + + def test_model_stack_underflow(self, fprime_test_api): + # Popping more bytes than are on the stack is an underflow. Tested at + # the directive level because the builtin always pushes the value + # before popping, so it can't underflow through normal usage. + from fpy.model import FpySequencerModel + + model = FpySequencerModel() + model.stack.extend([0x01, 0x02]) # only 2 bytes available + + directive = PopSerializableDirective(portIndex=0, size=4) # ask for 4 + error_code = model.handle_pop_serializable(directive) + assert error_code == DirectiveErrorCode.STACK_UNDERFLOW + + def test_model_execution_pops_bytes(self, fprime_test_api): + seq = ''' +value: U32 = 0x12345678 +pop(0, value) +''' + assert_run_success(fprime_test_api, seq) + + def test_expression_value(self, fprime_test_api): + # a cast expression is a valid, constant-sized value + seq = ''' +pop(0, U32(100 + 200)) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 4 # U32 + + def test_multiple_pops(self, fprime_test_api): + seq = ''' +v1: U32 = 1 +v2: U32 = 2 +v3: U32 = 3 +pop(0, v1) +pop(1, v2) +pop(2, v3) +''' + assert_run_success(fprime_test_api, seq) + + def test_bare_integer_literal_rejected(self, fprime_test_api): + # a bare int literal has no concrete type; it must be cast first + seq = ''' +pop(0, 42) +''' + assert_compile_failure(fprime_test_api, seq, match="concrete type") + + def test_bare_float_literal_rejected(self, fprime_test_api): + seq = ''' +pop(0, 3.14) +''' + assert_compile_failure(fprime_test_api, seq, match="concrete type") + + def test_bare_expression_rejected(self, fprime_test_api): + seq = ''' +pop(4, 100 + 200) +''' + assert_compile_failure(fprime_test_api, seq, match="concrete type") + + def test_float_port_rejected(self, fprime_test_api): + # the port index must be an integer or enum, not a float + seq = ''' +v: U32 = 42 +pop(1.5, v) +''' + assert_compile_failure(fprime_test_api, seq, match="integer or enum") + + def test_bool_port_rejected(self, fprime_test_api): + seq = ''' +v: U32 = 42 +pop(True, v) +''' + assert_compile_failure(fprime_test_api, seq, match="integer or enum") + + def test_struct_value(self, fprime_test_api): + # Ref.SignalPair: F32 * 2 = 8 bytes + seq = ''' +v: Ref.SignalPair = Ref.SignalPair(time=1.0, value=2.0) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 8 + assert_run_success(fprime_test_api, seq) + + def test_array_value(self, fprime_test_api): + # Ref.SignalSet: F32[4] = 16 bytes + seq = ''' +v: Ref.SignalSet = Ref.SignalSet(1.0, 2.0, 3.0, 4.0) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 16 + assert_run_success(fprime_test_api, seq) + + def test_enum_i32_value(self, fprime_test_api): + # Ref.Choice: I32 representation = 4 bytes + seq = ''' +v: Ref.Choice = Ref.Choice.RED +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 4 + assert_run_success(fprime_test_api, seq) + + def test_enum_u8_value(self, fprime_test_api): + # Svc.BlockState: U8 representation = 1 byte + seq = ''' +v: Svc.BlockState = Svc.BlockState.NO_BLOCK +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 1 + + def test_nested_struct_value(self, fprime_test_api): + # Ref.SignalInfo: SignalType(I32 enum)=4 + SignalSet(F32[4])=16 + # + SignalPairSet(SignalPair[4], 8 each)=32 => 52 bytes + seq = ''' +v: Ref.SignalInfo = Ref.SignalInfo( \\ + Ref.SignalType.TRIANGLE, \\ + Ref.SignalSet(0.0, 0.0, 0.0, 0.0), \\ + Ref.SignalPairSet( \\ + Ref.SignalPair(0.0, 0.0), \\ + Ref.SignalPair(0.0, 0.0), \\ + Ref.SignalPair(0.0, 0.0), \\ + Ref.SignalPair(0.0, 0.0))) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 52 + assert_run_success(fprime_test_api, seq) + + def test_array_of_struct_value(self, fprime_test_api): + # Ref.SignalPairSet: Ref.SignalPair[4], 8 bytes each => 32 bytes + seq = ''' +v: Ref.SignalPairSet = Ref.SignalPairSet( \\ + Ref.SignalPair(1.0, 2.0), \\ + Ref.SignalPair(3.0, 4.0), \\ + Ref.SignalPair(5.0, 6.0), \\ + Ref.SignalPair(7.0, 8.0)) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 32 + assert_run_success(fprime_test_api, seq) + + def test_nested_array_value(self, fprime_test_api): + # Ref.TooManyChoices: Ref.ManyChoices[2], each is Ref.Choice[2] + # Ref.Choice is I32 => 2 * (2 * 4) = 16 bytes + seq = ''' +v: Ref.TooManyChoices = Ref.TooManyChoices( \\ + Ref.ManyChoices(Ref.Choice.ONE, Ref.Choice.TWO), \\ + Ref.ManyChoices(Ref.Choice.RED, Ref.Choice.BLUE)) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 16 + + def test_member_array_struct_value(self, fprime_test_api): + # Ref.ChoiceSlurry: TooManyChoices=16 + Choice(I32)=4 + # + ChoicePair(2 * I32)=8 + U8[2]=2 => 30 bytes + seq = ''' +v: Ref.ChoiceSlurry = Ref.ChoiceSlurry( \\ + Ref.TooManyChoices( \\ + Ref.ManyChoices(Ref.Choice.ONE, Ref.Choice.TWO), \\ + Ref.ManyChoices(Ref.Choice.RED, Ref.Choice.BLUE)), \\ + Ref.Choice.ONE, \\ + Ref.ChoicePair(Ref.Choice.ONE, Ref.Choice.TWO), \\ + [1, 2]) +pop(0, v) +''' + directives, _ = compile_seq(fprime_test_api, seq) + pop_dirs = [d for d in directives if isinstance(d, PopSerializableDirective)] + assert len(pop_dirs) == 1 + assert pop_dirs[0].size == 30 + + def test_struct_with_string_members_rejected(self, fprime_test_api): + # a struct containing a string is not constant-sized + seq = ''' +pop(0, Ref.DpDemo.StructWithStringMembers("a", Ref.DpDemo.StringArray("x", "y"))) +''' + assert_compile_failure(fprime_test_api, seq, match="constant-sized") + + def test_string_array_rejected(self, fprime_test_api): + # an array of strings is not constant-sized + seq = ''' +pop(0, Ref.DpDemo.StringArray("a", "b")) +''' + assert_compile_failure(fprime_test_api, seq, match="constant-sized")