exabgp.configuration.core.tokeniser.Tokeniser.set_file() - Code Metrics - Inspection of "large change required to perform better route vali..." - Exa-Networks/exabgp - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Push — master ( 25ce4f...b9bc4f )

by Thomas

created 2019-09-24 00:20 UTC

Tokeniser.set_file() B

↳ Parent: exabgp.configuration.core.tokeniser

Complexity

Conditions

Size

Total Lines	24
Code Lines	22

Duplication

Lines	0
Ratio	0 %

Importance

Changes

Metric	Value
cc	7
eloc	22
nop	2
dl	0
loc	24
rs	7.952
c	0
b	0
f	0

# encoding: utf-8
"""
tokeniser.py

Created by Thomas Mangin on 2015-06-05.
Copyright (c) 2009-2017 Exa Networks. All rights reserved.
License: 3-clause BSD. (See the COPYRIGHT file)
"""

from exabgp.configuration.core.format import tokens
from exabgp.protocol.family import AFI
from collections import deque
from exabgp.vendoring import six


class Tokeniser (object):

	class Iterator (object):
		fname = ''  # This is ok to have a unique value as API parser do not use files

		def __init__ (self):
			self.next = deque()
			self.tokens = []
			self.generator = iter([])
			self.announce = True

		def replenish (self, content):
			self.next.clear()
			self.tokens = content
			self.generator = iter(content)
			return self

		def clear (self):
			self.replenish([])
			self.announce = True

		def __call__ (self):
			if self.next:
				return self.next.popleft()

			try:
				return six.next(self.generator)
			except StopIteration:
				return ''

		def peek (self):
			try:
				peaked = six.next(self.generator)
				self.next.append(peaked)
				return peaked
			except StopIteration:
				return ''

	@staticmethod
	def _off ():
		return iter([])

	def __init__ (self, scope, error, logger):
		self.scope = scope
		self.error = error
		self.logger = logger
		self.finished = False
		self.number = 0
		self.line = []
		self.iterate = Tokeniser.Iterator()
		self.end = ''
		self.index_column = 0
		self.index_line = 0
		self.fname = ''
		self.type = 'unset'
		self.afi = AFI.undefined

		self._tokens = Tokeniser._off
		self._next = None
		self._data = None

	def clear (self):
		self.finished = False
		self.number = 0
		self.line = []
		self.iterate.clear()
		self.end = ''
		self.index_column = 0
		self.index_line = 0
		self.fname = ''
		self.type = 'unset'
		if self._data:
			self._set(self._data)

	def params (self):
		if len(self.line) <= 2:
			return ''
		if self.end in ('{','}',';'):
			return "'%s'" % "' '".join(self.line[1:-1])
		return "'%s'" % "' '".join(self.line[1:])

	def _tokenise (self,iterator):
		for parsed in tokens(iterator):
			words = [word for y,x,word in parsed]
			self.line = ''.join(words)
			# ignore # lines
			# set Location information
			yield words

	def _set (self, function):
		try:
			self._tokens = function
			self._next = six.next(self._tokens)
		except IOError as exc:
			error = str(exc)
			if error.count(']'):
				self.error.set(error.split(']')[1].strip())
			else:
				self.error.set(error)
			self._tokens = Tokeniser._off
			self._next = []
			return self.error.set('issue setting the configuration parser')
		except StopIteration:
			self._tokens = Tokeniser._off
			self._next = []
			return self.error.set('issue setting the configuration parser, no data')
		return True

	def set_file (self, data):
		def _source (fname):
			with open(fname,'r') as fileobject:
				def formated ():
					line = ''
					for current in fileobject:
						self.index_line += 1
						current = current.rstrip()
						if current.endswith('\\'):
							line += current
							continue
						elif line:
							yield line + current
							line = ''
						else:
							yield current
					if line:
						yield line

				for _ in self._tokenise(formated()):
					yield _
		self.type = 'file'
		self.Iterator.fname = data
		return self._set(_source(data))

	def set_text (self, data):
		def _source (data):
			for _ in self._tokenise(data.split('\n')):
				yield _
		self.type = 'text'
		return self._set(_source(data))

	def set_api (self, line):
		return self._set(self._tokenise(iter([line])))

	def set_action(self, command):
		if command != 'announce':
			self.iterate.announce = False

	def __call__ (self):
		self.number += 1
		try:
			self.line, self._next = self._next, six.next(self._tokens)
			self.end = self.line[-1]
		except StopIteration:
			if not self.finished:
				self.finished = True
				self.line, self._next = self._next, []
				self.end = self.line[-1]
			else:
				self.line = []
				self.end = ''

		# should we raise a Location if called with no more data ?
		self.iterate.replenish(self.line[:-1])

		return self.line


