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
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
|
"""
QAPI introspection generator
Copyright (C) 2015-2021 Red Hat, Inc.
Authors:
Markus Armbruster <armbru@redhat.com>
John Snow <jsnow@redhat.com>
This work is licensed under the terms of the GNU GPL, version 2.
See the COPYING file in the top-level directory.
"""
from typing import (
Any,
Dict,
Generic,
Iterable,
List,
Optional,
Sequence,
Tuple,
TypeVar,
Union,
)
from .common import (
c_name,
gen_endif,
gen_if,
mcgen,
)
from .gen import QAPISchemaMonolithicCVisitor
from .schema import (
QAPISchema,
QAPISchemaArrayType,
QAPISchemaBuiltinType,
QAPISchemaEntity,
QAPISchemaEnumMember,
QAPISchemaFeature,
QAPISchemaObjectType,
QAPISchemaObjectTypeMember,
QAPISchemaType,
QAPISchemaVariant,
QAPISchemaVariants,
)
from .source import QAPISourceInfo
# This module constructs a tree data structure that is used to
# generate the introspection information for QEMU. It is shaped
# like a JSON value.
#
# A complexity over JSON is that our values may or may not be annotated.
#
# Un-annotated values may be:
# Scalar: str, bool, None.
# Non-scalar: List, Dict
# _value = Union[str, bool, None, Dict[str, JSONValue], List[JSONValue]]
#
# With optional annotations, the type of all values is:
# JSONValue = Union[_Value, Annotated[_Value]]
#
# Sadly, mypy does not support recursive types; so the _Stub alias is used to
# mark the imprecision in the type model where we'd otherwise use JSONValue.
_Stub = Any
_Scalar = Union[str, bool, None]
_NonScalar = Union[Dict[str, _Stub], List[_Stub]]
_Value = Union[_Scalar, _NonScalar]
JSONValue = Union[_Value, 'Annotated[_Value]']
# These types are based on structures defined in QEMU's schema, so we
# lack precise types for them here. Python 3.6 does not offer
# TypedDict constructs, so they are broadly typed here as simple
# Python Dicts.
SchemaInfo = Dict[str, object]
SchemaInfoObject = Dict[str, object]
SchemaInfoObjectVariant = Dict[str, object]
SchemaInfoObjectMember = Dict[str, object]
SchemaInfoCommand = Dict[str, object]
_ValueT = TypeVar('_ValueT', bound=_Value)
class Annotated(Generic[_ValueT]):
"""
Annotated generally contains a SchemaInfo-like type (as a dict),
But it also used to wrap comments/ifconds around scalar leaf values,
for the benefit of features and enums.
"""
# TODO: Remove after Python 3.7 adds @dataclass:
# pylint: disable=too-few-public-methods
def __init__(self, value: _ValueT, ifcond: Iterable[str],
comment: Optional[str] = None):
self.value = value
self.comment: Optional[str] = comment
self.ifcond: Tuple[str, ...] = tuple(ifcond)
def _tree_to_qlit(obj: JSONValue,
level: int = 0,
dict_value: bool = False) -> str:
"""
Convert the type tree into a QLIT C string, recursively.
:param obj: The value to convert.
This value may not be Annotated when dict_value is True.
:param level: The indentation level for this particular value.
:param dict_value: True when the value being processed belongs to a
dict key; which suppresses the output indent.
"""
def indent(level: int) -> str:
return level * 4 * ' '
if isinstance(obj, Annotated):
# NB: _tree_to_qlit is called recursively on the values of a
# key:value pair; those values can't be decorated with
# comments or conditionals.
msg = "dict values cannot have attached comments or if-conditionals."
assert not dict_value, msg
ret = ''
if obj.comment:
ret += indent(level) + f"/* {obj.comment} */\n"
if obj.ifcond:
ret += gen_if(obj.ifcond)
ret += _tree_to_qlit(obj.value, level)
if obj.ifcond:
ret += '\n' + gen_endif(obj.ifcond)
return ret
ret = ''
if not dict_value:
ret += indent(level)
# Scalars:
if obj is None:
ret += 'QLIT_QNULL'
elif isinstance(obj, str):
ret += f"QLIT_QSTR({to_c_string(obj)})"
elif isinstance(obj, bool):
ret += f"QLIT_QBOOL({str(obj).lower()})"
# Non-scalars:
elif isinstance(obj, list):
ret += 'QLIT_QLIST(((QLitObject[]) {\n'
for value in obj:
ret += _tree_to_qlit(value, level + 1).strip('\n') + '\n'
ret += indent(level + 1) + '{}\n'
ret += indent(level) + '}))'
elif isinstance(obj, dict):
ret += 'QLIT_QDICT(((QLitDictEntry[]) {\n'
for key, value in sorted(obj.items()):
ret += indent(level + 1) + "{{ {:s}, {:s} }},\n".format(
to_c_string(key),
_tree_to_qlit(value, level + 1, dict_value=True)
)
ret += indent(level + 1) + '{}\n'
ret += indent(level) + '}))'
else:
raise NotImplementedError(
f"type '{type(obj).__name__}' not implemented"
)
if level > 0:
ret += ','
return ret
def to_c_string(string: str) -> str:
return '"' + string.replace('\\', r'\\').replace('"', r'\"') + '"'
class QAPISchemaGenIntrospectVisitor(QAPISchemaMonolithicCVisitor):
def __init__(self, prefix: str, unmask: bool):
super().__init__(
prefix, 'qapi-introspect',
' * QAPI/QMP schema introspection', __doc__)
self._unmask = unmask
self._schema: Optional[QAPISchema] = None
self._trees: List[Annotated[SchemaInfo]] = []
self._used_types: List[QAPISchemaType] = []
self._name_map: Dict[str, str] = {}
self._genc.add(mcgen('''
#include "qemu/osdep.h"
#include "%(prefix)sqapi-introspect.h"
''',
prefix=prefix))
def visit_begin(self, schema: QAPISchema) -> None:
self._schema = schema
def visit_end(self) -> None:
# visit the types that are actually used
for typ in self._used_types:
typ.visit(self)
# generate C
name = c_name(self._prefix, protect=False) + 'qmp_schema_qlit'
self._genh.add(mcgen('''
#include "qapi/qmp/qlit.h"
extern const QLitObject %(c_name)s;
''',
c_name=c_name(name)))
self._genc.add(mcgen('''
const QLitObject %(c_name)s = %(c_string)s;
''',
c_name=c_name(name),
c_string=_tree_to_qlit(self._trees)))
self._schema = None
self._trees = []
self._used_types = []
self._name_map = {}
def visit_needed(self, entity: QAPISchemaEntity) -> bool:
# Ignore types on first pass; visit_end() will pick up used types
return not isinstance(entity, QAPISchemaType)
def _name(self, name: str) -> str:
if self._unmask:
return name
if name not in self._name_map:
self._name_map[name] = '%d' % len(self._name_map)
return self._name_map[name]
def _use_type(self, typ: QAPISchemaType) -> str:
assert self._schema is not None
# Map the various integer types to plain int
if typ.json_type() == 'int':
typ = self._schema.lookup_type('int')
elif (isinstance(typ, QAPISchemaArrayType) and
typ.element_type.json_type() == 'int'):
typ = self._schema.lookup_type('intList')
# Add type to work queue if new
if typ not in self._used_types:
self._used_types.append(typ)
# Clients should examine commands and events, not types. Hide
# type names as integers to reduce the temptation. Also, it
# saves a few characters on the wire.
if isinstance(typ, QAPISchemaBuiltinType):
return typ.name
if isinstance(typ, QAPISchemaArrayType):
return '[' + self._use_type(typ.element_type) + ']'
return self._name(typ.name)
@staticmethod
def _gen_features(features: Sequence[QAPISchemaFeature]
) -> List[Annotated[str]]:
return [Annotated(f.name, f.ifcond) for f in features]
def _gen_tree(self, name: str, mtype: str, obj: Dict[str, object],
ifcond: Sequence[str] = (),
features: Sequence[QAPISchemaFeature] = ()) -> None:
"""
Build and append a SchemaInfo object to self._trees.
:param name: The SchemaInfo's name.
:param mtype: The SchemaInfo's meta-type.
:param obj: Additional SchemaInfo members, as appropriate for
the meta-type.
:param ifcond: Conditionals to apply to the SchemaInfo.
:param features: The SchemaInfo's features.
Will be omitted from the output if empty.
"""
comment: Optional[str] = None
if mtype not in ('command', 'event', 'builtin', 'array'):
if not self._unmask:
# Output a comment to make it easy to map masked names
# back to the source when reading the generated output.
comment = f'"{self._name(name)}" = {name}'
name = self._name(name)
obj['name'] = name
obj['meta-type'] = mtype
if features:
obj['features'] = self._gen_features(features)
self._trees.append(Annotated(obj, ifcond, comment))
def _gen_member(self, member: QAPISchemaObjectTypeMember
) -> Annotated[SchemaInfoObjectMember]:
obj: SchemaInfoObjectMember = {
'name': member.name,
'type': self._use_type(member.type)
}
if member.optional:
obj['default'] = None
if member.features:
obj['features'] = self._gen_features(member.features)
return Annotated(obj, member.ifcond)
def _gen_variant(self, variant: QAPISchemaVariant
) -> Annotated[SchemaInfoObjectVariant]:
obj: SchemaInfoObjectVariant = {
'case': variant.name,
'type': self._use_type(variant.type)
}
return Annotated(obj, variant.ifcond)
def visit_builtin_type(self, name: str, info: Optional[QAPISourceInfo],
json_type: str) -> None:
self._gen_tree(name, 'builtin', {'json-type': json_type})
def visit_enum_type(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str],
features: List[QAPISchemaFeature],
members: List[QAPISchemaEnumMember],
prefix: Optional[str]) -> None:
self._gen_tree(
name, 'enum',
{'values': [Annotated(m.name, m.ifcond) for m in members]},
ifcond, features
)
def visit_array_type(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str],
element_type: QAPISchemaType) -> None:
element = self._use_type(element_type)
self._gen_tree('[' + element + ']', 'array', {'element-type': element},
ifcond)
def visit_object_type_flat(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str],
features: List[QAPISchemaFeature],
members: List[QAPISchemaObjectTypeMember],
variants: Optional[QAPISchemaVariants]) -> None:
obj: SchemaInfoObject = {
'members': [self._gen_member(m) for m in members]
}
if variants:
obj['tag'] = variants.tag_member.name
obj['variants'] = [self._gen_variant(v) for v in variants.variants]
self._gen_tree(name, 'object', obj, ifcond, features)
def visit_alternate_type(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str],
features: List[QAPISchemaFeature],
variants: QAPISchemaVariants) -> None:
self._gen_tree(
name, 'alternate',
{'members': [Annotated({'type': self._use_type(m.type)},
m.ifcond)
for m in variants.variants]},
ifcond, features
)
def visit_command(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str],
features: List[QAPISchemaFeature],
arg_type: Optional[QAPISchemaObjectType],
ret_type: Optional[QAPISchemaType], gen: bool,
success_response: bool, boxed: bool, allow_oob: bool,
allow_preconfig: bool, coroutine: bool) -> None:
assert self._schema is not None
arg_type = arg_type or self._schema.the_empty_object_type
ret_type = ret_type or self._schema.the_empty_object_type
obj: SchemaInfoCommand = {
'arg-type': self._use_type(arg_type),
'ret-type': self._use_type(ret_type)
}
if allow_oob:
obj['allow-oob'] = allow_oob
self._gen_tree(name, 'command', obj, ifcond, features)
def visit_event(self, name: str, info: Optional[QAPISourceInfo],
ifcond: Sequence[str], features: List[QAPISchemaFeature],
arg_type: Optional[QAPISchemaObjectType],
boxed: bool) -> None:
assert self._schema is not None
arg_type = arg_type or self._schema.the_empty_object_type
self._gen_tree(name, 'event', {'arg-type': self._use_type(arg_type)},
ifcond, features)
def gen_introspect(schema: QAPISchema, output_dir: str, prefix: str,
opt_unmask: bool) -> None:
vis = QAPISchemaGenIntrospectVisitor(prefix, opt_unmask)
schema.visit(vis)
vis.write(output_dir)
|