From 6ac0e103b24e8923129f8027f1459bda12cb01f8 Mon Sep 17 00:00:00 2001 From: Zakariya Ghazzol Date: Fri, 16 Oct 2026 01:49:32 -0400 Subject: [PATCH 1/2] Use configured time-context types in time() --- src/fpy/semantics.py | 2 +- src/fpy/state.py | 14 ++- test/fpy/test_time_context_config.py | 175 +++++++++++++++++++++++++++ 3 files changed, 186 insertions(+), 5 deletions(-) create mode 100644 test/fpy/test_time_context_config.py diff --git a/src/fpy/semantics.py b/src/fpy/semantics.py index ff841f4..e81e716 100644 --- a/src/fpy/semantics.py +++ b/src/fpy/semantics.py @@ -2295,7 +2295,7 @@ def _parse_time_string( TIME, { "timeBase": FpyValue(TIME_BASE, time_base), - "timeContext": FpyValue(U8, time_context), + "timeContext": FpyValue(TIME.members[1].type, time_context), "seconds": FpyValue(U32, seconds), "useconds": FpyValue(U32, useconds), }, diff --git a/src/fpy/state.py b/src/fpy/state.py index 4e5703f..44b886c 100644 --- a/src/fpy/state.py +++ b/src/fpy/state.py @@ -59,6 +59,7 @@ TIME_COMPARISON, TIME_INTERVAL, TLM_VALID, + U8, FpyType, FpyValue, TypeKind, @@ -423,10 +424,8 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None: def _update_time_context_type_from_dict( dict_type_name_dict: dict[str, FpyType], ) -> None: - """Update TIME's timeContext member type from FwTimeContextStoreType.""" - if "FwTimeContextStoreType" not in dict_type_name_dict: - return # Keep the default U8 - ctx_type = dict_type_name_dict["FwTimeContextStoreType"] + """Configure the time member and builtin argument, defaulting to U8.""" + ctx_type = dict_type_name_dict.get("FwTimeContextStoreType", U8) if not ctx_type.is_primitive: raise DictionaryError( "FwTimeContextStoreType", @@ -434,6 +433,10 @@ def _update_time_context_type_from_dict( f"it as {ctx_type}.", ) TIME.members[1].type = ctx_type + if TIME.member_defaults is not None: + default = TIME.member_defaults["timeContext"] + TIME.member_defaults["timeContext"] = FpyValue(ctx_type, default.val) + TIME_MACRO.args[2] = ("timeContext", ctx_type, FpyValue(ctx_type, 0)) def _get_elem_type_default(elem_type: FpyType) -> FpyValue: @@ -701,6 +704,9 @@ def get_base_compile_state( """return the initial state of the compiler, based on the given dict path""" if default_time_base is None: default_time_base = DEFAULT_TIME_BASE + # TIME and time() are shared by cached scopes. Restore this dictionary's + # context type even when _build_global_scopes returns a cached entry. + _update_time_context_type_from_dict(load_dictionary(dictionary)["type_defs"]) type_scope, callable_scope, values_scope, type_defs, type_ctors = ( _build_global_scopes(dictionary, default_time_base) ) diff --git a/test/fpy/test_time_context_config.py b/test/fpy/test_time_context_config.py new file mode 100644 index 0000000..4188afd --- /dev/null +++ b/test/fpy/test_time_context_config.py @@ -0,0 +1,175 @@ +"""Regression tests for dictionary-configured time contexts.""" + +import json +from pathlib import Path + +import pytest + +from fpy.compiler import ( + analysis_to_fpybc_directives, + analysis_to_wasm, + analyze_ast, + text_to_ast, +) +from fpy.dictionary import load_dictionary +from fpy.error import CompileError +from fpy.macros import TIME_MACRO +from fpy.state import _build_global_scopes, get_base_compile_state +from fpy.test_helpers import FPYBC, assert_run_success, default_dictionary +from fpy.types import TIME, U8 + + +@pytest.fixture +def context_dictionary(tmp_path): + paths = {} + + def make(width): + if width in paths: + return paths[width] + data = json.loads(Path(default_dictionary).read_text(encoding="utf-8")) + integer = { + "name": f"U{width or 8}", + "kind": "integer", + "size": width or 8, + "signed": False, + } + for definition in data["typeDefinitions"]: + if definition["qualifiedName"] == "FwTimeContextStoreType": + definition["type"] = integer + definition["underlyingType"] = dict(integer) + break + else: + raise AssertionError("Reference dictionary has no time-context alias") + if width is None: + data["typeDefinitions"].remove(definition) + for definition in data["typeDefinitions"]: + if definition["qualifiedName"] == "Fw.TimeValue": + definition["members"]["timeContext"]["type"] = integer + path = tmp_path / f"time-context-u{width}.json" + path.write_text(json.dumps(data), encoding="utf-8") + paths[width] = str(path) + return paths[width] + + yield make + load_dictionary.cache_clear() + _build_global_scopes.cache_clear() + get_base_compile_state(default_dictionary) + + +def _time_value(dictionary, context): + suffix = "" if context is None else f", timeContext={context}" + body = text_to_ast(f'time("2023-11-14T22:13:20.5Z"{suffix})\n') + call = body.stmts[0] + state = analyze_ast(body, get_base_compile_state(dictionary)) + return state.const_expr_values[call] + + +@pytest.mark.parametrize("width", [8, 16, 32, 64]) +def test_time_builtin_context_parameter_type(context_dictionary, width): + dictionary = context_dictionary(width) + get_base_compile_state(dictionary) + configured = load_dictionary(dictionary)["type_defs"]["FwTimeContextStoreType"] + assert TIME_MACRO.args[2][1] == configured + assert TIME_MACRO.args[2][2].type == configured + + +@pytest.mark.parametrize("width", [8, 16, 32, 64]) +@pytest.mark.parametrize("context", [None, 9], ids=["default-context", "context-nine"]) +def test_time_constant_context_layout(context_dictionary, width, context): + dictionary = context_dictionary(width) + value = _time_value(dictionary, context) + types = load_dictionary(dictionary)["type_defs"] + configured = types["FwTimeContextStoreType"] + expected = ( + types["TimeBase"] + .enum_dict["TB_WORKSTATION_TIME"] + .to_bytes(types["TimeBase"].max_size, "big") + + (0 if context is None else context).to_bytes(width // 8, "big") + + (1_700_000_000).to_bytes(4, "big") + + (500_000).to_bytes(4, "big") + ) + serialized = value.serialize() + assert len(serialized) == len(expected) + assert serialized == expected + assert value.val["timeContext"].type == configured + assert len(serialized) == value.type.max_size + + +@pytest.mark.parametrize("width", [16, 32, 64]) +def test_wide_time_context_accepts_300(context_dictionary, width): + value = _time_value(context_dictionary(width), 300) + assert value.val["timeContext"].val == 300 + assert value.val["timeContext"].type.bits == width + + +def test_time_assignment_compiles_for_configured_context( + context_dictionary, single_backend +): + body = text_to_ast( + 't: Fw.Time = time("2023-11-14T22:13:20.5Z", timeContext=9)\n' + "assert t.timeContext == 9\n" + ) + state = analyze_ast(body, get_base_compile_state(context_dictionary(16))) + if single_backend == FPYBC: + analysis_to_fpybc_directives(state) + else: + analysis_to_wasm(state) + + +@pytest.mark.parametrize("width", [8, 16, 32, 64]) +def test_time_context_range_boundaries(context_dictionary, width): + dictionary = context_dictionary(width) + maximum = (1 << width) - 1 + assert _time_value(dictionary, maximum).val["timeContext"].val == maximum + for invalid in (-1, maximum + 1): + with pytest.raises(CompileError, match=f"out of range for type U{width}"): + _time_value(dictionary, invalid) + + +@pytest.mark.parametrize( + "widths", [(8, 16, 8), (16, 32, 16), (64, 8, 64), (16, None, 16)] +) +def test_time_context_reapplied_on_scope_cache_hit(context_dictionary, widths): + _build_global_scopes.cache_clear() + for index, width in enumerate(widths): + dictionary = context_dictionary(width) + before = _build_global_scopes.cache_info() + value = _time_value(dictionary, None) + if index == 2: + assert _build_global_scopes.cache_info().hits == before.hits + 1 + configured = load_dictionary(dictionary)["type_defs"].get( + "FwTimeContextStoreType", U8 + ) + assert TIME.members[1].type == configured + assert TIME_MACRO.args[2][1] == configured + assert TIME_MACRO.args[2][2].type == configured + assert TIME.member_defaults["timeContext"].type == configured + assert value.val["timeContext"].type == configured + assert len(value.serialize()) == value.type.max_size + body = text_to_ast("Fw.TimeValue()\n") + state = analyze_ast(body, get_base_compile_state(dictionary)) + default = state.const_expr_values[body.stmts[0]] + assert default.val["timeContext"].type == configured + assert len(default.serialize()) == default.type.max_size + + +@pytest.mark.parametrize( + "context", [None, 9, "maximum"], ids=["default", "nine", "maximum"] +) +def test_time_context_runs_on_sequencer(fprime_test_api, single_backend, context): + configured = load_dictionary(default_dictionary)["type_defs"][ + "FwTimeContextStoreType" + ] + if context == "maximum": + context = (1 << configured.bits) - 1 + suffix = "" if context is None else f", timeContext={context}" + expected = 0 if context is None else context + seq = ( + f't: Fw.Time = time("2023-11-14T22:13:20.5Z"{suffix})\n' + f"assert t.timeContext == {expected}\n" + "assert t.timeBase == TimeBase.TB_WORKSTATION_TIME\n" + "assert t.seconds == 1700000000\n" + "assert t.useconds == 500000\n" + "assert (t + Fw.TimeIntervalValue(1, 0)).timeContext == t.timeContext\n" + ) + assert_run_success(fprime_test_api, seq) From cabb756914c23f345345a9bb27dc5fc9b21c20ee Mon Sep 17 00:00:00 2001 From: Zakariya Ghazzol Date: Fri, 23 Oct 2026 02:43:39 -0400 Subject: [PATCH 2/2] Simplify time-context tests and isolate cached state --- src/fpy/state.py | 13 +-- test/fpy/test_time_context_config.py | 160 +++++++++++++-------------- 2 files changed, 78 insertions(+), 95 deletions(-) diff --git a/src/fpy/state.py b/src/fpy/state.py index 44b886c..66b8ab5 100644 --- a/src/fpy/state.py +++ b/src/fpy/state.py @@ -59,7 +59,6 @@ TIME_COMPARISON, TIME_INTERVAL, TLM_VALID, - U8, FpyType, FpyValue, TypeKind, @@ -424,8 +423,10 @@ def _update_seq_args_from_dict(dict_type_name_dict: dict[str, FpyType]) -> None: def _update_time_context_type_from_dict( dict_type_name_dict: dict[str, FpyType], ) -> None: - """Configure the time member and builtin argument, defaulting to U8.""" - ctx_type = dict_type_name_dict.get("FwTimeContextStoreType", U8) + """Configure the time member and builtin argument from the dictionary.""" + if "FwTimeContextStoreType" not in dict_type_name_dict: + return # Keep the default U8 + ctx_type = dict_type_name_dict["FwTimeContextStoreType"] if not ctx_type.is_primitive: raise DictionaryError( "FwTimeContextStoreType", @@ -433,9 +434,6 @@ def _update_time_context_type_from_dict( f"it as {ctx_type}.", ) TIME.members[1].type = ctx_type - if TIME.member_defaults is not None: - default = TIME.member_defaults["timeContext"] - TIME.member_defaults["timeContext"] = FpyValue(ctx_type, default.val) TIME_MACRO.args[2] = ("timeContext", ctx_type, FpyValue(ctx_type, 0)) @@ -704,9 +702,6 @@ def get_base_compile_state( """return the initial state of the compiler, based on the given dict path""" if default_time_base is None: default_time_base = DEFAULT_TIME_BASE - # TIME and time() are shared by cached scopes. Restore this dictionary's - # context type even when _build_global_scopes returns a cached entry. - _update_time_context_type_from_dict(load_dictionary(dictionary)["type_defs"]) type_scope, callable_scope, values_scope, type_defs, type_ctors = ( _build_global_scopes(dictionary, default_time_base) ) diff --git a/test/fpy/test_time_context_config.py b/test/fpy/test_time_context_config.py index 4188afd..694afaa 100644 --- a/test/fpy/test_time_context_config.py +++ b/test/fpy/test_time_context_config.py @@ -16,46 +16,52 @@ from fpy.macros import TIME_MACRO from fpy.state import _build_global_scopes, get_base_compile_state from fpy.test_helpers import FPYBC, assert_run_success, default_dictionary -from fpy.types import TIME, U8 - - -@pytest.fixture -def context_dictionary(tmp_path): - paths = {} - - def make(width): - if width in paths: - return paths[width] - data = json.loads(Path(default_dictionary).read_text(encoding="utf-8")) - integer = { - "name": f"U{width or 8}", - "kind": "integer", - "size": width or 8, - "signed": False, - } - for definition in data["typeDefinitions"]: - if definition["qualifiedName"] == "FwTimeContextStoreType": - definition["type"] = integer - definition["underlyingType"] = dict(integer) - break - else: - raise AssertionError("Reference dictionary has no time-context alias") - if width is None: - data["typeDefinitions"].remove(definition) - for definition in data["typeDefinitions"]: - if definition["qualifiedName"] == "Fw.TimeValue": - definition["members"]["timeContext"]["type"] = integer - path = tmp_path / f"time-context-u{width}.json" - path.write_text(json.dumps(data), encoding="utf-8") - paths[width] = str(path) - return paths[width] - - yield make +from fpy.types import TIME + + +@pytest.fixture(autouse=True) +def clear_dictionary_caches(): + """Each test loads one configuration without reusing another test's scopes.""" load_dictionary.cache_clear() _build_global_scopes.cache_clear() + # TIME also keeps derived defaults outside the cached scopes. + TIME.member_defaults = None + yield + load_dictionary.cache_clear() + _build_global_scopes.cache_clear() + TIME.member_defaults = None get_base_compile_state(default_dictionary) +@pytest.fixture( + params=[ + {"name": "U8", "kind": "integer", "size": 8, "signed": False}, + {"name": "U16", "kind": "integer", "size": 16, "signed": False}, + {"name": "U32", "kind": "integer", "size": 32, "signed": False}, + {"name": "U64", "kind": "integer", "size": 64, "signed": False}, + ], + ids=["U8", "U16", "U32", "U64"], +) +def context_dictionary(request, tmp_path): + """Use the reference dictionary with one explicit time-context type.""" + data = json.loads(Path(default_dictionary).read_text(encoding="utf-8")) + data["typeDefinitions"] = [ + definition + for definition in data["typeDefinitions"] + if definition["qualifiedName"] != "FwTimeContextStoreType" + ] + [ + { + "kind": "alias", + "qualifiedName": "FwTimeContextStoreType", + "type": request.param, + "underlyingType": request.param, + } + ] + path = tmp_path / "dictionary.json" + path.write_text(json.dumps(data), encoding="utf-8") + return str(path) + + def _time_value(dictionary, context): suffix = "" if context is None else f", timeContext={context}" body = text_to_ast(f'time("2023-11-14T22:13:20.5Z"{suffix})\n') @@ -64,27 +70,27 @@ def _time_value(dictionary, context): return state.const_expr_values[call] -@pytest.mark.parametrize("width", [8, 16, 32, 64]) -def test_time_builtin_context_parameter_type(context_dictionary, width): - dictionary = context_dictionary(width) - get_base_compile_state(dictionary) - configured = load_dictionary(dictionary)["type_defs"]["FwTimeContextStoreType"] +def test_time_builtin_context_parameter_type(context_dictionary): + get_base_compile_state(context_dictionary) + configured = load_dictionary(context_dictionary)["type_defs"][ + "FwTimeContextStoreType" + ] + assert TIME.members[1].type == configured + assert TIME.member_defaults["timeContext"].type == configured assert TIME_MACRO.args[2][1] == configured assert TIME_MACRO.args[2][2].type == configured -@pytest.mark.parametrize("width", [8, 16, 32, 64]) @pytest.mark.parametrize("context", [None, 9], ids=["default-context", "context-nine"]) -def test_time_constant_context_layout(context_dictionary, width, context): - dictionary = context_dictionary(width) - value = _time_value(dictionary, context) - types = load_dictionary(dictionary)["type_defs"] +def test_time_constant_context_layout(context_dictionary, context): + value = _time_value(context_dictionary, context) + types = load_dictionary(context_dictionary)["type_defs"] configured = types["FwTimeContextStoreType"] expected = ( types["TimeBase"] .enum_dict["TB_WORKSTATION_TIME"] .to_bytes(types["TimeBase"].max_size, "big") - + (0 if context is None else context).to_bytes(width // 8, "big") + + (0 if context is None else context).to_bytes(configured.max_size, "big") + (1_700_000_000).to_bytes(4, "big") + (500_000).to_bytes(4, "big") ) @@ -95,11 +101,17 @@ def test_time_constant_context_layout(context_dictionary, width, context): assert len(serialized) == value.type.max_size -@pytest.mark.parametrize("width", [16, 32, 64]) -def test_wide_time_context_accepts_300(context_dictionary, width): - value = _time_value(context_dictionary(width), 300) - assert value.val["timeContext"].val == 300 - assert value.val["timeContext"].type.bits == width +def test_time_context_300_matches_configured_range(context_dictionary): + configured = load_dictionary(context_dictionary)["type_defs"][ + "FwTimeContextStoreType" + ] + if configured.bits == 8: + with pytest.raises(CompileError, match="out of range for type U8"): + _time_value(context_dictionary, 300) + else: + value = _time_value(context_dictionary, 300) + assert value.val["timeContext"].val == 300 + assert value.val["timeContext"].type == configured def test_time_assignment_compiles_for_configured_context( @@ -109,48 +121,24 @@ def test_time_assignment_compiles_for_configured_context( 't: Fw.Time = time("2023-11-14T22:13:20.5Z", timeContext=9)\n' "assert t.timeContext == 9\n" ) - state = analyze_ast(body, get_base_compile_state(context_dictionary(16))) + state = analyze_ast(body, get_base_compile_state(context_dictionary)) if single_backend == FPYBC: analysis_to_fpybc_directives(state) else: analysis_to_wasm(state) -@pytest.mark.parametrize("width", [8, 16, 32, 64]) -def test_time_context_range_boundaries(context_dictionary, width): - dictionary = context_dictionary(width) - maximum = (1 << width) - 1 - assert _time_value(dictionary, maximum).val["timeContext"].val == maximum +def test_time_context_range_boundaries(context_dictionary): + configured = load_dictionary(context_dictionary)["type_defs"][ + "FwTimeContextStoreType" + ] + maximum = (1 << configured.bits) - 1 + assert _time_value(context_dictionary, maximum).val["timeContext"].val == maximum for invalid in (-1, maximum + 1): - with pytest.raises(CompileError, match=f"out of range for type U{width}"): - _time_value(dictionary, invalid) - - -@pytest.mark.parametrize( - "widths", [(8, 16, 8), (16, 32, 16), (64, 8, 64), (16, None, 16)] -) -def test_time_context_reapplied_on_scope_cache_hit(context_dictionary, widths): - _build_global_scopes.cache_clear() - for index, width in enumerate(widths): - dictionary = context_dictionary(width) - before = _build_global_scopes.cache_info() - value = _time_value(dictionary, None) - if index == 2: - assert _build_global_scopes.cache_info().hits == before.hits + 1 - configured = load_dictionary(dictionary)["type_defs"].get( - "FwTimeContextStoreType", U8 - ) - assert TIME.members[1].type == configured - assert TIME_MACRO.args[2][1] == configured - assert TIME_MACRO.args[2][2].type == configured - assert TIME.member_defaults["timeContext"].type == configured - assert value.val["timeContext"].type == configured - assert len(value.serialize()) == value.type.max_size - body = text_to_ast("Fw.TimeValue()\n") - state = analyze_ast(body, get_base_compile_state(dictionary)) - default = state.const_expr_values[body.stmts[0]] - assert default.val["timeContext"].type == configured - assert len(default.serialize()) == default.type.max_size + with pytest.raises( + CompileError, match=f"out of range for type {configured.name}" + ): + _time_value(context_dictionary, invalid) @pytest.mark.parametrize(