1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
|
# Copyright (c) Pelagicore AB 2016
from jinja2 import Environment, Template
from jinja2 import FileSystemLoader, PackageLoader, ChoiceLoader
from jinja2 import TemplateSyntaxError, TemplateNotFound, TemplateError
from path import Path
from antlr4 import FileStream, CommonTokenStream, ParseTreeWalker
from antlr4.error import DiagnosticErrorListener, ErrorListener
import shelve
import logging
import hashlib
import yaml
import click
import sys
from .idl.parser.TLexer import TLexer
from .idl.parser.TParser import TParser
from .idl.parser.TListener import TListener
from .idl.domain import System
from .idl.listener import DomainListener
from .utils import merge
from .filters import filters
try:
from yaml import CLoader as Loader, CDumper as Dumper
except ImportError:
from yaml import Loader, Dumper
logger = logging.getLogger(__name__)
"""
Provides an API for accessing the file system and controlling the generator
"""
class ReportingErrorListener(ErrorListener.ErrorListener):
def __init__(self, document):
self.document = document
def syntaxError(self, recognizer, offendingSymbol, line, column, msg, e):
msg = '{0}:{1}:{2} {2}'.format(self.document, line, column, msg)
click.secho(msg, fg='red')
raise ValueError(msg)
def reportAmbiguity(self, recognizer, dfa, startIndex, stopIndex, exact, ambigAlts, configs):
click.secho('ambiguity', fg='red')
def reportAttemptingFullContext(self, recognizer, dfa, startIndex, stopIndex, conflictingAlts, configs):
click.secho('reportAttemptingFullContext', fg='red')
def reportContextSensitivity(self, recognizer, dfa, startIndex, stopIndex, prediction, configs):
click.secho('reportContextSensitivity', fg='red')
class Generator(object):
"""Manages the templates and applies your context data"""
strict = False
""" enables strict code generation """
def __init__(self, search_path: str, context: dict={}):
loader = ChoiceLoader([
FileSystemLoader(search_path),
PackageLoader('qface')
])
self.env = Environment(
loader=loader,
trim_blocks=True,
lstrip_blocks=True
)
self.env.filters.update(filters)
self._destination = Path()
self.context = context
@property
def destination(self):
"""destination prefix for generator write"""
return self._destination
@destination.setter
def destination(self, dst: str):
if dst:
self._destination = Path(self.apply(dst, self.context))
@property
def filters(self):
return self.env.filters
@filters.setter
def filters(self, filters):
self.env.filters.update(filters)
def get_template(self, name: str):
"""Retrieves a single template file from the template loader"""
return self.env.get_template(name)
def render(self, name: str, context: dict):
"""Returns the rendered text from a single template file from the
template loader using the given context data"""
template = self.get_template(name)
return template.render(context)
def apply(self, template: str, context: dict):
"""Return the rendered text of a template instance"""
return self.env.from_string(template).render(context)
def write(self, file_path: Path, template: str, context: dict={}, preserve: bool = False):
"""Using a template file name it renders a template
into a file given a context
"""
if not context:
context = self.context
error = False
try:
self._write(file_path, template, context, preserve)
except TemplateSyntaxError as exc:
# import pdb; pdb.set_trace()
message = '{0}:{1} error: {2}'.format(exc.filename, exc.lineno, exc.message)
click.secho(message, fg='red')
error = True
except TemplateNotFound as exc:
message = '{0} error: Template not found'.format(exc.name)
click.secho(message, fg='red')
error = True
except TemplateError as exc:
message = 'error: {0}'.format(exc.message)
click.secho(message, fg='red')
error = True
if error and Generator.strict:
sys.exit(-1)
def _write(self, file_path: Path, template: str, context: dict, preserve: bool = False):
path = self.destination / Path(self.apply(file_path, context))
path.parent.makedirs_p()
logger.info('write {0}'.format(path))
data = self.render(template, context)
if self._has_different_content(data, path):
if path.exists() and preserve:
click.secho('preserve changed file: {0}'.format(path), fg='blue')
else:
click.secho('write changed file: {0}'.format(path), fg='blue')
path.open('w', encoding='utf-8').write(data)
def _has_different_content(self, data, path):
if not path.exists():
return True
dataHash = hashlib.new('md5', data.encode('utf-8')).digest()
pathHash = path.read_hash('md5')
return dataHash != pathHash
def register_filter(self, name, callback):
"""Register your custom template filter"""
self.env.filters[name] = callback
class RuleGenerator(Generator):
"""Generates documents based on a rule YAML document"""
def __init__(self, search_path: str, destination: Path, context: dict= {}):
super().__init__(search_path, context)
self.context.update({
'dst': destination,
'project': Path(destination).name,
})
self.destination = '{{dst}}'
def process_rules(self, document: Path, system: System):
"""writes the templates read from the rules document"""
self.context.update({'system': system})
rules = FileSystem.load_yaml(document, required=True)
for name, target in rules.items():
click.secho('process target: {0}'.format(name), fg='green')
self._process_target(target, system)
def _process_target(self, rules: dict, system: System):
""" process a set of rules for a target """
self.context.update(rules.get('context', {}))
self.destination = rules.get('destination', '{{dst}}')
self._process_rule(rules.get('system', None), {'system': system})
for module in system.modules:
self._process_rule(rules.get('module', None), {'module': module})
for interface in module.interfaces:
self._process_rule(rules.get('interface', None), {'interface': interface})
for struct in module.structs:
self._process_rule(rules.get('struct', None), {'struct': struct})
for enum in module.enums:
self._process_rule(rules.get('enum', None), {'enum': enum})
def _process_rule(self, rule: dict, context: dict):
""" process a single rule """
if not rule:
return
self.context.update(context)
self.context.update(rule.get('context', {}))
self.destination = rule.get('destination', None)
for target, source in rule.get('documents', {}).items():
self.write(target, source)
class FileSystem(object):
"""QFace helper functions to work with the file system"""
strict = False
""" enables strict parsing """
@staticmethod
def parse_document(document: Path, system: System = None):
error = False
try:
return FileSystem._parse_document(document, system)
except FileNotFoundError as e:
click.secho('{0}: file not found'.format(document), fg='red')
error = True
except ValueError as e:
click.secho('Error parsing document {0}'.format(document))
error = True
if error and FileSystem.strict:
sys.exit(-1)
@staticmethod
def _parse_document(document: Path, system: System = None):
"""Parses a document and returns the resulting domain system
:param path: document path to parse
:param system: system to be used (optional)
"""
logger.debug('parse document: {0}'.format(document))
stream = FileStream(str(document), encoding='utf-8')
system = FileSystem._parse_stream(stream, system, document)
FileSystem.merge_annotations(system, document.stripext() + '.yaml')
return system
@staticmethod
def _parse_stream(stream, system: System = None, document=None):
logger.debug('parse stream')
system = system or System()
lexer = TLexer(stream)
stream = CommonTokenStream(lexer)
parser = TParser(stream)
parser.addErrorListener(ReportingErrorListener(document))
tree = parser.documentSymbol()
walker = ParseTreeWalker()
walker.walk(DomainListener(system), tree)
return system
@staticmethod
def merge_annotations(system, document):
"""Read a YAML document and for each root symbol identifier
updates the tag information of that symbol
"""
meta = FileSystem.load_yaml(document)
click.secho('merge tags from {0}'.format(document), fg='blue')
for identifier, data in meta.items():
symbol = system.lookup(identifier)
if symbol:
merge(symbol.tags, data)
@staticmethod
def parse(input, identifier: str = None, use_cache=False, clear_cache=True, pattern="*.qface"):
"""Input can be either a file or directory or a list of files or directory.
A directory will be parsed recursively. The function returns the resulting system.
Stores the result of the run in the domain cache named after the identifier.
:param path: directory to parse
:param identifier: identifies the parse run. Used to name the cache
:param clear_cache: clears the domain cache (defaults to true)
"""
inputs = input if isinstance(input, (list, tuple)) else [input]
logger.debug('parse input={0}'.format(inputs))
identifier = 'system' if not identifier else identifier
system = System()
cache = None
if use_cache:
cache = shelve.open('qface.cache')
if identifier in cache and clear_cache:
del cache[identifier]
if identifier in cache:
# use the cached domain model
system = cache[identifier]
# if domain model not cached generate it
for input in inputs:
path = Path.getcwd() / str(input)
if path.isfile():
FileSystem.parse_document(path, system)
else:
for document in path.walkfiles(pattern):
FileSystem.parse_document(document, system)
if use_cache:
cache[identifier] = system
return system
@staticmethod
def load_yaml(document: Path, required=False):
document = Path(document)
if not document.exists():
if required:
click.secho('yaml document does not exists: {0}'.format(document), fg='red')
return {}
try:
return yaml.load(document.text(), Loader=Loader)
except yaml.YAMLError as exc:
click.secho(exc, fg='red')
return {}
|