Processor.resolve_fqn() - Code Metrics - Inspection of "Added Processor class that handles model imports a..." - zayfod/pyfranca - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( b15539...e299b1 )

by Kaloyan

created 2017-02-16 16:29 UTC

Processor.resolve_fqn() B

↳ Parent: Processor

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	1
CRAP Score	25.0182

Importance

Changes

Metric	Value
cc	5
dl	0
loc	14
ccs	1
cts	14
cp	0.0714
crap	25.0182
rs	8.5454
c	0
b	0
f	0


import os
from collections import OrderedDict
from pyfranca import franca_parser, ast


class ProcessorException(Exception):

    def __init__(self, message):
class SomeParent:
    def __init__(self):
        self.x = 1

class SomeChild(SomeParent):
    def __init__(self):
        # Initialize the super class
        SomeParent.__init__(self)
        self.message = message

    def __str__(self):
        return self.message


class Processor:
    """
    Franca IDL processor.
    """

    def __init__(self):
        """
        Constructor.
        """
        # Default package paths.
        self.package_paths = ["."]
        self.files = {}
        self.packages = {}

    @staticmethod
    def basename(namespace):
        """
        Extract the type or namespace name from a Franca FQN.
        """
        dot = namespace.rfind(".")
        if dot == -1:
            return namespace
        else:
            return namespace[dot + 1:]

    @staticmethod
    def packagename(namespace):
        """
        Extract the package name from a Franca FQN.
        """
        dot = namespace.rfind(".")
        if dot == -1:
            return None
        else:
            return namespace[0:dot]

    # TODO: Remove?
    def resolve_fqn(self, fqn):
        package_fqn = fqn
        while True:
            package_fqn = self.packagename(package_fqn)
            if package_fqn is None:
                return None
            if package_fqn in self.packages:
                break
        package = self.packages[package_fqn]
        namespace_name = fqn[len(package_fqn)+1:]
        if namespace_name not in package:
            return None
        namespace = package[namespace_name]
        return package, namespace

    @staticmethod
    def resolve(namespace, name):
        """
        Resolve type references.

        :param namespace: context ast.Namespace object.
        :param name: name string.
        :return: Dereferenced ast.Type object.
        """
        # TODO: relative and FQN imports
        # FIXME: Circular references
        if not isinstance(namespace, ast.Namespace) or \
                not isinstance(name, str):
            raise ValueError("Unexpected input.")
        # Look in the type's namespace
        if name in namespace:
            return namespace[name]
        # Look in other type collections in the type's package
        for typecollection in namespace.package.typecollections.values():
            if name in typecollection:
                return typecollection[name]
        # Look in imports
        for package_import in namespace.package.imports:
            if package_import.namespace_reference:
                # Look in namespaces imported in the type's package
                if name in package_import.namespace_reference:
                    return package_import.namespace_reference[name]
            else:
                # Look in typecollections of packages imported in the type's
                #   package using FQNs.
                # FIXME: FQNs
                for typecollection in \
                        package_import.package_reference.typecollections:
                    if name in typecollection:
                        return typecollection[name]
        # Give up
        raise ProcessorException(
            "Unresolved reference \"{}\".".format(name))

    def _udpate_complextype_references(self, name):
        """
        Update type references in a complex type.

        :param name: ast.ComplexType object.
        """
        if isinstance(name, ast.Enumeration):
            # TODO: Handle extends
            if name.extends:
                self._update_type_references(name.namespace, name.extends)
        elif isinstance(name, ast.Struct):
            for field in name.fields.values():
                self._update_type_references(name.namespace, field.type)
            # TODO: Handle extends
            if name.extends:
                self._update_type_references(name.namespace, name.extends)
        elif isinstance(name, ast.Array):
            self._update_type_references(name.namespace, name.type)
        elif isinstance(name, ast.Map):
            self._update_type_references(name.namespace, name.key_type)
            self._update_type_references(name.namespace, name.value_type)
        else:
            assert False

    def _update_type_references(self, namespace, name):
        """
        Update type references in a type.

        :param namespace: ast.Namespace context.
        :param name: ast.Type object.
        """
        if isinstance(name, ast.Typedef):
            self._update_type_references(name.namespace, name.type)
        elif isinstance(name, ast.PrimitiveType):
            pass
        elif isinstance(name, ast.ComplexType):
            self._udpate_complextype_references(name)
        elif isinstance(name, ast.Reference):
            if not name.namespace:
                name.namespace = namespace
            if not name.reference:
                resolved_name = self.resolve(namespace, name.name)
                name.reference = resolved_name
        elif isinstance(name, ast.Attribute):
            self._update_type_references(name.namespace, name.type)
        elif isinstance(name, ast.Method):
            for arg in name.in_args.values():
                self._update_type_references(name.namespace, arg.type)
            for arg in name.out_args.values():
                self._update_type_references(name.namespace, arg.type)
            if isinstance(name.errors, OrderedDict):
                for arg in name.errors.values():
                    self._update_type_references(name.namespace, arg.type)
            else:
                # Errors can be a reference to an enumeration
                self._update_type_references(name.namespace, name.errors)
                # FIXME: Check the reference type
        elif isinstance(name, ast.Broadcast):
            for arg in name.out_args.values():
                self._update_type_references(name.namespace, arg.type)
        else:
            assert False

    def _update_namespace_references(self, namespace):
        """
        Update type references in a namespace.

        :param namespace: ast.Namespace object.
        """
        for name in namespace.typedefs.values():
            self._update_type_references(namespace, name)
        for name in namespace.enumerations.values():
            self._update_type_references(namespace, name)
        for name in namespace.structs.values():
            self._update_type_references(namespace, name)
        for name in namespace.arrays.values():
            self._update_type_references(namespace, name)
        for name in namespace.maps.values():
            self._update_type_references(namespace, name)

    def _update_interface_references(self, namespace):
        """
        Update type references in an interface.

        :param namespace: ast.Interface object.
        """
        self._update_namespace_references(namespace)
        for name in namespace.attributes.values():
            self._update_type_references(namespace, name)
        for name in namespace.methods.values():
            self._update_type_references(namespace, name)
        for name in namespace.broadcasts.values():
            self._update_type_references(namespace, name)
        # TODO: Handle extends
        if namespace.extends:
            self._update_type_references(namespace, namespace.extends)

    def _update_package_references(self, package):
        """
        Update type references in a package.

        :param package: ast.Package object.
        """
        for package_import in package.imports:
            assert package_import.package_reference is not None
            if package_import.namespace:
                # Namespace import
                package_reference = package_import.package_reference
                if not package_import.namespace.endswith(".*"):
                    raise ProcessorException(
                        "Invalid namespace import {}.".format(
                            package_import.namespace))
                namespace_name = \
                    package_import.namespace[len(package_reference.name)+1:-2]
                # Update namespace reference
                if namespace_name in package_reference:
                    namespace = package_reference[namespace_name]
                    package_import.namespace_reference = namespace
                else:
                    raise ProcessorException(
                        "Namespace \"{}\" not found.".format(
                            package_import.namespace))
            else:
                # Model import
                assert package_import.namespace_reference is None
        for namespace in package.typecollections:
            self._update_namespace_references(
                package.typecollections[namespace])
        for namespace in package.interfaces:
            self._update_interface_references(
                package.interfaces[namespace])

    def import_package(self, fspec, package, references=None):
        """
        Import an ast.Package into the processor.

        :param fspec: File specification of the package.
        :param package: ast.Package object.
        :param references: A list of package references.
        """
        if not isinstance(package, ast.Package):
            ValueError("Expected ast.Package as input.")
        if not references:
            references = []
        # Check for circular package dependencies.
        if package.name in references:
            raise ProcessorException(
                "Circular dependency for package \"{}\".".format(package.name))
        # Check whether package is already imported
        if package.name in self.packages:
            if fspec != self.packages[package.name].file:
                raise ProcessorException(
                    "Package \"{}\" defined in multiple files.".format(
                        package.name))
            else:
                return
        # Register the package in the processor.
        self.packages[package.name] = package
        self.files[fspec] = package
        # Process package imports
        for package_import in package.imports:
            imported_package = self.import_file(
                package_import.file, references + [package.name])
            # Update import reference
            package_import.package_reference = imported_package
        # Update type references
        self._update_package_references(package)

    def import_string(self, fspec, fidl, references=None):
        """
        Parse an FIDL string and import it into the processor as package.

        :param fspec: File specification of the package.
        :param fidl: FIDL string.
        :param references: A list of package references.
        :return: The parsed ast.Package.
        """
        # Parse the string.
        parser = franca_parser.Parser()
        package = parser.parse(fidl)
        # Import the package in the processor.
        self.import_package(fspec, package, references)
        return package

    def import_file(self, fspec, references=None):
        """
        Parse an FIDL file and import it into the processor as package.

        :param fspec: File specification.
        :param references: A list of package references.
        :return: The parsed ast.Package.
        """
        if fspec in self.files:
            # File already loaded.
            return self.files[fspec]
        if not os.path.exists(fspec):
            if os.path.isabs(fspec):
                # Absolute specification
                raise ProcessorException(
                    "Model \"{}\" not found.".format(fspec))
            else:
                # Relative specification - check in the package path list.
                for path in self.package_paths:
                    temp_fspec = os.path.join(path, fspec)
                    if os.path.exists(temp_fspec):
                        fspec = temp_fspec
                        break
                else:
                    raise ProcessorException(
                        "Model \"{}\" not found.".format(fspec))
        # Parse the file.
        parser = franca_parser.Parser()
        package = parser.parse_file(fspec)
        # Import the package in the processor.
        self.import_package(fspec, package, references)
        return package


