From 914595287860bcafb70535e5e48cd0703009633e Mon Sep 17 00:00:00 2001 From: xXenvy Date: Sat, 19 Jul 2025 02:21:01 +0200 Subject: [PATCH 1/6] feat(parsing): Implementation of FilterableParser --- .../premade/components/parser/parser.py | 104 ++++++++++++------ .../premade/components/parser/test_parser.py | 30 ++++- 2 files changed, 96 insertions(+), 38 deletions(-) diff --git a/src/devana/preprocessing/premade/components/parser/parser.py b/src/devana/preprocessing/premade/components/parser/parser.py index 5f0023e..a7d995f 100644 --- a/src/devana/preprocessing/premade/components/parser/parser.py +++ b/src/devana/preprocessing/premade/components/parser/parser.py @@ -1,10 +1,10 @@ -from typing import List, Dict, Type +from typing import List, Dict, Type, Callable, Optional import typing from enum import Enum from devana.preprocessing.premade.components.parser.extractor import IExtractor from devana.preprocessing.premade.components.parser.argumentsparser import (ArgumentsParser, IParsable, ArgumentGenericTypeParser) -from devana.preprocessing.premade.components.parser.functionparser import FunctionParser +from devana.preprocessing.premade.components.parser.functionparser import FunctionParser, FunctionEntity from devana.preprocessing.premade.components.parser.typechecker import is_arguments_valid from devana.preprocessing.premade.components.executor.executable import CallFrame, Signature from devana.preprocessing.preprocessor import ISource @@ -45,46 +45,78 @@ def _find_enum(cls, hint) -> List[Type[Enum]]: result += cls._find_enum(arg) return result + def _create_call_frame(self, function: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: + arguments = self._arguments_parser.tokenize(function.arguments) + positional_arguments: List[CallFrame.Arguments.Value] = [] + named_arguments: Dict[str, CallFrame.Arguments.Value] = {} + + # we need to search named arguments and unnamed to create the right arguments entry + for argument in arguments: + if isinstance(argument, dict): + if len(argument.keys()) != 0: + raise ValueError("Internal error. Argument parser provide too many keys in dictionary.") + key = list(argument.keys())[0] + if not isinstance(key, str): + raise ValueError("Internal error. Argument parser provide wrong dictionary key type.") + named_arguments[key] = CallFrame.Arguments.Value(argument[key]) + else: + positional_arguments.append(CallFrame.Arguments.Value(argument)) + + # find proper signature for this function. + matched: List[Signature] = list(filter( + lambda s: s.name == function.name and s.namespaces == function.namespaces, + self._signatures + )) + if len(matched) > 1: + raise ValueError(f"Duplicated signatures found for function: {function.namespaces}::{function.name}") + if len(matched) == 0: + raise ValueError(f"Cannot find signature for function: {function.namespaces}::{function.name}") + signature: Signature = matched[0] + + arguments_fame = CallFrame.Arguments(positional_arguments, named_arguments) + if not is_arguments_valid(arguments_fame, signature.arguments): + raise ValueError(f"Cannot match arguments for signature: {function.namespaces}::{function.name}") + return Environment.CallingData(arguments_fame, parent, signature) + @classmethod def get_produced_type(cls) -> Type: return Environment.CallingData + def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: + result = [] + function_parser = FunctionParser() + for data in self._extractor.extract(): + functions = function_parser.parse(data.text) + for function in functions: + result.append(self._create_call_frame(function, data.parent)) + return result + + +class FilterableParser(Parser): + """Extends Parser by invoking a user-provided callback for each parsed function entity. + + For each parsed function entity, the callback is invoked with: + - function: the FunctionEntity instance. + - parent: the ExtractedFunction parent (context for creating CallingData). + + The callback should return: + - None to use the default call-frame logic. + - A CallingData to assign a custom invocation (signature, arguments) to the function.""" + def __init__( + self, + extractor: IExtractor, + signatures: List[Signature], + call_data_callback: Callable[[FunctionEntity, ISyntaxElement], Optional[Environment.CallingData[ISyntaxElement]]] + ): + super().__init__(extractor, signatures) + self._call_data_callback = call_data_callback def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: result = [] - text_datas = self._extractor.extract() - for data in text_datas: - function_parsr = FunctionParser() - functions = function_parsr.parse(data.text) + function_parser = FunctionParser() + for data in self._extractor.extract(): + functions = function_parser.parse(data.text) for function in functions: - arguments = self._arguments_parser.tokenize(function.arguments) - positional_arguments: List[CallFrame.Arguments.Value] = [] - named_arguments: Dict[str, CallFrame.Arguments.Value] = {} - - # we need to search named arguments and unnamed to create the right arguments entry - for argument in arguments: - if isinstance(argument, dict): - if len(argument.keys()) != 0: - raise ValueError("Internal error. Argument parser provide too many keys in dictionary.") - key = list(argument.keys())[0] - if not isinstance(key, str): - raise ValueError("Internal error. Argument parser provide wrong dictionary key type.") - named_arguments[key] = CallFrame.Arguments.Value(argument[key]) - else: - positional_arguments.append(CallFrame.Arguments.Value(argument)) - - # now find match signature - match_signatures: List[Signature] = [signature for signature in self._signatures - if signature.name == function.name and signature.namespaces == function.namespaces] - if len(match_signatures) > 1: - raise ValueError(f"Duplicated signatures found for function: " - f"{function.namespaces}::{function.name}") - if len(match_signatures) == 0: - raise ValueError(f"Cannot find signature for function: {function.namespaces}::{function.name}") - - arguments_fame = CallFrame.Arguments(positional_arguments, named_arguments) - if not is_arguments_valid(arguments_fame, match_signatures[0].arguments): - raise ValueError(f"Cannot match arguments for signature: {function.namespaces}::{function.name}") - call_frame = Environment.CallingData(arguments_fame, data.parent, match_signatures[0]) + call_frame = self._call_data_callback(function, data.parent) or self._create_call_frame(function, data.parent) result.append(call_frame) - return result + return result \ No newline at end of file diff --git a/tests/preprocessing/unit/premade/components/parser/test_parser.py b/tests/preprocessing/unit/premade/components/parser/test_parser.py index bc925d8..591c827 100644 --- a/tests/preprocessing/unit/premade/components/parser/test_parser.py +++ b/tests/preprocessing/unit/premade/components/parser/test_parser.py @@ -1,8 +1,10 @@ import unittest -from typing import List +from typing import List, Optional from enum import Enum, auto -from devana.preprocessing.premade.components.parser.parser import Parser, Signature +from devana.syntax_abstraction.syntax import ISyntaxElement +from devana.preprocessing.premade.components.parser.parser import Parser, FilterableParser, Signature, FunctionEntity from devana.preprocessing.premade.components.parser.extractor import IExtractor, ExtractedFunction +from devana.preprocessing.premade.components.executor.environment import Environment, CallFrame from devana.syntax_abstraction.externc import ExternC @@ -28,6 +30,30 @@ def test_parser(self): self.assertEqual(result[0].arguments.positional[1].content, 7) self.assertEqual(result[0].arguments.positional[2].content, True) + def test_filterable_parser(self): + + def callback(entity: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: + self.assertEqual(entity.name, 'foo') + self.assertEqual(entity.namespaces, ['text_namespace']) + self.assertEqual(entity.arguments, '"text", 7, true') + self.assertTrue(isinstance(parent, ExternC)) + return Environment.CallingData( + arguments=CallFrame.Arguments([], {}), + target=parent, + signature=Signature("foo", ["text_namespace"]) + ) # only one case so always return the same calling data + + parser = FilterableParser( + self.ExtractorForTestsAll(), + [], + callback + ) + result = parser.feed() + self.assertEqual(1, len(result)) + self.assertEqual(result[0].signature.name, 'foo') + self.assertEqual(result[0].signature.namespaces, ['text_namespace']) + self.assertEqual(len(result[0].arguments.positional), 0) + def test_parsing_with_enum(self): class TestEnum(Enum): From 7a84a05064f8877cc964c6987050c70b14288c53 Mon Sep 17 00:00:00 2001 From: xXenvy Date: Sat, 19 Jul 2025 02:24:28 +0200 Subject: [PATCH 2/6] refactor: rename function --- .../preprocessing/premade/components/parser/parser.py | 10 +++++----- 1 file changed, 5 insertions(+), 5 deletions(-) diff --git a/src/devana/preprocessing/premade/components/parser/parser.py b/src/devana/preprocessing/premade/components/parser/parser.py index a7d995f..1896bb0 100644 --- a/src/devana/preprocessing/premade/components/parser/parser.py +++ b/src/devana/preprocessing/premade/components/parser/parser.py @@ -45,7 +45,7 @@ def _find_enum(cls, hint) -> List[Type[Enum]]: result += cls._find_enum(arg) return result - def _create_call_frame(self, function: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: + def _create_calling_data(self, function: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: arguments = self._arguments_parser.tokenize(function.arguments) positional_arguments: List[CallFrame.Arguments.Value] = [] named_arguments: Dict[str, CallFrame.Arguments.Value] = {} @@ -88,7 +88,7 @@ def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: for data in self._extractor.extract(): functions = function_parser.parse(data.text) for function in functions: - result.append(self._create_call_frame(function, data.parent)) + result.append(self._create_calling_data(function, data.parent)) return result @@ -106,10 +106,10 @@ def __init__( self, extractor: IExtractor, signatures: List[Signature], - call_data_callback: Callable[[FunctionEntity, ISyntaxElement], Optional[Environment.CallingData[ISyntaxElement]]] + calling_data_callback: Callable[[FunctionEntity, ISyntaxElement], Optional[Environment.CallingData[ISyntaxElement]]] ): super().__init__(extractor, signatures) - self._call_data_callback = call_data_callback + self._calling_data_callback = calling_data_callback def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: result = [] @@ -117,6 +117,6 @@ def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: for data in self._extractor.extract(): functions = function_parser.parse(data.text) for function in functions: - call_frame = self._call_data_callback(function, data.parent) or self._create_call_frame(function, data.parent) + call_frame = self._calling_data_callback(function, data.parent) or self._create_calling_data(function, data.parent) result.append(call_frame) return result \ No newline at end of file From f1106ee9a21d2294b029d02b9905c50b22d6b536 Mon Sep 17 00:00:00 2001 From: xXenvy Date: Sat, 19 Jul 2025 02:26:18 +0200 Subject: [PATCH 3/6] fix: pylint warning --- src/devana/preprocessing/premade/components/parser/parser.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/devana/preprocessing/premade/components/parser/parser.py b/src/devana/preprocessing/premade/components/parser/parser.py index 1896bb0..6eec5eb 100644 --- a/src/devana/preprocessing/premade/components/parser/parser.py +++ b/src/devana/preprocessing/premade/components/parser/parser.py @@ -119,4 +119,4 @@ def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: for function in functions: call_frame = self._calling_data_callback(function, data.parent) or self._create_calling_data(function, data.parent) result.append(call_frame) - return result \ No newline at end of file + return result From 85d6bf1dbe56c431d8db665c48d2d00e6c744d4d Mon Sep 17 00:00:00 2001 From: xXenvy Date: Sun, 27 Jul 2025 19:45:21 +0200 Subject: [PATCH 4/6] refactor: data generation directly from SourceModule --- .../premade/components/flow/dataprovider.py | 31 +++++ .../premade/components/flow/sourcemerge.py | 4 +- .../premade/components/parser/parser.py | 107 ++++++------------ .../flow/source_files/elements/elements.hpp | 9 ++ .../components/flow/test_dataprovider.py | 37 ++++++ .../premade/components/parser/test_parser.py | 28 +---- 6 files changed, 117 insertions(+), 99 deletions(-) create mode 100644 src/devana/preprocessing/premade/components/flow/dataprovider.py create mode 100644 tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp create mode 100644 tests/preprocessing/unit/premade/components/flow/test_dataprovider.py diff --git a/src/devana/preprocessing/premade/components/flow/dataprovider.py b/src/devana/preprocessing/premade/components/flow/dataprovider.py new file mode 100644 index 0000000..2505737 --- /dev/null +++ b/src/devana/preprocessing/premade/components/flow/dataprovider.py @@ -0,0 +1,31 @@ +from typing import List, Type, Optional, Callable + +from devana.preprocessing.premade.components.executor.environment import Environment +from devana.syntax_abstraction.organizers.sourcemodule import SourceModule +from devana.syntax_abstraction.syntax import ISyntaxElement +from devana.preprocessing.preprocessor import ISource + +class SyntaxElementDataProvider(ISource): + """Scans syntax elements across provided source modules and produces a list of + Environment.CallingData instances via a custom factory function.""" + + def __init__( + self, + modules: List[SourceModule], + data_factory: Callable[[ISyntaxElement], Optional[Environment.CallingData]] + ): + self._modules = modules + self._data_factory = data_factory + + @classmethod + def get_produced_type(cls) -> Type: + return Environment.CallingData + + def feed(self) -> List[Environment.CallingData]: + result = [] + elements = (content for module in self._modules for file in module.files for content in file.content) + for element in elements: + if data := self._data_factory(element): + assert isinstance(data, Environment.CallingData), "Data is not an instance of Environment.CallingData" + result.append(data) + return result diff --git a/src/devana/preprocessing/premade/components/flow/sourcemerge.py b/src/devana/preprocessing/premade/components/flow/sourcemerge.py index 1d7930f..f9a359f 100644 --- a/src/devana/preprocessing/premade/components/flow/sourcemerge.py +++ b/src/devana/preprocessing/premade/components/flow/sourcemerge.py @@ -1,11 +1,11 @@ -from typing import List, Type, Any, TypeVar +from typing import List, Type, Any from devana.preprocessing.preprocessor import ISource from devana.preprocessing.premade.components.executor.environment import Environment -T = TypeVar('T') class SourceMergeCallingData(ISource): """Merges multiple sources into one.""" + def __init__(self, sources: List[ISource]): if len(sources) == 0: raise ValueError("No sources provided.") diff --git a/src/devana/preprocessing/premade/components/parser/parser.py b/src/devana/preprocessing/premade/components/parser/parser.py index 6eec5eb..bcf6f1a 100644 --- a/src/devana/preprocessing/premade/components/parser/parser.py +++ b/src/devana/preprocessing/premade/components/parser/parser.py @@ -1,15 +1,15 @@ -from typing import List, Dict, Type, Callable, Optional +from typing import List, Dict, Type import typing from enum import Enum from devana.preprocessing.premade.components.parser.extractor import IExtractor from devana.preprocessing.premade.components.parser.argumentsparser import (ArgumentsParser, IParsable, ArgumentGenericTypeParser) -from devana.preprocessing.premade.components.parser.functionparser import FunctionParser, FunctionEntity +from devana.preprocessing.premade.components.parser.functionparser import FunctionParser from devana.preprocessing.premade.components.parser.typechecker import is_arguments_valid from devana.preprocessing.premade.components.executor.executable import CallFrame, Signature -from devana.preprocessing.preprocessor import ISource -from devana.syntax_abstraction.syntax import ISyntaxElement from devana.preprocessing.premade.components.executor.environment import Environment +from devana.syntax_abstraction.syntax import ISyntaxElement +from devana.preprocessing.preprocessor import ISource class Parser(ISource): @@ -45,78 +45,45 @@ def _find_enum(cls, hint) -> List[Type[Enum]]: result += cls._find_enum(arg) return result - def _create_calling_data(self, function: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: - arguments = self._arguments_parser.tokenize(function.arguments) - positional_arguments: List[CallFrame.Arguments.Value] = [] - named_arguments: Dict[str, CallFrame.Arguments.Value] = {} - - # we need to search named arguments and unnamed to create the right arguments entry - for argument in arguments: - if isinstance(argument, dict): - if len(argument.keys()) != 0: - raise ValueError("Internal error. Argument parser provide too many keys in dictionary.") - key = list(argument.keys())[0] - if not isinstance(key, str): - raise ValueError("Internal error. Argument parser provide wrong dictionary key type.") - named_arguments[key] = CallFrame.Arguments.Value(argument[key]) - else: - positional_arguments.append(CallFrame.Arguments.Value(argument)) - - # find proper signature for this function. - matched: List[Signature] = list(filter( - lambda s: s.name == function.name and s.namespaces == function.namespaces, - self._signatures - )) - if len(matched) > 1: - raise ValueError(f"Duplicated signatures found for function: {function.namespaces}::{function.name}") - if len(matched) == 0: - raise ValueError(f"Cannot find signature for function: {function.namespaces}::{function.name}") - signature: Signature = matched[0] - - arguments_fame = CallFrame.Arguments(positional_arguments, named_arguments) - if not is_arguments_valid(arguments_fame, signature.arguments): - raise ValueError(f"Cannot match arguments for signature: {function.namespaces}::{function.name}") - return Environment.CallingData(arguments_fame, parent, signature) - @classmethod def get_produced_type(cls) -> Type: return Environment.CallingData def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: result = [] - function_parser = FunctionParser() - for data in self._extractor.extract(): - functions = function_parser.parse(data.text) - for function in functions: - result.append(self._create_calling_data(function, data.parent)) - return result - - -class FilterableParser(Parser): - """Extends Parser by invoking a user-provided callback for each parsed function entity. - - For each parsed function entity, the callback is invoked with: - - function: the FunctionEntity instance. - - parent: the ExtractedFunction parent (context for creating CallingData). - - The callback should return: - - None to use the default call-frame logic. - - A CallingData to assign a custom invocation (signature, arguments) to the function.""" - def __init__( - self, - extractor: IExtractor, - signatures: List[Signature], - calling_data_callback: Callable[[FunctionEntity, ISyntaxElement], Optional[Environment.CallingData[ISyntaxElement]]] - ): - super().__init__(extractor, signatures) - self._calling_data_callback = calling_data_callback - - def feed(self) -> List[Environment.CallingData[ISyntaxElement]]: - result = [] - function_parser = FunctionParser() - for data in self._extractor.extract(): - functions = function_parser.parse(data.text) + text_datas = self._extractor.extract() + for data in text_datas: + function_parsr = FunctionParser() + functions = function_parsr.parse(data.text) for function in functions: - call_frame = self._calling_data_callback(function, data.parent) or self._create_calling_data(function, data.parent) + arguments = self._arguments_parser.tokenize(function.arguments) + positional_arguments: List[CallFrame.Arguments.Value] = [] + named_arguments: Dict[str, CallFrame.Arguments.Value] = {} + + # we need to search named arguments and unnamed to create the right arguments entry + for argument in arguments: + if isinstance(argument, dict): + if len(argument.keys()) != 0: + raise ValueError("Internal error. Argument parser provide too many keys in dictionary.") + key = list(argument.keys())[0] + if not isinstance(key, str): + raise ValueError("Internal error. Argument parser provide wrong dictionary key type.") + named_arguments[key] = CallFrame.Arguments.Value(argument[key]) + else: + positional_arguments.append(CallFrame.Arguments.Value(argument)) + + # now find match signature + match_signatures: List[Signature] = [signature for signature in self._signatures + if signature.name == function.name and signature.namespaces == function.namespaces] + if len(match_signatures) > 1: + raise ValueError(f"Duplicated signatures found for function: " + f"{function.namespaces}::{function.name}") + if len(match_signatures) == 0: + raise ValueError(f"Cannot find signature for function: {function.namespaces}::{function.name}") + + arguments_fame = CallFrame.Arguments(positional_arguments, named_arguments) + if not is_arguments_valid(arguments_fame, match_signatures[0].arguments): + raise ValueError(f"Cannot match arguments for signature: {function.namespaces}::{function.name}") + call_frame = Environment.CallingData(arguments_fame, data.parent, match_signatures[0]) result.append(call_frame) return result diff --git a/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp b/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp new file mode 100644 index 0000000..1ab6fb4 --- /dev/null +++ b/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp @@ -0,0 +1,9 @@ + + +void p_function(); + +struct Abc { + int a; +}; + +class p_Class; \ No newline at end of file diff --git a/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py new file mode 100644 index 0000000..5f0fce6 --- /dev/null +++ b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py @@ -0,0 +1,37 @@ +import unittest +import os + +from typing import Optional +from devana.preprocessing.premade.components.flow.dataprovider import SyntaxElementDataProvider +from devana.preprocessing.premade.components.executor.environment import Environment, CallFrame +from devana.preprocessing.premade.components.parser.parser import Signature +from devana.syntax_abstraction.organizers.sourcemodule import SourceModule +from devana.syntax_abstraction.syntax import ISyntaxElement + + +def data_factory(element: ISyntaxElement) -> Optional[Environment.CallingData]: + name: Optional[str] = getattr(element, "name", None) + if name and name.startswith("p_"): + return Environment.CallingData( + arguments=CallFrame.Arguments(positional=[], named={}), + target=element, + signature=Signature(name="Parsable") + ) + + +class TestSyntaxElementDataProvider(unittest.TestCase): + + def setUp(self): + self._module = SourceModule("Elements", os.path.dirname(__file__) + r"/source_files/elements") + + def test_syntax_element_data_provider(self): + provider = SyntaxElementDataProvider([self._module], data_factory) + result = provider.feed() + self.assertEqual(len(result), 2) + + for data in result: + self.assertTrue(isinstance(data, Environment.CallingData)) + self.assertEqual(len(data.arguments.positional), 0) + self.assertEqual(len(data.arguments.named), 0) + self.assertEqual(data.signature.name, "Parsable") + self.assertEqual(data.signature.namespaces, []) diff --git a/tests/preprocessing/unit/premade/components/parser/test_parser.py b/tests/preprocessing/unit/premade/components/parser/test_parser.py index 591c827..ded34c7 100644 --- a/tests/preprocessing/unit/premade/components/parser/test_parser.py +++ b/tests/preprocessing/unit/premade/components/parser/test_parser.py @@ -1,10 +1,8 @@ import unittest from typing import List, Optional from enum import Enum, auto -from devana.syntax_abstraction.syntax import ISyntaxElement -from devana.preprocessing.premade.components.parser.parser import Parser, FilterableParser, Signature, FunctionEntity +from devana.preprocessing.premade.components.parser.parser import Parser, Signature from devana.preprocessing.premade.components.parser.extractor import IExtractor, ExtractedFunction -from devana.preprocessing.premade.components.executor.environment import Environment, CallFrame from devana.syntax_abstraction.externc import ExternC @@ -30,30 +28,6 @@ def test_parser(self): self.assertEqual(result[0].arguments.positional[1].content, 7) self.assertEqual(result[0].arguments.positional[2].content, True) - def test_filterable_parser(self): - - def callback(entity: FunctionEntity, parent: ISyntaxElement) -> Environment.CallingData: - self.assertEqual(entity.name, 'foo') - self.assertEqual(entity.namespaces, ['text_namespace']) - self.assertEqual(entity.arguments, '"text", 7, true') - self.assertTrue(isinstance(parent, ExternC)) - return Environment.CallingData( - arguments=CallFrame.Arguments([], {}), - target=parent, - signature=Signature("foo", ["text_namespace"]) - ) # only one case so always return the same calling data - - parser = FilterableParser( - self.ExtractorForTestsAll(), - [], - callback - ) - result = parser.feed() - self.assertEqual(1, len(result)) - self.assertEqual(result[0].signature.name, 'foo') - self.assertEqual(result[0].signature.namespaces, ['text_namespace']) - self.assertEqual(len(result[0].arguments.positional), 0) - def test_parsing_with_enum(self): class TestEnum(Enum): From e83a72adf13c8c2a486fa7bdbd6d2b42ab75211a Mon Sep 17 00:00:00 2001 From: xXenvy Date: Fri, 8 Aug 2025 23:24:48 +0200 Subject: [PATCH 5/6] refactor: allow to pass the list of factory functions --- .../premade/components/flow/dataprovider.py | 13 +++++++------ .../unit/premade/components/flow/__init__.py | 0 .../premade/components/flow/test_dataprovider.py | 2 +- 3 files changed, 8 insertions(+), 7 deletions(-) create mode 100644 tests/preprocessing/unit/premade/components/flow/__init__.py diff --git a/src/devana/preprocessing/premade/components/flow/dataprovider.py b/src/devana/preprocessing/premade/components/flow/dataprovider.py index 2505737..2df79f5 100644 --- a/src/devana/preprocessing/premade/components/flow/dataprovider.py +++ b/src/devana/preprocessing/premade/components/flow/dataprovider.py @@ -7,15 +7,15 @@ class SyntaxElementDataProvider(ISource): """Scans syntax elements across provided source modules and produces a list of - Environment.CallingData instances via a custom factory function.""" + Environment.CallingData instances using custom factory functions.""" def __init__( self, modules: List[SourceModule], - data_factory: Callable[[ISyntaxElement], Optional[Environment.CallingData]] + data_factories: List[Callable[[ISyntaxElement], Optional[Environment.CallingData]]] ): self._modules = modules - self._data_factory = data_factory + self._data_factories = data_factories @classmethod def get_produced_type(cls) -> Type: @@ -25,7 +25,8 @@ def feed(self) -> List[Environment.CallingData]: result = [] elements = (content for module in self._modules for file in module.files for content in file.content) for element in elements: - if data := self._data_factory(element): - assert isinstance(data, Environment.CallingData), "Data is not an instance of Environment.CallingData" - result.append(data) + for factory in self._data_factories: + if data := factory(element): + assert isinstance(data, Environment.CallingData), "Data is not an instance of Environment.CallingData" + result.append(data) return result diff --git a/tests/preprocessing/unit/premade/components/flow/__init__.py b/tests/preprocessing/unit/premade/components/flow/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py index 5f0fce6..11551eb 100644 --- a/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py +++ b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py @@ -25,7 +25,7 @@ def setUp(self): self._module = SourceModule("Elements", os.path.dirname(__file__) + r"/source_files/elements") def test_syntax_element_data_provider(self): - provider = SyntaxElementDataProvider([self._module], data_factory) + provider = SyntaxElementDataProvider([self._module], [data_factory]) result = provider.feed() self.assertEqual(len(result), 2) From 43640e2d26788adbe5505b48af0d25d9d8db0a53 Mon Sep 17 00:00:00 2001 From: xXenvy Date: Sat, 9 Aug 2025 20:49:08 +0200 Subject: [PATCH 6/6] feat(provider): `deep_search` attribute --- .../premade/components/flow/dataprovider.py | 29 +++++++++++++++---- .../flow/source_files/elements/elements.hpp | 6 +++- .../components/flow/test_dataprovider.py | 14 ++++++++- 3 files changed, 42 insertions(+), 7 deletions(-) diff --git a/src/devana/preprocessing/premade/components/flow/dataprovider.py b/src/devana/preprocessing/premade/components/flow/dataprovider.py index 2df79f5..1288e61 100644 --- a/src/devana/preprocessing/premade/components/flow/dataprovider.py +++ b/src/devana/preprocessing/premade/components/flow/dataprovider.py @@ -1,30 +1,49 @@ -from typing import List, Type, Optional, Callable +from typing import List, Type, Optional, Callable, Iterable from devana.preprocessing.premade.components.executor.environment import Environment from devana.syntax_abstraction.organizers.sourcemodule import SourceModule +from devana.syntax_abstraction.organizers.codecontainer import CodeContainer from devana.syntax_abstraction.syntax import ISyntaxElement from devana.preprocessing.preprocessor import ISource class SyntaxElementDataProvider(ISource): - """Scans syntax elements across provided source modules and produces a list of - Environment.CallingData instances using custom factory functions.""" + """ + Scans syntax elements across provided source modules and produces a list of + Environment.CallingData instances using custom factory functions. + + If `deep_search` is True, method `feed()` will also look inside nested containers + (e.g. namespaces, classes) to find elements. If False, it only + checks the top-level elements that are directly inside each file. + """ def __init__( self, modules: List[SourceModule], - data_factories: List[Callable[[ISyntaxElement], Optional[Environment.CallingData]]] + data_factories: List[Callable[[ISyntaxElement], Optional[Environment.CallingData]]], + deep_search: bool = True ): self._modules = modules self._data_factories = data_factories + self._deep_search = deep_search @classmethod def get_produced_type(cls) -> Type: return Environment.CallingData def feed(self) -> List[Environment.CallingData]: + def get_syntax_elements(container: CodeContainer) -> Iterable[ISyntaxElement]: + """Yield all ISyntaxElement instances from container recursively.""" + for content in container.content: + yield content + if isinstance(content, CodeContainer) and self._deep_search: + yield from get_syntax_elements(content) + result = [] - elements = (content for module in self._modules for file in module.files for content in file.content) + files = (file for module in self._modules for file in module.files) + elements = (element for file in files for element in get_syntax_elements(file)) + for element in elements: + # call every factory with the current element for factory in self._data_factories: if data := factory(element): assert isinstance(data, Environment.CallingData), "Data is not an instance of Environment.CallingData" diff --git a/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp b/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp index 1ab6fb4..505b642 100644 --- a/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp +++ b/tests/preprocessing/unit/premade/components/flow/source_files/elements/elements.hpp @@ -6,4 +6,8 @@ struct Abc { int a; }; -class p_Class; \ No newline at end of file +class p_Class; + +namespace test_namespace { + struct p_Struct{}; +} \ No newline at end of file diff --git a/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py index 11551eb..2b8396e 100644 --- a/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py +++ b/tests/preprocessing/unit/premade/components/flow/test_dataprovider.py @@ -24,9 +24,21 @@ class TestSyntaxElementDataProvider(unittest.TestCase): def setUp(self): self._module = SourceModule("Elements", os.path.dirname(__file__) + r"/source_files/elements") - def test_syntax_element_data_provider(self): + def test_syntax_element_data_provider_with_deep_search(self): provider = SyntaxElementDataProvider([self._module], [data_factory]) result = provider.feed() + self.assertEqual(len(result), 3) + + for data in result: + self.assertTrue(isinstance(data, Environment.CallingData)) + self.assertEqual(len(data.arguments.positional), 0) + self.assertEqual(len(data.arguments.named), 0) + self.assertEqual(data.signature.name, "Parsable") + self.assertEqual(data.signature.namespaces, []) + + def test_syntax_element_data_provider_without_deep_search(self): + provider = SyntaxElementDataProvider([self._module], [data_factory], deep_search=False) + result = provider.feed() self.assertEqual(len(result), 2) for data in result: