From e3657360accbdaaec3302e895e067bee669bd565 Mon Sep 17 00:00:00 2001 From: zimri-leisher Date: Tue, 10 Nov 2026 11:02:33 -0500 Subject: [PATCH 1/2] Add fprime-fpy-info sequence info tool --- README.md | 23 ++++ pyproject.toml | 1 + src/fpy/main.py | 249 +++++++++++++++++++++++++++++++----------- test/fpy/test_main.py | 89 +++++++++++++++ 4 files changed, 296 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index 0a275377..90a0d1e1 100644 --- a/README.md +++ b/README.md @@ -730,6 +730,29 @@ Some useful compiler flags are: Compiles a single line of Fpy source (one command with constant arguments) and emits it, e.g. `fprime-fpy-cmd 'Ref.seqDisp.RUN_ARGS("seq.bin", NO_WAIT)' -d dict.json`. The `--emit` flag selects how: `zmq` (uplink to a running GDS over ZMQ, the default; address set by `--zmq-addr`), `tcp` (uplink to the GDS TCP server; address set by `--tcp-addr host:port`), or `stdout` (write the binary command packet to stdout). +### `fprime-fpy-info` + +Describes a sequence: its arguments, the sequences it calls, the files it imports, and the commands, telemetry channels and parameters it uses. Only code the sequence can execute counts, so a command in a function that is never called is not listed. + +``` +$ fprime-fpy-info example.fpy -d TopologyDictionary.json +Arguments: + foo: U32 + bar: bool +Called sequences: + (none) +Imports: + /home/user/seqs/helper_lib.fpy +Commands: + CdhCore.cmdDisp.CMD_NO_OP +Telemetry channels: + CdhCore.cmdDisp.CommandsDispatched +Parameters: + (none) +``` + +It takes the same `--seq-map`, `--imports` and `--time-base` flags as `fprime-fpyc`. + ### `fprime-fpy-asm` `fprime-fpy-asm` assembles human-readable `.fpybc` bytecode files into binary `.bin` files. diff --git a/pyproject.toml b/pyproject.toml index 620e7b54..44ba43ad 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -65,6 +65,7 @@ fprime-fpyc = "fpy.main:compile_main" fprime-fpy-asm = "fpy.main:assemble_main" fprime-fpy-disasm = "fpy.main:disassemble_main" fprime-fpy-cmd = "fpy.main:cmd_main" +fprime-fpy-info = "fpy.main:info_main" #### # The test suite lives in test/fpy. The fprime submodule and the harness diff --git a/src/fpy/main.py b/src/fpy/main.py index b5df1bf8..4210129f 100644 --- a/src/fpy/main.py +++ b/src/fpy/main.py @@ -30,9 +30,12 @@ text_to_ast, analysis_to_fpybc_directives, ) -from fpy.state import get_base_compile_state -from fpy.types import DEFAULT_TIME_BASE -from fpy.error import parse_warning_set +from fpy.state import CompileState, get_base_compile_state +from fpy.symbols import CommandSymbol, Symbol +from fpy.syntax import AstDef, AstFuncCall, AstReference +from fpy.types import DEFAULT_TIME_BASE, ChDef, CmdDef, PrmDef +from fpy.error import WarningType, parse_warning_set +from fpy.visitors import STOP_DESCENT, TopDownVisitor def human_readable_size(size_bytes): @@ -111,6 +114,87 @@ def _add_seq_map_argument(arg_parser: argparse.ArgumentParser): ) +def _add_imports_argument(arg_parser: argparse.ArgumentParser): + arg_parser.add_argument( + "-i", + "--imports", + type=Path, + action="append", + default=[], + metavar="DIR", + dest="imports", + help=( + "Directory to search when resolving absolute `import` statements " + "(repeatable). An import that resolves in more than one --imports " + "directory is ambiguous and an error." + ), + ) + + +def analyze_input( + parsed_args: argparse.Namespace, + ignored_warnings: set[WarningType] | None = None, + error_warnings: set[WarningType] | None = None, +) -> CompileState: + """Analyze the input .fpy file named by the parsed arguments. + + Prints the error and exits if the file cannot be analyzed.""" + try: + seq_maps = parse_seq_maps(parsed_args.seq_map) + except ValueError as e: + print(e, file=sys.stderr) + sys.exit(1) + + if not parsed_args.input.exists(): + print(f"Input file {parsed_args.input} does not exist") + sys.exit(1) + fpy.error.file_name = str(parsed_args.input) + + # The import directories are the -i/--imports directories, in order; the + # input file's own directory anchors its relative imports but is not + # among them. Exact duplicate directories are dropped: a repeated -i + # flag carries no information. + import_directories = list( + dict.fromkeys(str(d.resolve()) for d in parsed_args.imports) + ) + + # reading dictionary + try: + state = get_base_compile_state( + str(parsed_args.dictionary.resolve()), + seq_maps, + ignored_warnings=ignored_warnings, + error_warnings=error_warnings, + import_directories=import_directories, + main_file_dir=str(parsed_args.input.parent.resolve()), + main_file_path=str(parsed_args.input.resolve()), + default_time_base=parsed_args.time_base, + ) + except fpy.error.DictionaryError as e: + print(e, file=sys.stderr) + sys.exit(1) + + # syntax + try: + body = text_to_ast(parsed_args.input.read_text(encoding="utf-8")) + except RecursionError: + print("Recursion limit exceeded in parsing", file=sys.stderr) + sys.exit(1) + except fpy.error.CompileError as e: + print(e, file=sys.stderr) + sys.exit(1) + + # semantics + try: + return analyze_ast(body, state) + except RecursionError: + print("Recursion limit exceeded in semantics passes", file=sys.stderr) + sys.exit(1) + except fpy.error.CompileError as e: + print(e, file=sys.stderr) + sys.exit(1) + + def compile_main(args: list[str] = None): backend_version = get_backend_version_str() compiler_version = get_version_str() @@ -155,20 +239,7 @@ def compile_main(args: list[str] = None): ) _add_seq_map_argument(arg_parser) _add_time_base_argument(arg_parser) - arg_parser.add_argument( - "-i", - "--imports", - type=Path, - action="append", - default=[], - metavar="DIR", - dest="imports", - help=( - "Directory to search when resolving absolute `import` statements " - "(repeatable). An import that resolves in more than one --imports " - "directory is ambiguous and an error." - ), - ) + _add_imports_argument(arg_parser) arg_parser.add_argument( "--ignore", type=str, @@ -192,7 +263,6 @@ def compile_main(args: list[str] = None): try: ignored_warnings = parse_warning_set(parsed_args.ignore) error_warnings = parse_warning_set(parsed_args.error) - seq_maps = parse_seq_maps(parsed_args.seq_map) except ValueError as e: print(e, file=sys.stderr) sys.exit(1) @@ -205,54 +275,7 @@ def compile_main(args: list[str] = None): ) sys.exit(1) - if not parsed_args.input.exists(): - print(f"Input file {parsed_args.input} does not exist") - sys.exit(1) - fpy.error.file_name = str(parsed_args.input) - - # The import directories are the -i/--imports directories, in order; the - # input file's own directory anchors its relative imports but is not - # among them. Exact duplicate directories are dropped: a repeated -i - # flag carries no information. - import_directories = list( - dict.fromkeys(str(d.resolve()) for d in parsed_args.imports) - ) - - # reading dictionary - try: - state = get_base_compile_state( - str(parsed_args.dictionary.resolve()), - seq_maps, - ignored_warnings=ignored_warnings, - error_warnings=error_warnings, - import_directories=import_directories, - main_file_dir=str(parsed_args.input.parent.resolve()), - main_file_path=str(parsed_args.input.resolve()), - default_time_base=parsed_args.time_base, - ) - except fpy.error.DictionaryError as e: - print(e, file=sys.stderr) - sys.exit(1) - - # syntax - try: - body = text_to_ast(parsed_args.input.read_text(encoding="utf-8")) - except RecursionError: - print("Recursion limit exceeded in parsing", file=sys.stderr) - sys.exit(1) - except fpy.error.CompileError as e: - print(e, file=sys.stderr) - sys.exit(1) - - # semantics - try: - state = analyze_ast(body, state) - except RecursionError: - print("Recursion limit exceeded in semantics passes", file=sys.stderr) - sys.exit(1) - except fpy.error.CompileError as e: - print(e, file=sys.stderr) - sys.exit(1) + state = analyze_input(parsed_args, ignored_warnings, error_warnings) # codegen try: @@ -315,6 +338,100 @@ def compile_main(args: list[str] = None): assert False, parsed_args.emit +def runs_sequence(cmd: CmdDef) -> bool: + """Whether the command runs a sequence file: its first two arguments are + a file name and a Svc.BlockState.""" + arg_types = [arg_type for _, _, arg_type in cmd.args] + return ( + len(arg_types) >= 2 + and arg_types[0].is_string + and arg_types[1].name == "Svc.BlockState" + ) + + +class FindUses(TopDownVisitor): + """Collects what the code a sequence can execute uses: its top-level code + and the bodies of the functions it can call.""" + + def __init__(self): + super().__init__() + self.symbols: list[Symbol] = [] + """the symbol of each reference""" + self.called_sequences: list[str] = [] + """the file name of each sequence run""" + + def visit_AstDef(self, node: AstDef, state: CompileState): + if node not in state.used_funcs: + return STOP_DESCENT + + def visit_AstReference(self, node: AstReference, state: CompileState): + symbol = state.resolved_symbols.get(node) + if symbol is not None: + self.symbols.append(symbol) + + def visit_AstFuncCall(self, node: AstFuncCall, state: CompileState): + func = state.resolved_symbols.get(node.func) + if isinstance(func, CommandSymbol) and runs_sequence(func.cmd): + file_name = state.const_expr_values[state.resolved_args[node][0]] + self.called_sequences.append(file_name.val) + + +def seq_info(state: CompileState) -> dict[str, list[str]]: + """Describe an analyzed sequence as titled sections of lines, in display + order.""" + uses = FindUses() + uses.run(state.root_block, state) + + def names(symbol_type: type) -> list[str]: + return sorted( + {symbol.name for symbol in uses.symbols if isinstance(symbol, symbol_type)} + ) + + return { + "Arguments": [ + f"{name}: {arg_type.name}" for name, arg_type in state.this_seq_arg_specs + ], + "Called sequences": sorted(set(uses.called_sequences)), + "Imports": sorted( + file_path + for file_path, sequence in state.loaded_sequences.items() + if sequence is not state.main_sequence + ), + "Commands": names(CommandSymbol), + "Telemetry channels": names(ChDef), + "Parameters": names(PrmDef), + } + + +def info_main(args: list[str] = None): + arg_parser = argparse.ArgumentParser( + description=f"Describe an Fpy sequence {get_version_str()}" + ) + arg_parser.add_argument( + "--version", action="version", version=f"%(prog)s {get_version_str()}" + ) + arg_parser.add_argument("input", type=Path, help="The input .fpy file") + arg_parser.add_argument( + "-d", + "--dictionary", + type=Path, + required=True, + help="The FPrime dictionary .json file", + ) + _add_seq_map_argument(arg_parser) + _add_time_base_argument(arg_parser) + _add_imports_argument(arg_parser) + if args is not None: + parsed_args = arg_parser.parse_args(args) + else: + parsed_args = arg_parser.parse_args() + + for title, lines in seq_info(analyze_input(parsed_args)).items(): + print(f"{title}:") + for line in lines or ["(none)"]: + print(f" {line}") + + def assemble_main(args: list[str] = None): arg_parser = argparse.ArgumentParser( description=f"Fpy assembler {get_version_str()}" diff --git a/test/fpy/test_main.py b/test/fpy/test_main.py index 5dbe6332..21aa11a5 100644 --- a/test/fpy/test_main.py +++ b/test/fpy/test_main.py @@ -5,6 +5,7 @@ from fpy import main as fpy_main from fpy.bytecode.directives import ConstCmdDirective import fpy.error as fpy_error +from fpy.test_helpers import default_dictionary def fake_compile_state(**kwargs): @@ -787,3 +788,91 @@ def test_cmd_main_invalid_tcp_addr(monkeypatch, capsys): assert exc.value.code == 1 assert "Invalid --tcp-addr format" in capsys.readouterr().err + + +def test_info_main(tmp_path, capsys): + """The sections cover the code the sequence can execute: its top-level + code and the functions it can call, wherever they are defined.""" + (tmp_path / "lib.fpy").write_text( + "def used():\n" + " CdhCore.cmdDisp.CMD_NO_OP()\n" + "\n" + "def unused():\n" + " CdhCore.cmdDisp.CMD_CLEAR_TRACKING()\n" + ) + (tmp_path / "child.fpy").write_text("sequence(foo: U32, bar: bool)\n") + input_path = tmp_path / "seq.fpy" + input_path.write_text( + "sequence(count: U32, mode: Ref.Choice)\n" + "from .lib import used\n" + "\n" + "def unused():\n" + " CdhCore.cmdDisp.CMD_TEST_CMD_1(1, 2.0, 3)\n" + "\n" + "used()\n" + "if CdhCore.cmdDisp.CommandsDispatched > count:\n" + ' Ref.seqDisp.RUN_ARGS("child.bin", Svc.BlockState.BLOCK, 1, bar=True)\n' + 'Ref.cmdSeq0.RUN("other.bin", Svc.BlockState.NO_BLOCK)\n' + "if Ref.sendBuffComp.parameter3 > 0:\n" + ' Ref.cmdSeq0.RUN("other.bin", Svc.BlockState.BLOCK)\n' + ) + + fpy_main.info_main( + [str(input_path), "-d", default_dictionary, "--seq-map", f"={tmp_path}/"] + ) + + assert capsys.readouterr().out == ( + "Arguments:\n" + " count: U32\n" + " mode: Ref.Choice\n" + "Called sequences:\n" + " child.bin\n" + " other.bin\n" + "Imports:\n" + f" {(tmp_path / 'lib.fpy').resolve()}\n" + "Commands:\n" + " CdhCore.cmdDisp.CMD_NO_OP\n" + " Ref.cmdSeq0.RUN\n" + " Ref.seqDisp.RUN_ARGS\n" + "Telemetry channels:\n" + " CdhCore.cmdDisp.CommandsDispatched\n" + "Parameters:\n" + " Ref.sendBuffComp.parameter3\n" + ) + + +def test_info_main_empty_sections(tmp_path, capsys): + """A section with nothing to list says so.""" + input_path = tmp_path / "seq.fpy" + input_path.write_text("pass\n") + + fpy_main.info_main([str(input_path), "-d", default_dictionary]) + + assert capsys.readouterr().out == ( + "Arguments:\n" + " (none)\n" + "Called sequences:\n" + " (none)\n" + "Imports:\n" + " (none)\n" + "Commands:\n" + " (none)\n" + "Telemetry channels:\n" + " (none)\n" + "Parameters:\n" + " (none)\n" + ) + + +def test_info_main_compile_error(tmp_path, capsys): + """A sequence that does not compile is reported to stderr and exits 1.""" + input_path = tmp_path / "seq.fpy" + input_path.write_text("not_defined()\n") + + with pytest.raises(SystemExit) as exc: + fpy_main.info_main([str(input_path), "-d", default_dictionary]) + + assert exc.value.code == 1 + captured = capsys.readouterr() + assert captured.out == "" + assert "not_defined" in captured.err From 8d7c8c8db408faa6224b43a341d2682aaff0a4df Mon Sep 17 00:00:00 2001 From: zimri-leisher Date: Tue, 10 Nov 2026 11:38:52 -0500 Subject: [PATCH 2/2] Output JSON from fprime-fpy-info --- README.md | 34 +++++++++++++--------- src/fpy/main.py | 28 +++++++++--------- test/fpy/test_main.py | 66 +++++++++++++++++++------------------------ 3 files changed, 62 insertions(+), 66 deletions(-) diff --git a/README.md b/README.md index 90a0d1e1..38ee791d 100644 --- a/README.md +++ b/README.md @@ -732,23 +732,29 @@ Compiles a single line of Fpy source (one command with constant arguments) and e ### `fprime-fpy-info` -Describes a sequence: its arguments, the sequences it calls, the files it imports, and the commands, telemetry channels and parameters it uses. Only code the sequence can execute counts, so a command in a function that is never called is not listed. +Describes a sequence as JSON: its arguments, the sequences it calls, the files it imports, and the commands, telemetry channels and parameters it uses. Only code the sequence can execute counts, so a command in a function that is never called is not listed. ``` $ fprime-fpy-info example.fpy -d TopologyDictionary.json -Arguments: - foo: U32 - bar: bool -Called sequences: - (none) -Imports: - /home/user/seqs/helper_lib.fpy -Commands: - CdhCore.cmdDisp.CMD_NO_OP -Telemetry channels: - CdhCore.cmdDisp.CommandsDispatched -Parameters: - (none) +{ + "arguments": [ + { + "name": "foo", + "type": "U32" + } + ], + "calledSequences": [], + "imports": [ + "/home/user/seqs/helper_lib.fpy" + ], + "commands": [ + "CdhCore.cmdDisp.CMD_NO_OP" + ], + "telemetryChannels": [ + "CdhCore.cmdDisp.CommandsDispatched" + ], + "parameters": [] +} ``` It takes the same `--seq-map`, `--imports` and `--time-base` flags as `fprime-fpyc`. diff --git a/src/fpy/main.py b/src/fpy/main.py index 4210129f..5c1e5962 100644 --- a/src/fpy/main.py +++ b/src/fpy/main.py @@ -3,6 +3,7 @@ import argparse import getpass from importlib.metadata import version +import json from pathlib import Path import socket import struct @@ -376,9 +377,8 @@ def visit_AstFuncCall(self, node: AstFuncCall, state: CompileState): self.called_sequences.append(file_name.val) -def seq_info(state: CompileState) -> dict[str, list[str]]: - """Describe an analyzed sequence as titled sections of lines, in display - order.""" +def seq_info(state: CompileState) -> dict[str, list]: + """Describe an analyzed sequence as a JSON object.""" uses = FindUses() uses.run(state.root_block, state) @@ -388,24 +388,25 @@ def names(symbol_type: type) -> list[str]: ) return { - "Arguments": [ - f"{name}: {arg_type.name}" for name, arg_type in state.this_seq_arg_specs + "arguments": [ + {"name": name, "type": arg_type.name} + for name, arg_type in state.this_seq_arg_specs ], - "Called sequences": sorted(set(uses.called_sequences)), - "Imports": sorted( + "calledSequences": sorted(set(uses.called_sequences)), + "imports": sorted( file_path for file_path, sequence in state.loaded_sequences.items() if sequence is not state.main_sequence ), - "Commands": names(CommandSymbol), - "Telemetry channels": names(ChDef), - "Parameters": names(PrmDef), + "commands": names(CommandSymbol), + "telemetryChannels": names(ChDef), + "parameters": names(PrmDef), } def info_main(args: list[str] = None): arg_parser = argparse.ArgumentParser( - description=f"Describe an Fpy sequence {get_version_str()}" + description=f"Describe an Fpy sequence as JSON {get_version_str()}" ) arg_parser.add_argument( "--version", action="version", version=f"%(prog)s {get_version_str()}" @@ -426,10 +427,7 @@ def info_main(args: list[str] = None): else: parsed_args = arg_parser.parse_args() - for title, lines in seq_info(analyze_input(parsed_args)).items(): - print(f"{title}:") - for line in lines or ["(none)"]: - print(f" {line}") + print(json.dumps(seq_info(analyze_input(parsed_args)), indent=2)) def assemble_main(args: list[str] = None): diff --git a/test/fpy/test_main.py b/test/fpy/test_main.py index 21aa11a5..80f96525 100644 --- a/test/fpy/test_main.py +++ b/test/fpy/test_main.py @@ -1,3 +1,4 @@ +import json from types import SimpleNamespace import pytest @@ -791,7 +792,7 @@ def test_cmd_main_invalid_tcp_addr(monkeypatch, capsys): def test_info_main(tmp_path, capsys): - """The sections cover the code the sequence can execute: its top-level + """The output covers the code the sequence can execute: its top-level code and the functions it can call, wherever they are defined.""" (tmp_path / "lib.fpy").write_text( "def used():\n" @@ -821,47 +822,38 @@ def test_info_main(tmp_path, capsys): [str(input_path), "-d", default_dictionary, "--seq-map", f"={tmp_path}/"] ) - assert capsys.readouterr().out == ( - "Arguments:\n" - " count: U32\n" - " mode: Ref.Choice\n" - "Called sequences:\n" - " child.bin\n" - " other.bin\n" - "Imports:\n" - f" {(tmp_path / 'lib.fpy').resolve()}\n" - "Commands:\n" - " CdhCore.cmdDisp.CMD_NO_OP\n" - " Ref.cmdSeq0.RUN\n" - " Ref.seqDisp.RUN_ARGS\n" - "Telemetry channels:\n" - " CdhCore.cmdDisp.CommandsDispatched\n" - "Parameters:\n" - " Ref.sendBuffComp.parameter3\n" - ) - - -def test_info_main_empty_sections(tmp_path, capsys): - """A section with nothing to list says so.""" + assert json.loads(capsys.readouterr().out) == { + "arguments": [ + {"name": "count", "type": "U32"}, + {"name": "mode", "type": "Ref.Choice"}, + ], + "calledSequences": ["child.bin", "other.bin"], + "imports": [str((tmp_path / "lib.fpy").resolve())], + "commands": [ + "CdhCore.cmdDisp.CMD_NO_OP", + "Ref.cmdSeq0.RUN", + "Ref.seqDisp.RUN_ARGS", + ], + "telemetryChannels": ["CdhCore.cmdDisp.CommandsDispatched"], + "parameters": ["Ref.sendBuffComp.parameter3"], + } + + +def test_info_main_empty(tmp_path, capsys): + """A sequence that uses nothing has every member, each empty.""" input_path = tmp_path / "seq.fpy" input_path.write_text("pass\n") fpy_main.info_main([str(input_path), "-d", default_dictionary]) - assert capsys.readouterr().out == ( - "Arguments:\n" - " (none)\n" - "Called sequences:\n" - " (none)\n" - "Imports:\n" - " (none)\n" - "Commands:\n" - " (none)\n" - "Telemetry channels:\n" - " (none)\n" - "Parameters:\n" - " (none)\n" - ) + assert json.loads(capsys.readouterr().out) == { + "arguments": [], + "calledSequences": [], + "imports": [], + "commands": [], + "telemetryChannels": [], + "parameters": [], + } def test_info_main_compile_error(tmp_path, capsys):