1
2	1	import os
3	1	from collections import OrderedDict
4	1	from pyfranca import franca_parser, ast
5
6
7	1	class ProcessorException(Exception):
8
9	1	def __init__(self, message):
		0 ignored issues – show Bug introduced 2017-02-16 16:31 UTC by Report Bug Copy Issue Report The `__init__` method of the super-class `Exception` is not called. It is generally advisable to initialize the super-class by calling its `__init__` method: class SomeParent: def __init__(self): self.x = 1 class SomeChild(SomeParent): def __init__(self): # Initialize the super class SomeParent.__init__(self) Loading history...
10	1	self.message = message
11
12	1	def __str__(self):
13	1	return self.message
14
15
16	1	class Processor:
17		"""
18		Franca IDL processor.
19		"""
20
21	1	def __init__(self):
22		"""
23		Constructor.
24		"""
25		# Default package paths.
26	1	self.package_paths = ["."]
27	1	self.files = {}
28	1	self.packages = {}
29
30	1	@staticmethod
31		def basename(namespace):
32		"""
33		Extract the type or namespace name from a Franca FQN.
34		"""
35		dot = namespace.rfind(".")
36		if dot == -1:
37		return namespace
38		else:
39		return namespace[dot + 1:]
40
41	1	@staticmethod
42		def packagename(namespace):
43		"""
44		Extract the package name from a Franca FQN.
45		"""
46		dot = namespace.rfind(".")
47		if dot == -1:
48		return None
49		else:
50		return namespace[0:dot]
51
52		# TODO: Remove?
53	1	def resolve_fqn(self, fqn):
54		package_fqn = fqn
55		while True:
56		package_fqn = self.packagename(package_fqn)
57		if package_fqn is None:
58		return None
59		if package_fqn in self.packages:
60		break
61		package = self.packages[package_fqn]
62		namespace_name = fqn[len(package_fqn)+1:]
63		if namespace_name not in package:
64		return None
65		namespace = package[namespace_name]
66		return package, namespace
67
68	1	@staticmethod
69		def resolve(namespace, name):
70		"""
71		Resolve type references.
72
73		:param namespace: context ast.Namespace object.
74		:param name: name string.
75		:return: Dereferenced ast.Type object.
76		"""
77		# TODO: relative and FQN imports
78		# FIXME: Circular references
79	1	if not isinstance(namespace, ast.Namespace) or \
80		not isinstance(name, str):
81		raise ValueError("Unexpected input.")
82		# Look in the type's namespace
83	1	if name in namespace:
84	1	return namespace[name]
85		# Look in other type collections in the type's package
86	1	for typecollection in namespace.package.typecollections.values():
87	1	if name in typecollection:
88	1	return typecollection[name]
89		# Look in imports
90	1	for package_import in namespace.package.imports:
91	1	if package_import.namespace_reference:
92		# Look in namespaces imported in the type's package
93	1	if name in package_import.namespace_reference:
94	1	return package_import.namespace_reference[name]
95		else:
96		# Look in typecollections of packages imported in the type's
97		# package using FQNs.
98		# FIXME: FQNs
99		for typecollection in \
100		package_import.package_reference.typecollections:
101		if name in typecollection:
102		return typecollection[name]
103		# Give up
104	1	raise ProcessorException(
105		"Unresolved reference \"{}\".".format(name))
106
107	1	def _udpate_complextype_references(self, name):
108		"""
109		Update type references in a complex type.
110
111		:param name: ast.ComplexType object.
112		"""
113	1	if isinstance(name, ast.Enumeration):
114		# TODO: Handle extends
115		if name.extends:
116		self._update_type_references(name.namespace, name.extends)
117	1	elif isinstance(name, ast.Struct):
118	1	for field in name.fields.values():
119	1	self._update_type_references(name.namespace, field.type)
120		# TODO: Handle extends
121		if name.extends:
122		self._update_type_references(name.namespace, name.extends)
123	1	elif isinstance(name, ast.Array):
124	1	self._update_type_references(name.namespace, name.type)
125	1	elif isinstance(name, ast.Map):
126	1	self._update_type_references(name.namespace, name.key_type)
127	1	self._update_type_references(name.namespace, name.value_type)
128		else:
129		assert False
130
131	1	def _update_type_references(self, namespace, name):
132		"""
133		Update type references in a type.
134
135		:param namespace: ast.Namespace context.
136		:param name: ast.Type object.
137		"""
138	1	if isinstance(name, ast.Typedef):
139	1	self._update_type_references(name.namespace, name.type)
140	1	elif isinstance(name, ast.PrimitiveType):
141	1	pass
142	1	elif isinstance(name, ast.ComplexType):
143	1	self._udpate_complextype_references(name)
144	1	elif isinstance(name, ast.Reference):
145	1	if not name.namespace:
146	1	name.namespace = namespace
147	1	if not name.reference:
148	1	resolved_name = self.resolve(namespace, name.name)
149	1	name.reference = resolved_name
150	1	elif isinstance(name, ast.Attribute):
151	1	self._update_type_references(name.namespace, name.type)
152	1	elif isinstance(name, ast.Method):
153	1	for arg in name.in_args.values():
154	1	self._update_type_references(name.namespace, arg.type)
155	1	for arg in name.out_args.values():
156	1	self._update_type_references(name.namespace, arg.type)
157	1	if isinstance(name.errors, OrderedDict):
158		for arg in name.errors.values():
159		self._update_type_references(name.namespace, arg.type)
160		else:
161		# Errors can be a reference to an enumeration
162	1	self._update_type_references(name.namespace, name.errors)
163		# FIXME: Check the reference type
164	1	elif isinstance(name, ast.Broadcast):
165	1	for arg in name.out_args.values():
166	1	self._update_type_references(name.namespace, arg.type)
167		else:
168		assert False
169
170	1	def _update_namespace_references(self, namespace):
171		"""
172		Update type references in a namespace.
173
174		:param namespace: ast.Namespace object.
175		"""
176	1	for name in namespace.typedefs.values():
177	1	self._update_type_references(namespace, name)
178	1	for name in namespace.enumerations.values():
179		self._update_type_references(namespace, name)
180	1	for name in namespace.structs.values():
181	1	self._update_type_references(namespace, name)
182	1	for name in namespace.arrays.values():
183	1	self._update_type_references(namespace, name)
184	1	for name in namespace.maps.values():
185	1	self._update_type_references(namespace, name)
186
187	1	def _update_interface_references(self, namespace):
188		"""
189		Update type references in an interface.
190
191		:param namespace: ast.Interface object.
192		"""
193	1	self._update_namespace_references(namespace)
194	1	for name in namespace.attributes.values():
195	1	self._update_type_references(namespace, name)
196	1	for name in namespace.methods.values():
197	1	self._update_type_references(namespace, name)
198	1	for name in namespace.broadcasts.values():
199	1	self._update_type_references(namespace, name)
200		# TODO: Handle extends
201	1	if namespace.extends:
202		self._update_type_references(namespace, namespace.extends)
203
204	1	def _update_package_references(self, package):
205		"""
206		Update type references in a package.
207
208		:param package: ast.Package object.
209		"""
210	1	for package_import in package.imports:
211	1	assert package_import.package_reference is not None
212	1	if package_import.namespace:
213		# Namespace import
214	1	package_reference = package_import.package_reference
215	1	if not package_import.namespace.endswith(".*"):
216		raise ProcessorException(
217		"Invalid namespace import {}.".format(
218		package_import.namespace))
219	1	namespace_name = \
220		package_import.namespace[len(package_reference.name)+1:-2]
221		# Update namespace reference
222	1	if namespace_name in package_reference:
223	1	namespace = package_reference[namespace_name]
224	1	package_import.namespace_reference = namespace
225		else:
226	1	raise ProcessorException(
227		"Namespace \"{}\" not found.".format(
228		package_import.namespace))
229		else:
230		# Model import
231		assert package_import.namespace_reference is None
232	1	for namespace in package.typecollections:
233	1	self._update_namespace_references(
234		package.typecollections[namespace])
235	1	for namespace in package.interfaces:
236	1	self._update_interface_references(
237		package.interfaces[namespace])
238
239	1	def import_package(self, fspec, package, references=None):
240		"""
241		Import an ast.Package into the processor.
242
243		:param fspec: File specification of the package.
244		:param package: ast.Package object.
245		:param references: A list of package references.
246		"""
247	1	if not isinstance(package, ast.Package):
248		ValueError("Expected ast.Package as input.")
249	1	if not references:
250	1	references = []
251		# Check for circular package dependencies.
252	1	if package.name in references:
253		raise ProcessorException(
254		"Circular dependency for package \"{}\".".format(package.name))
255		# Check whether package is already imported
256	1	if package.name in self.packages:
257	1	if fspec != self.packages[package.name].file:
258	1	raise ProcessorException(
259		"Package \"{}\" defined in multiple files.".format(
260		package.name))
261		else:
262		return
263		# Register the package in the processor.
264	1	self.packages[package.name] = package
265	1	self.files[fspec] = package
266		# Process package imports
267	1	for package_import in package.imports:
268	1	imported_package = self.import_file(
269		package_import.file, references + [package.name])
270		# Update import reference
271	1	package_import.package_reference = imported_package
272		# Update type references
273	1	self._update_package_references(package)
274
275	1	def import_string(self, fspec, fidl, references=None):
276		"""
277		Parse an FIDL string and import it into the processor as package.
278
279		:param fspec: File specification of the package.
280		:param fidl: FIDL string.
281		:param references: A list of package references.
282		:return: The parsed ast.Package.
283		"""
284		# Parse the string.
285	1	parser = franca_parser.Parser()
286	1	package = parser.parse(fidl)
287		# Import the package in the processor.
288	1	self.import_package(fspec, package, references)
289	1	return package
290
291	1	def import_file(self, fspec, references=None):
292		"""
293		Parse an FIDL file and import it into the processor as package.
294
295		:param fspec: File specification.
296		:param references: A list of package references.
297		:return: The parsed ast.Package.
298		"""
299	1	if fspec in self.files:
300		# File already loaded.
301	1	return self.files[fspec]
302	1	if not os.path.exists(fspec):
303	1	if os.path.isabs(fspec):
304		# Absolute specification
305		raise ProcessorException(
306		"Model \"{}\" not found.".format(fspec))
307		else:
308		# Relative specification - check in the package path list.
309	1	for path in self.package_paths:
310	1	temp_fspec = os.path.join(path, fspec)
311	1	if os.path.exists(temp_fspec):
312		fspec = temp_fspec
313		break
314		else:
315	1	raise ProcessorException(
316		"Model \"{}\" not found.".format(fspec))
317		# Parse the file.
318		parser = franca_parser.Parser()
319		package = parser.parse_file(fspec)
320		# Import the package in the processor.
321		self.import_package(fspec, package, references)
322		return package
323

zayfod / pyfranca

GitHub Access Token became invalid

Push — master ( b15539...e299b1 )

Processor.resolve_fqn() B

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like