1			# encoding: utf-8
2			"""
3			tokeniser.py
4
5			Created by Thomas Mangin on 2015-06-05.
6			Copyright (c) 2009-2017 Exa Networks. All rights reserved.
7			License: 3-clause BSD. (See the COPYRIGHT file)
8			"""
9
10			from exabgp.configuration.core.format import tokens
11			from exabgp.protocol.family import AFI
12			from collections import deque
13			from exabgp.vendoring import six
14
15
16			class Tokeniser (object):
17
18			class Iterator (object):
19			fname = '' # This is ok to have a unique value as API parser do not use files
20
21			def __init__ (self):
22			self.next = deque()
23			self.tokens = []
24			self.generator = iter([])
25			self.announce = True
26
27			def replenish (self, content):
28			self.next.clear()
29			self.tokens = content
30			self.generator = iter(content)
31			return self
32
33			def clear (self):
34			self.replenish([])
35			self.announce = True
36
37			def __call__ (self):
38			if self.next:
39			return self.next.popleft()
40
41			try:
42			return six.next(self.generator)
43			except StopIteration:
44			return ''
45
46			def peek (self):
47			try:
48			peaked = six.next(self.generator)
49			self.next.append(peaked)
50			return peaked
51			except StopIteration:
52			return ''
53
54			@staticmethod
55			def _off ():
56			return iter([])
57
58			def __init__ (self, scope, error, logger):
59			self.scope = scope
60			self.error = error
61			self.logger = logger
62			self.finished = False
63			self.number = 0
64			self.line = []
65			self.iterate = Tokeniser.Iterator()
66			self.end = ''
67			self.index_column = 0
68			self.index_line = 0
69			self.fname = ''
70			self.type = 'unset'
71			self.afi = AFI.undefined
72
73			self._tokens = Tokeniser._off
74			self._next = None
75			self._data = None
76
77			def clear (self):
78			self.finished = False
79			self.number = 0
80			self.line = []
81			self.iterate.clear()
82			self.end = ''
83			self.index_column = 0
84			self.index_line = 0
85			self.fname = ''
86			self.type = 'unset'
87			if self._data:
88			self._set(self._data)
89
90			def params (self):
91			if len(self.line) <= 2:
92			return ''
93			if self.end in ('{','}',';'):
94			return "'%s'" % "' '".join(self.line[1:-1])
95			return "'%s'" % "' '".join(self.line[1:])
96
97			def _tokenise (self,iterator):
98			for parsed in tokens(iterator):
99			words = [word for y,x,word in parsed]
100			self.line = ''.join(words)
101			# ignore # lines
102			# set Location information
103			yield words
104
105			def _set (self, function):
106			try:
107			self._tokens = function
108			self._next = six.next(self._tokens)
109			except IOError as exc:
110			error = str(exc)
111			if error.count(']'):
112			self.error.set(error.split(']')[1].strip())
113			else:
114			self.error.set(error)
115			self._tokens = Tokeniser._off
116			self._next = []
117			return self.error.set('issue setting the configuration parser')
118			except StopIteration:
119			self._tokens = Tokeniser._off
120			self._next = []
121			return self.error.set('issue setting the configuration parser, no data')
122			return True
123
124			def set_file (self, data):
125			def _source (fname):
126			with open(fname,'r') as fileobject:
127			def formated ():
128			line = ''
129			for current in fileobject:
130			self.index_line += 1
131			current = current.rstrip()
132			if current.endswith('\\'):
133			line += current
134			continue
135			elif line:
136			yield line + current
137			line = ''
138			else:
139			yield current
140			if line:
141			yield line
142
143			for _ in self._tokenise(formated()):
144			yield _
145			self.type = 'file'
146			self.Iterator.fname = data
147			return self._set(_source(data))
148
149			def set_text (self, data):
150			def _source (data):
151			for _ in self._tokenise(data.split('\n')):
152			yield _
153			self.type = 'text'
154			return self._set(_source(data))
155
156			def set_api (self, line):
157			return self._set(self._tokenise(iter([line])))
158
159			def set_action(self, command):
160			if command != 'announce':
161			self.iterate.announce = False
162
163			def __call__ (self):
164			self.number += 1
165			try:
166			self.line, self._next = self._next, six.next(self._tokens)
167			self.end = self.line[-1]
168			except StopIteration:
169			if not self.finished:
170			self.finished = True
171			self.line, self._next = self._next, []
172			self.end = self.line[-1]
173			else:
174			self.line = []
175			self.end = ''
176
177			# should we raise a Location if called with no more data ?
178			self.iterate.replenish(self.line[:-1])
179
180			return self.line
181

Exa-Networks / exabgp

Push — master ( 25ce4f...b9bc4f )

Tokeniser.set_file() B

Complexity

Size

Duplication

Importance

Duplication Side-by-Side

Filter issues like