# -*- coding: utf-8 -*-
#
# QAPI schema internal representation
#
# Copyright (c) 2015-2019 Red Hat Inc.
#
# Authors:
#  Markus Armbruster <armbru@redhat.com>
#  Eric Blake <eblake@redhat.com>
#  Marc-André Lureau <marcandre.lureau@redhat.com>
#
# This work is licensed under the terms of the GNU GPL, version 2.
# See the COPYING file in the top-level directory.

# pylint: disable=too-many-lines

# TODO catching name collisions in generated code would be nice

from __future__ import annotations

from abc import ABC, abstractmethod
from collections import OrderedDict
import os
import re
from typing import (
    Any,
    Callable,
    Dict,
    List,
    Optional,
    Union,
    ValuesView,
    cast,
)

from .common import (
    POINTER_SUFFIX,
    c_name,
    cgen_ifcond,
    docgen_ifcond,
    gen_endif,
    gen_if,
)
from .error import QAPIError, QAPISemError, QAPISourceError
from .expr import check_exprs
from .parser import QAPIDoc, QAPIExpression, QAPISchemaParser
from .source import QAPISourceInfo


class QAPISchemaIfCond:
    def __init__(
        self,
        ifcond: Optional[Union[str, Dict[str, object]]] = None,
    ) -> None:
        self.ifcond = ifcond

    def _cgen(self) -> str:
        return cgen_ifcond(self.ifcond)

    def gen_if(self) -> str:
        return gen_if(self._cgen())

    def gen_endif(self) -> str:
        return gen_endif(self._cgen())

    def docgen(self) -> str:
        return docgen_ifcond(self.ifcond)

    def is_present(self) -> bool:
        return bool(self.ifcond)


class QAPISchemaEntity:
    """
    A schema entity.

    This is either a directive, such as include, or a definition.
    The latter uses sub-class `QAPISchemaDefinition`.
    """
    def __init__(self, info: Optional[QAPISourceInfo]):
        self._module: Optional[QAPISchemaModule] = None
        # For explicitly defined entities, info points to the (explicit)
        # definition.  For builtins (and their arrays), info is None.
        # For implicitly defined entities, info points to a place that
        # triggered the implicit definition (there may be more than one
        # such place).
        self.info = info
        self._checked = False

    def __repr__(self) -> str:
        return "<%s at 0x%x>" % (type(self).__name__, id(self))

    def check(self, schema: QAPISchema) -> None:
        # pylint: disable=unused-argument
        self._checked = True

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        pass

    def _set_module(
        self, schema: QAPISchema, info: Optional[QAPISourceInfo]
    ) -> None:
        assert self._checked
        fname = info.fname if info else QAPISchemaModule.BUILTIN_MODULE_NAME
        self._module = schema.module_by_fname(fname)
        self._module.add_entity(self)

    def set_module(self, schema: QAPISchema) -> None:
        self._set_module(schema, self.info)

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        # pylint: disable=unused-argument
        assert self._checked


class QAPISchemaDefinition(QAPISchemaEntity):
    meta: str

    def __init__(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        doc: Optional[QAPIDoc],
        ifcond: Optional[QAPISchemaIfCond] = None,
        features: Optional[List[QAPISchemaFeature]] = None,
    ):
        super().__init__(info)
        for f in features or []:
            f.set_defined_in(name)
        self.name = name
        self.doc = doc
        self._ifcond = ifcond or QAPISchemaIfCond()
        self.features = features or []

    def __repr__(self) -> str:
        return "<%s:%s at 0x%x>" % (type(self).__name__, self.name,
                                    id(self))

    def c_name(self) -> str:
        return c_name(self.name)

    def check(self, schema: QAPISchema) -> None:
        assert not self._checked
        super().check(schema)
        seen: Dict[str, QAPISchemaMember] = {}
        for f in self.features:
            f.check_clash(self.info, seen)

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        if doc:
            for f in self.features:
                doc.connect_feature(f)

    @property
    def ifcond(self) -> QAPISchemaIfCond:
        assert self._checked
        return self._ifcond

    def is_implicit(self) -> bool:
        return not self.info

    def describe(self) -> str:
        return "%s '%s'" % (self.meta, self.name)


class QAPISchemaVisitor:
    def visit_begin(self, schema: QAPISchema) -> None:
        pass

    def visit_end(self) -> None:
        pass

    def visit_module(self, name: str) -> None:
        pass

    def visit_needed(self, entity: QAPISchemaEntity) -> bool:
        # pylint: disable=unused-argument
        # Default to visiting everything
        return True

    def visit_include(self, name: str, info: Optional[QAPISourceInfo]) -> None:
        pass

    def visit_builtin_type(
        self, name: str, info: Optional[QAPISourceInfo], json_type: str
    ) -> None:
        pass

    def visit_enum_type(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        members: List[QAPISchemaEnumMember],
        prefix: Optional[str],
    ) -> None:
        pass

    def visit_array_type(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        element_type: QAPISchemaType,
    ) -> None:
        pass

    def visit_object_type(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        base: Optional[QAPISchemaObjectType],
        members: List[QAPISchemaObjectTypeMember],
        branches: Optional[QAPISchemaBranches],
    ) -> None:
        pass

    def visit_object_type_flat(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        members: List[QAPISchemaObjectTypeMember],
        branches: Optional[QAPISchemaBranches],
    ) -> None:
        pass

    def visit_alternate_type(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        alternatives: QAPISchemaAlternatives,
    ) -> None:
        pass

    def visit_command(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        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:
        pass

    def visit_event(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        arg_type: Optional[QAPISchemaObjectType],
        boxed: bool,
    ) -> None:
        pass


class QAPISchemaModule:

    BUILTIN_MODULE_NAME = './builtin'

    def __init__(self, name: str):
        self.name = name
        self._entity_list: List[QAPISchemaEntity] = []

    @staticmethod
    def is_system_module(name: str) -> bool:
        """
        System modules are internally defined modules.

        Their names start with the "./" prefix.
        """
        return name.startswith('./')

    @classmethod
    def is_user_module(cls, name: str) -> bool:
        """
        User modules are those defined by the user in qapi JSON files.

        They do not start with the "./" prefix.
        """
        return not cls.is_system_module(name)

    @classmethod
    def is_builtin_module(cls, name: str) -> bool:
        """
        The built-in module is a single System module for the built-in types.

        It is always "./builtin".
        """
        return name == cls.BUILTIN_MODULE_NAME

    def add_entity(self, ent: QAPISchemaEntity) -> None:
        self._entity_list.append(ent)

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        visitor.visit_module(self.name)
        for entity in self._entity_list:
            if visitor.visit_needed(entity):
                entity.visit(visitor)


class QAPISchemaInclude(QAPISchemaEntity):
    def __init__(self, sub_module: QAPISchemaModule, info: QAPISourceInfo):
        super().__init__(info)
        self._sub_module = sub_module

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_include(self._sub_module.name, self.info)


class QAPISchemaType(QAPISchemaDefinition, ABC):
    # Return the C type for common use.
    # For the types we commonly box, this is a pointer type.
    @abstractmethod
    def c_type(self) -> str:
        pass

    # Return the C type to be used in a parameter list.
    def c_param_type(self) -> str:
        return self.c_type()

    # Return the C type to be used where we suppress boxing.
    def c_unboxed_type(self) -> str:
        return self.c_type()

    @abstractmethod
    def json_type(self) -> str:
        pass

    def alternate_qtype(self) -> Optional[str]:
        json2qtype = {
            'null':    'QTYPE_QNULL',
            'string':  'QTYPE_QSTRING',
            'number':  'QTYPE_QNUM',
            'int':     'QTYPE_QNUM',
            'boolean': 'QTYPE_QBOOL',
            'array':   'QTYPE_QLIST',
            'object':  'QTYPE_QDICT'
        }
        return json2qtype.get(self.json_type())

    def doc_type(self) -> Optional[str]:
        if self.is_implicit():
            return None
        return self.name

    def need_has_if_optional(self) -> bool:
        # When FOO is a pointer, has_FOO == !!FOO, i.e. has_FOO is redundant.
        # Except for arrays; see QAPISchemaArrayType.need_has_if_optional().
        return not self.c_type().endswith(POINTER_SUFFIX)

    def check(self, schema: QAPISchema) -> None:
        super().check(schema)
        for feat in self.features:
            if feat.is_special():
                raise QAPISemError(
                    self.info,
                    f"feature '{feat.name}' is not supported for types")

    def describe(self) -> str:
        return "%s type '%s'" % (self.meta, self.name)


class QAPISchemaBuiltinType(QAPISchemaType):
    meta = 'built-in'

    def __init__(self, name: str, json_type: str, c_type: str):
        super().__init__(name, None, None)
        assert json_type in ('string', 'number', 'int', 'boolean', 'null',
                             'value')
        self._json_type_name = json_type
        self._c_type_name = c_type

    def c_name(self) -> str:
        return self.name

    def c_type(self) -> str:
        return self._c_type_name

    def c_param_type(self) -> str:
        if self.name == 'str':
            return 'const ' + self._c_type_name
        return self._c_type_name

    def json_type(self) -> str:
        return self._json_type_name

    def doc_type(self) -> str:
        return self.json_type()

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_builtin_type(self.name, self.info, self.json_type())


class QAPISchemaEnumType(QAPISchemaType):
    meta = 'enum'

    def __init__(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        doc: Optional[QAPIDoc],
        ifcond: Optional[QAPISchemaIfCond],
        features: Optional[List[QAPISchemaFeature]],
        members: List[QAPISchemaEnumMember],
        prefix: Optional[str],
    ):
        super().__init__(name, info, doc, ifcond, features)
        for m in members:
            m.set_defined_in(name)
        self.members = members
        self.prefix = prefix

    def check(self, schema: QAPISchema) -> None:
        super().check(schema)
        seen: Dict[str, QAPISchemaMember] = {}
        for m in self.members:
            m.check_clash(self.info, seen)

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        for m in self.members:
            m.connect_doc(doc)

    def is_implicit(self) -> bool:
        # See QAPISchema._def_predefineds()
        return self.name == 'QType'

    def c_type(self) -> str:
        return c_name(self.name)

    def member_names(self) -> List[str]:
        return [m.name for m in self.members]

    def json_type(self) -> str:
        return 'string'

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_enum_type(
            self.name, self.info, self.ifcond, self.features,
            self.members, self.prefix)


class QAPISchemaArrayType(QAPISchemaType):
    meta = 'array'

    def __init__(
        self, name: str, info: Optional[QAPISourceInfo], element_type: str
    ):
        super().__init__(name, info, None)
        self._element_type_name = element_type
        self.element_type: QAPISchemaType

    def need_has_if_optional(self) -> bool:
        # When FOO is an array, we still need has_FOO to distinguish
        # absent (!has_FOO) from present and empty (has_FOO && !FOO).
        return True

    def check(self, schema: QAPISchema) -> None:
        super().check(schema)
        self.element_type = schema.resolve_type(
            self._element_type_name, self.info,
            self.info.defn_meta if self.info else None)
        assert not isinstance(self.element_type, QAPISchemaArrayType)

    def set_module(self, schema: QAPISchema) -> None:
        self._set_module(schema, self.element_type.info)

    @property
    def ifcond(self) -> QAPISchemaIfCond:
        assert self._checked
        return self.element_type.ifcond

    def is_implicit(self) -> bool:
        return True

    def c_type(self) -> str:
        return c_name(self.name) + POINTER_SUFFIX

    def json_type(self) -> str:
        return 'array'

    def doc_type(self) -> Optional[str]:
        elt_doc_type = self.element_type.doc_type()
        if not elt_doc_type:
            return None
        return 'array of ' + elt_doc_type

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_array_type(self.name, self.info, self.ifcond,
                                 self.element_type)

    def describe(self) -> str:
        return "%s type ['%s']" % (self.meta, self._element_type_name)


class QAPISchemaObjectType(QAPISchemaType):
    def __init__(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        doc: Optional[QAPIDoc],
        ifcond: Optional[QAPISchemaIfCond],
        features: Optional[List[QAPISchemaFeature]],
        base: Optional[str],
        local_members: List[QAPISchemaObjectTypeMember],
        branches: Optional[QAPISchemaBranches],
    ):
        # struct has local_members, optional base, and no branches
        # union has base, branches, and no local_members
        super().__init__(name, info, doc, ifcond, features)
        self.meta = 'union' if branches else 'struct'
        for m in local_members:
            m.set_defined_in(name)
        if branches is not None:
            branches.set_defined_in(name)
        self._base_name = base
        self.base = None
        self.local_members = local_members
        self.branches = branches
        self.members: List[QAPISchemaObjectTypeMember]
        self._check_complete = False

    def check(self, schema: QAPISchema) -> None:
        # This calls another type T's .check() exactly when the C
        # struct emitted by gen_object() contains that T's C struct
        # (pointers don't count).
        if self._check_complete:
            # A previous .check() completed: nothing to do
            return
        if self._checked:
            # Recursed: C struct contains itself
            raise QAPISemError(self.info,
                               "object %s contains itself" % self.name)

        super().check(schema)
        assert self._checked and not self._check_complete

        seen = OrderedDict()
        if self._base_name:
            self.base = schema.resolve_type(self._base_name, self.info,
                                            "'base'")
            if (not isinstance(self.base, QAPISchemaObjectType)
                    or self.base.branches):
                raise QAPISemError(
                    self.info,
                    "'base' requires a struct type, %s isn't"
                    % self.base.describe())
            self.base.check(schema)
            self.base.check_clash(self.info, seen)
        for m in self.local_members:
            m.check(schema)
            m.check_clash(self.info, seen)

        # self.check_clash() works in terms of the supertype, but
        # self.members is declared List[QAPISchemaObjectTypeMember].
        # Cast down to the subtype.
        members = cast(List[QAPISchemaObjectTypeMember], list(seen.values()))

        if self.branches:
            self.branches.check(schema, seen)
            self.branches.check_clash(self.info, seen)

        self.members = members
        self._check_complete = True  # mark completed

    # Check that the members of this type do not cause duplicate JSON members,
    # and update seen to track the members seen so far. Report any errors
    # on behalf of info, which is not necessarily self.info
    def check_clash(
        self,
        info: Optional[QAPISourceInfo],
        seen: Dict[str, QAPISchemaMember],
    ) -> None:
        assert self._checked
        for m in self.members:
            m.check_clash(info, seen)
        if self.branches:
            self.branches.check_clash(info, seen)

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        if self.base and self.base.is_implicit():
            self.base.connect_doc(doc)
        for m in self.local_members:
            m.connect_doc(doc)

    def is_implicit(self) -> bool:
        # See QAPISchema._make_implicit_object_type(), as well as
        # _def_predefineds()
        return self.name.startswith('q_')

    def is_empty(self) -> bool:
        return not self.members and not self.branches

    def has_conditional_members(self) -> bool:
        return any(m.ifcond.is_present() for m in self.members)

    def c_name(self) -> str:
        assert self.name != 'q_empty'
        return super().c_name()

    def c_type(self) -> str:
        assert not self.is_implicit()
        return c_name(self.name) + POINTER_SUFFIX

    def c_unboxed_type(self) -> str:
        return c_name(self.name)

    def json_type(self) -> str:
        return 'object'

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_object_type(
            self.name, self.info, self.ifcond, self.features,
            self.base, self.local_members, self.branches)
        visitor.visit_object_type_flat(
            self.name, self.info, self.ifcond, self.features,
            self.members, self.branches)


class QAPISchemaAlternateType(QAPISchemaType):
    meta = 'alternate'

    def __init__(
        self,
        name: str,
        info: QAPISourceInfo,
        doc: Optional[QAPIDoc],
        ifcond: Optional[QAPISchemaIfCond],
        features: List[QAPISchemaFeature],
        alternatives: QAPISchemaAlternatives,
    ):
        super().__init__(name, info, doc, ifcond, features)
        assert alternatives.tag_member
        alternatives.set_defined_in(name)
        alternatives.tag_member.set_defined_in(self.name)
        self.alternatives = alternatives

    def check(self, schema: QAPISchema) -> None:
        super().check(schema)
        self.alternatives.tag_member.check(schema)
        # Not calling self.alternatives.check_clash(), because there's
        # nothing to clash with
        self.alternatives.check(schema, {})
        # Alternate branch names have no relation to the tag enum values;
        # so we have to check for potential name collisions ourselves.
        seen: Dict[str, QAPISchemaMember] = {}
        types_seen: Dict[str, str] = {}
        for v in self.alternatives.variants:
            v.check_clash(self.info, seen)
            qtype = v.type.alternate_qtype()
            if not qtype:
                raise QAPISemError(
                    self.info,
                    "%s cannot use %s"
                    % (v.describe(self.info), v.type.describe()))
            conflicting = set([qtype])
            if qtype == 'QTYPE_QSTRING':
                if isinstance(v.type, QAPISchemaEnumType):
                    for m in v.type.members:
                        if m.name in ['on', 'off']:
                            conflicting.add('QTYPE_QBOOL')
                        if re.match(r'[-+0-9.]', m.name):
                            # lazy, could be tightened
                            conflicting.add('QTYPE_QNUM')
                else:
                    conflicting.add('QTYPE_QNUM')
                    conflicting.add('QTYPE_QBOOL')
            for qt in conflicting:
                if qt in types_seen:
                    raise QAPISemError(
                        self.info,
                        "%s can't be distinguished from '%s'"
                        % (v.describe(self.info), types_seen[qt]))
                types_seen[qt] = v.name

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        for v in self.alternatives.variants:
            v.connect_doc(doc)

    def c_type(self) -> str:
        return c_name(self.name) + POINTER_SUFFIX

    def json_type(self) -> str:
        return 'value'

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_alternate_type(
            self.name, self.info, self.ifcond, self.features,
            self.alternatives)


class QAPISchemaVariants:
    def __init__(
        self,
        info: QAPISourceInfo,
        variants: List[QAPISchemaVariant],
    ):
        self.info = info
        self.tag_member: QAPISchemaObjectTypeMember
        self.variants = variants

    def set_defined_in(self, name: str) -> None:
        for v in self.variants:
            v.set_defined_in(name)

    # pylint: disable=unused-argument
    def check(
            self, schema: QAPISchema, seen: Dict[str, QAPISchemaMember]
    ) -> None:
        for v in self.variants:
            v.check(schema)


class QAPISchemaBranches(QAPISchemaVariants):
    def __init__(self,
                 info: QAPISourceInfo,
                 variants: List[QAPISchemaVariant],
                 tag_name: str):
        super().__init__(info, variants)
        self._tag_name = tag_name

    def check(
            self, schema: QAPISchema, seen: Dict[str, QAPISchemaMember]
    ) -> None:
        # We need to narrow the member type:
        tag_member = seen.get(c_name(self._tag_name))
        assert (tag_member is None
                or isinstance(tag_member, QAPISchemaObjectTypeMember))

        base = "'base'"
        # Pointing to the base type when not implicit would be
        # nice, but we don't know it here
        if not tag_member or self._tag_name != tag_member.name:
            raise QAPISemError(
                self.info,
                "discriminator '%s' is not a member of %s"
                % (self._tag_name, base))
        self.tag_member = tag_member
        # Here we do:
        assert tag_member.defined_in
        base_type = schema.lookup_type(tag_member.defined_in)
        assert base_type
        if not base_type.is_implicit():
            base = "base type '%s'" % tag_member.defined_in
        if not isinstance(tag_member.type, QAPISchemaEnumType):
            raise QAPISemError(
                self.info,
                "discriminator member '%s' of %s must be of enum type"
                % (self._tag_name, base))
        if tag_member.optional:
            raise QAPISemError(
                self.info,
                "discriminator member '%s' of %s must not be optional"
                % (self._tag_name, base))
        if tag_member.ifcond.is_present():
            raise QAPISemError(
                self.info,
                "discriminator member '%s' of %s must not be conditional"
                % (self._tag_name, base))
        # branches that are not explicitly covered get an empty type
        assert tag_member.defined_in
        cases = {v.name for v in self.variants}
        for m in tag_member.type.members:
            if m.name not in cases:
                v = QAPISchemaVariant(m.name, self.info,
                                      'q_empty', m.ifcond)
                v.set_defined_in(tag_member.defined_in)
                self.variants.append(v)
        if not self.variants:
            raise QAPISemError(self.info, "union has no branches")
        for v in self.variants:
            v.check(schema)
            # Union names must match enum values; alternate names are
            # checked separately. Use 'seen' to tell the two apart.
            if seen:
                if v.name not in tag_member.type.member_names():
                    raise QAPISemError(
                        self.info,
                        "branch '%s' is not a value of %s"
                        % (v.name, tag_member.type.describe()))
                if not isinstance(v.type, QAPISchemaObjectType):
                    raise QAPISemError(
                        self.info,
                        "%s cannot use %s"
                        % (v.describe(self.info), v.type.describe()))
                v.type.check(schema)

    def check_clash(
        self,
        info: Optional[QAPISourceInfo],
        seen: Dict[str, QAPISchemaMember],
    ) -> None:
        for v in self.variants:
            # Reset seen map for each variant, since qapi names from one
            # branch do not affect another branch.
            #
            # v.type's typing is enforced in check() above.
            assert isinstance(v.type, QAPISchemaObjectType)
            v.type.check_clash(info, dict(seen))


class QAPISchemaAlternatives(QAPISchemaVariants):
    def __init__(self,
                 info: QAPISourceInfo,
                 variants: List[QAPISchemaVariant],
                 tag_member: QAPISchemaObjectTypeMember):
        super().__init__(info, variants)
        self.tag_member = tag_member

    def check(
            self, schema: QAPISchema, seen: Dict[str, QAPISchemaMember]
    ) -> None:
        super().check(schema, seen)
        assert isinstance(self.tag_member.type, QAPISchemaEnumType)
        assert not self.tag_member.optional
        assert not self.tag_member.ifcond.is_present()


class QAPISchemaMember:
    """ Represents object members, enum members and features """
    role = 'member'

    def __init__(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: Optional[QAPISchemaIfCond] = None,
    ):
        self.name = name
        self.info = info
        self.ifcond = ifcond or QAPISchemaIfCond()
        self.defined_in: Optional[str] = None

    def set_defined_in(self, name: str) -> None:
        assert not self.defined_in
        self.defined_in = name

    def check_clash(
        self,
        info: Optional[QAPISourceInfo],
        seen: Dict[str, QAPISchemaMember],
    ) -> None:
        cname = c_name(self.name)
        if cname in seen:
            raise QAPISemError(
                info,
                "%s collides with %s"
                % (self.describe(info), seen[cname].describe(info)))
        seen[cname] = self

    def connect_doc(self, doc: Optional[QAPIDoc]) -> None:
        if doc:
            doc.connect_member(self)

    def describe(self, info: Optional[QAPISourceInfo]) -> str:
        role = self.role
        meta = 'type'
        defined_in = self.defined_in
        assert defined_in

        if defined_in.startswith('q_obj_'):
            # See QAPISchema._make_implicit_object_type() - reverse the
            # mapping there to create a nice human-readable description
            defined_in = defined_in[6:]
            if defined_in.endswith('-arg'):
                # Implicit type created for a command's dict 'data'
                assert role == 'member'
                role = 'parameter'
                meta = 'command'
                defined_in = defined_in[:-4]
            elif defined_in.endswith('-base'):
                # Implicit type created for a union's dict 'base'
                role = 'base ' + role
                defined_in = defined_in[:-5]
            else:
                assert False

        assert info is not None
        if defined_in != info.defn_name:
            return "%s '%s' of %s '%s'" % (role, self.name, meta, defined_in)
        return "%s '%s'" % (role, self.name)


class QAPISchemaEnumMember(QAPISchemaMember):
    role = 'value'

    def __init__(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        ifcond: Optional[QAPISchemaIfCond] = None,
        features: Optional[List[QAPISchemaFeature]] = None,
    ):
        super().__init__(name, info, ifcond)
        for f in features or []:
            f.set_defined_in(name)
        self.features = features or []

    def connect_doc(self, doc: Optional[QAPIDoc]) -> None:
        super().connect_doc(doc)
        if doc:
            for f in self.features:
                doc.connect_feature(f)


class QAPISchemaFeature(QAPISchemaMember):
    role = 'feature'

    # Features which are standardized across all schemas
    SPECIAL_NAMES = ['deprecated', 'unstable']

    def is_special(self) -> bool:
        return self.name in QAPISchemaFeature.SPECIAL_NAMES


class QAPISchemaObjectTypeMember(QAPISchemaMember):
    def __init__(
        self,
        name: str,
        info: QAPISourceInfo,
        typ: str,
        optional: bool,
        ifcond: Optional[QAPISchemaIfCond] = None,
        features: Optional[List[QAPISchemaFeature]] = None,
    ):
        super().__init__(name, info, ifcond)
        for f in features or []:
            f.set_defined_in(name)
        self._type_name = typ
        self.type: QAPISchemaType  # set during check()
        self.optional = optional
        self.features = features or []

    def need_has(self) -> bool:
        return self.optional and self.type.need_has_if_optional()

    def check(self, schema: QAPISchema) -> None:
        assert self.defined_in
        self.type = schema.resolve_type(self._type_name, self.info,
                                        self.describe)
        seen: Dict[str, QAPISchemaMember] = {}
        for f in self.features:
            f.check_clash(self.info, seen)

    def connect_doc(self, doc: Optional[QAPIDoc]) -> None:
        super().connect_doc(doc)
        if doc:
            for f in self.features:
                doc.connect_feature(f)


class QAPISchemaVariant(QAPISchemaObjectTypeMember):
    role = 'branch'

    def __init__(
        self,
        name: str,
        info: QAPISourceInfo,
        typ: str,
        ifcond: QAPISchemaIfCond,
    ):
        super().__init__(name, info, typ, False, ifcond)


class QAPISchemaCommand(QAPISchemaDefinition):
    meta = 'command'

    def __init__(
        self,
        name: str,
        info: QAPISourceInfo,
        doc: Optional[QAPIDoc],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        arg_type: Optional[str],
        ret_type: Optional[str],
        gen: bool,
        success_response: bool,
        boxed: bool,
        allow_oob: bool,
        allow_preconfig: bool,
        coroutine: bool,
    ):
        super().__init__(name, info, doc, ifcond, features)
        self._arg_type_name = arg_type
        self.arg_type: Optional[QAPISchemaObjectType] = None
        self._ret_type_name = ret_type
        self.ret_type: Optional[QAPISchemaType] = None
        self.gen = gen
        self.success_response = success_response
        self.boxed = boxed
        self.allow_oob = allow_oob
        self.allow_preconfig = allow_preconfig
        self.coroutine = coroutine

    def check(self, schema: QAPISchema) -> None:
        assert self.info is not None
        super().check(schema)
        if self._arg_type_name:
            arg_type = schema.resolve_type(
                self._arg_type_name, self.info, "command's 'data'")
            if not isinstance(arg_type, QAPISchemaObjectType):
                raise QAPISemError(
                    self.info,
                    "command's 'data' cannot take %s"
                    % arg_type.describe())
            self.arg_type = arg_type
            if self.arg_type.branches and not self.boxed:
                raise QAPISemError(
                    self.info,
                    "command's 'data' can take %s only with 'boxed': true"
                    % self.arg_type.describe())
            self.arg_type.check(schema)
            if self.arg_type.has_conditional_members() and not self.boxed:
                raise QAPISemError(
                    self.info,
                    "conditional command arguments require 'boxed': true")
        if self._ret_type_name:
            self.ret_type = schema.resolve_type(
                self._ret_type_name, self.info, "command's 'returns'")
            if self.name not in self.info.pragma.command_returns_exceptions:
                typ = self.ret_type
                if isinstance(typ, QAPISchemaArrayType):
                    typ = typ.element_type
                if not isinstance(typ, QAPISchemaObjectType):
                    raise QAPISemError(
                        self.info,
                        "command's 'returns' cannot take %s"
                        % self.ret_type.describe())

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        if doc:
            if self.arg_type and self.arg_type.is_implicit():
                self.arg_type.connect_doc(doc)

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_command(
            self.name, self.info, self.ifcond, self.features,
            self.arg_type, self.ret_type, self.gen, self.success_response,
            self.boxed, self.allow_oob, self.allow_preconfig,
            self.coroutine)


class QAPISchemaEvent(QAPISchemaDefinition):
    meta = 'event'

    def __init__(
        self,
        name: str,
        info: QAPISourceInfo,
        doc: Optional[QAPIDoc],
        ifcond: QAPISchemaIfCond,
        features: List[QAPISchemaFeature],
        arg_type: Optional[str],
        boxed: bool,
    ):
        super().__init__(name, info, doc, ifcond, features)
        self._arg_type_name = arg_type
        self.arg_type: Optional[QAPISchemaObjectType] = None
        self.boxed = boxed

    def check(self, schema: QAPISchema) -> None:
        super().check(schema)
        if self._arg_type_name:
            typ = schema.resolve_type(
                self._arg_type_name, self.info, "event's 'data'")
            if not isinstance(typ, QAPISchemaObjectType):
                raise QAPISemError(
                    self.info,
                    "event's 'data' cannot take %s"
                    % typ.describe())
            self.arg_type = typ
            if self.arg_type.branches and not self.boxed:
                raise QAPISemError(
                    self.info,
                    "event's 'data' can take %s only with 'boxed': true"
                    % self.arg_type.describe())
            self.arg_type.check(schema)
            if self.arg_type.has_conditional_members() and not self.boxed:
                raise QAPISemError(
                    self.info,
                    "conditional event arguments require 'boxed': true")

    def connect_doc(self, doc: Optional[QAPIDoc] = None) -> None:
        super().connect_doc(doc)
        doc = doc or self.doc
        if doc:
            if self.arg_type and self.arg_type.is_implicit():
                self.arg_type.connect_doc(doc)

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        super().visit(visitor)
        visitor.visit_event(
            self.name, self.info, self.ifcond, self.features,
            self.arg_type, self.boxed)


class QAPISchema:
    def __init__(self, fname: str):
        self.fname = fname

        try:
            parser = QAPISchemaParser(fname)
        except OSError as err:
            raise QAPIError(
                f"can't read schema file '{fname}': {err.strerror}"
            ) from err

        exprs = check_exprs(parser.exprs)
        self.docs = parser.docs
        self._entity_list: List[QAPISchemaEntity] = []
        self._entity_dict: Dict[str, QAPISchemaDefinition] = {}
        self._module_dict: Dict[str, QAPISchemaModule] = OrderedDict()
        # NB, values in the dict will identify the first encountered
        # usage of a named feature only
        self._feature_dict: Dict[str, QAPISchemaFeature] = OrderedDict()

        # All schemas get the names defined in the QapiSpecialFeature enum.
        # Rely on dict iteration order matching insertion order so that
        # the special names are emitted first when generating code.
        for f in QAPISchemaFeature.SPECIAL_NAMES:
            self._feature_dict[f] = QAPISchemaFeature(f, None)

        self._schema_dir = os.path.dirname(fname)
        self._make_module(QAPISchemaModule.BUILTIN_MODULE_NAME)
        self._make_module(fname)
        self._predefining = True
        self._def_predefineds()
        self._predefining = False
        self._def_exprs(exprs)
        self.check()

    def features(self) -> ValuesView[QAPISchemaFeature]:
        return self._feature_dict.values()

    def _def_entity(self, ent: QAPISchemaEntity) -> None:
        self._entity_list.append(ent)

    def _def_definition(self, defn: QAPISchemaDefinition) -> None:
        # Only the predefined types are allowed to not have info
        assert defn.info or self._predefining
        self._def_entity(defn)
        # TODO reject names that differ only in '_' vs. '.'  vs. '-',
        # because they're liable to clash in generated C.
        other_defn = self._entity_dict.get(defn.name)
        if other_defn:
            if other_defn.info:
                where = QAPISourceError(other_defn.info, "previous definition")
                raise QAPISemError(
                    defn.info,
                    "'%s' is already defined\n%s" % (defn.name, where))
            raise QAPISemError(
                defn.info, "%s is already defined" % other_defn.describe())
        self._entity_dict[defn.name] = defn

    def lookup_entity(self, name: str) -> Optional[QAPISchemaEntity]:
        return self._entity_dict.get(name)

    def lookup_type(self, name: str) -> Optional[QAPISchemaType]:
        typ = self.lookup_entity(name)
        if isinstance(typ, QAPISchemaType):
            return typ
        return None

    def resolve_type(
        self,
        name: str,
        info: Optional[QAPISourceInfo],
        what: Union[None, str, Callable[[QAPISourceInfo], str]],
    ) -> QAPISchemaType:
        typ = self.lookup_type(name)
        if not typ:
            assert info and what  # built-in types must not fail lookup
            if callable(what):
                what = what(info)
            raise QAPISemError(
                info, "%s uses unknown type '%s'" % (what, name))
        return typ

    def _module_name(self, fname: str) -> str:
        if QAPISchemaModule.is_system_module(fname):
            return fname
        return os.path.relpath(fname, self._schema_dir)

    def _make_module(self, fname: str) -> QAPISchemaModule:
        name = self._module_name(fname)
        if name not in self._module_dict:
            self._module_dict[name] = QAPISchemaModule(name)
        return self._module_dict[name]

    def module_by_fname(self, fname: str) -> QAPISchemaModule:
        name = self._module_name(fname)
        return self._module_dict[name]

    def _def_include(self, expr: QAPIExpression) -> None:
        include = expr['include']
        assert expr.doc is None
        self._def_entity(
            QAPISchemaInclude(self._make_module(include), expr.info))

    def _def_builtin_type(
        self, name: str, json_type: str, c_type: str
    ) -> None:
        self._def_definition(QAPISchemaBuiltinType(name, json_type, c_type))
        # Instantiating only the arrays that are actually used would
        # be nice, but we can't as long as their generated code
        # (qapi-builtin-types.[ch]) may be shared by some other
        # schema.
        self._make_array_type(name, None)

    def _def_predefineds(self) -> None:
        for t in [('str',    'string',  'char' + POINTER_SUFFIX),
                  ('number', 'number',  'double'),
                  ('int',    'int',     'int64_t'),
                  ('int8',   'int',     'int8_t'),
                  ('int16',  'int',     'int16_t'),
                  ('int32',  'int',     'int32_t'),
                  ('int64',  'int',     'int64_t'),
                  ('uint8',  'int',     'uint8_t'),
                  ('uint16', 'int',     'uint16_t'),
                  ('uint32', 'int',     'uint32_t'),
                  ('uint64', 'int',     'uint64_t'),
                  ('size',   'int',     'uint64_t'),
                  ('bool',   'boolean', 'bool'),
                  ('any',    'value',   'QObject' + POINTER_SUFFIX),
                  ('null',   'null',    'QNull' + POINTER_SUFFIX)]:
            self._def_builtin_type(*t)
        self.the_empty_object_type = QAPISchemaObjectType(
            'q_empty', None, None, None, None, None, [], None)
        self._def_definition(self.the_empty_object_type)

        qtypes = ['none', 'qnull', 'qnum', 'qstring', 'qdict', 'qlist',
                  'qbool']
        qtype_values = self._make_enum_members(
            [{'name': n} for n in qtypes], None)

        self._def_definition(QAPISchemaEnumType(
            'QType', None, None, None, None, qtype_values, None))

    def _make_features(
        self,
        features: Optional[List[Dict[str, Any]]],
        info: Optional[QAPISourceInfo],
    ) -> List[QAPISchemaFeature]:
        if features is None:
            return []

        for f in features:
            feat = QAPISchemaFeature(f['name'], info)
            if feat.name not in self._feature_dict:
                self._feature_dict[feat.name] = feat

        return [QAPISchemaFeature(f['name'], info,
                                  QAPISchemaIfCond(f.get('if')))
                for f in features]

    def _make_enum_member(
        self,
        name: str,
        ifcond: Optional[Union[str, Dict[str, Any]]],
        features: Optional[List[Dict[str, Any]]],
        info: Optional[QAPISourceInfo],
    ) -> QAPISchemaEnumMember:
        return QAPISchemaEnumMember(name, info,
                                    QAPISchemaIfCond(ifcond),
                                    self._make_features(features, info))

    def _make_enum_members(
        self, values: List[Dict[str, Any]], info: Optional[QAPISourceInfo]
    ) -> List[QAPISchemaEnumMember]:
        return [self._make_enum_member(v['name'], v.get('if'),
                                       v.get('features'), info)
                for v in values]

    def _make_array_type(
        self, element_type: str, info: Optional[QAPISourceInfo]
    ) -> str:
        name = element_type + 'List'    # reserved by check_defn_name_str()
        if not self.lookup_type(name):
            self._def_definition(QAPISchemaArrayType(
                name, info, element_type))
        return name

    def _make_implicit_object_type(
        self,
        name: str,
        info: QAPISourceInfo,
        ifcond: QAPISchemaIfCond,
        role: str,
        members: List[QAPISchemaObjectTypeMember],
    ) -> Optional[str]:
        if not members:
            return None
        # See also QAPISchemaObjectTypeMember.describe()
        name = 'q_obj_%s-%s' % (name, role)
        typ = self.lookup_entity(name)
        if typ:
            assert isinstance(typ, QAPISchemaObjectType)
            # The implicit object type has multiple users.  This can
            # only be a duplicate definition, which will be flagged
            # later.
        else:
            self._def_definition(QAPISchemaObjectType(
                name, info, None, ifcond, None, None, members, None))
        return name

    def _def_enum_type(self, expr: QAPIExpression) -> None:
        name = expr['enum']
        data = expr['data']
        prefix = expr.get('prefix')
        ifcond = QAPISchemaIfCond(expr.get('if'))
        info = expr.info
        features = self._make_features(expr.get('features'), info)
        self._def_definition(QAPISchemaEnumType(
            name, info, expr.doc, ifcond, features,
            self._make_enum_members(data, info), prefix))

    def _make_member(
        self,
        name: str,
        typ: Union[List[str], str],
        ifcond: QAPISchemaIfCond,
        features: Optional[List[Dict[str, Any]]],
        info: QAPISourceInfo,
    ) -> QAPISchemaObjectTypeMember:
        optional = False
        if name.startswith('*'):
            name = name[1:]
            optional = True
        if isinstance(typ, list):
            assert len(typ) == 1
            typ = self._make_array_type(typ[0], info)
        return QAPISchemaObjectTypeMember(name, info, typ, optional, ifcond,
                                          self._make_features(features, info))

    def _make_members(
        self,
        data: Dict[str, Any],
        info: QAPISourceInfo,
    ) -> List[QAPISchemaObjectTypeMember]:
        return [self._make_member(key, value['type'],
                                  QAPISchemaIfCond(value.get('if')),
                                  value.get('features'), info)
                for (key, value) in data.items()]

    def _def_struct_type(self, expr: QAPIExpression) -> None:
        name = expr['struct']
        base = expr.get('base')
        data = expr['data']
        info = expr.info
        ifcond = QAPISchemaIfCond(expr.get('if'))
        features = self._make_features(expr.get('features'), info)
        self._def_definition(QAPISchemaObjectType(
            name, info, expr.doc, ifcond, features, base,
            self._make_members(data, info),
            None))

    def _make_variant(
        self,
        case: str,
        typ: str,
        ifcond: QAPISchemaIfCond,
        info: QAPISourceInfo,
    ) -> QAPISchemaVariant:
        if isinstance(typ, list):
            assert len(typ) == 1
            typ = self._make_array_type(typ[0], info)
        return QAPISchemaVariant(case, info, typ, ifcond)

    def _def_union_type(self, expr: QAPIExpression) -> None:
        name = expr['union']
        base = expr['base']
        tag_name = expr['discriminator']
        data = expr['data']
        assert isinstance(data, dict)
        info = expr.info
        ifcond = QAPISchemaIfCond(expr.get('if'))
        features = self._make_features(expr.get('features'), info)
        if isinstance(base, dict):
            base = self._make_implicit_object_type(
                name, info, ifcond,
                'base', self._make_members(base, info))
        variants = [
            self._make_variant(key, value['type'],
                               QAPISchemaIfCond(value.get('if')),
                               info)
            for (key, value) in data.items()]
        members: List[QAPISchemaObjectTypeMember] = []
        self._def_definition(
            QAPISchemaObjectType(name, info, expr.doc, ifcond, features,
                                 base, members,
                                 QAPISchemaBranches(
                                     info, variants, tag_name)))

    def _def_alternate_type(self, expr: QAPIExpression) -> None:
        name = expr['alternate']
        data = expr['data']
        assert isinstance(data, dict)
        ifcond = QAPISchemaIfCond(expr.get('if'))
        info = expr.info
        features = self._make_features(expr.get('features'), info)
        variants = [
            self._make_variant(key, value['type'],
                               QAPISchemaIfCond(value.get('if')),
                               info)
            for (key, value) in data.items()]
        tag_member = QAPISchemaObjectTypeMember('type', info, 'QType', False)
        self._def_definition(
            QAPISchemaAlternateType(
                name, info, expr.doc, ifcond, features,
                QAPISchemaAlternatives(info, variants, tag_member)))

    def _def_command(self, expr: QAPIExpression) -> None:
        name = expr['command']
        data = expr.get('data')
        rets = expr.get('returns')
        gen = expr.get('gen', True)
        success_response = expr.get('success-response', True)
        boxed = expr.get('boxed', False)
        allow_oob = expr.get('allow-oob', False)
        allow_preconfig = expr.get('allow-preconfig', False)
        coroutine = expr.get('coroutine', False)
        ifcond = QAPISchemaIfCond(expr.get('if'))
        info = expr.info
        features = self._make_features(expr.get('features'), info)
        if isinstance(data, OrderedDict):
            data = self._make_implicit_object_type(
                name, info, ifcond,
                'arg', self._make_members(data, info))
        if isinstance(rets, list):
            assert len(rets) == 1
            rets = self._make_array_type(rets[0], info)
        self._def_definition(
            QAPISchemaCommand(name, info, expr.doc, ifcond, features, data,
                              rets, gen, success_response, boxed, allow_oob,
                              allow_preconfig, coroutine))

    def _def_event(self, expr: QAPIExpression) -> None:
        name = expr['event']
        data = expr.get('data')
        boxed = expr.get('boxed', False)
        ifcond = QAPISchemaIfCond(expr.get('if'))
        info = expr.info
        features = self._make_features(expr.get('features'), info)
        if isinstance(data, OrderedDict):
            data = self._make_implicit_object_type(
                name, info, ifcond,
                'arg', self._make_members(data, info))
        self._def_definition(QAPISchemaEvent(name, info, expr.doc, ifcond,
                                             features, data, boxed))

    def _def_exprs(self, exprs: List[QAPIExpression]) -> None:
        for expr in exprs:
            if 'enum' in expr:
                self._def_enum_type(expr)
            elif 'struct' in expr:
                self._def_struct_type(expr)
            elif 'union' in expr:
                self._def_union_type(expr)
            elif 'alternate' in expr:
                self._def_alternate_type(expr)
            elif 'command' in expr:
                self._def_command(expr)
            elif 'event' in expr:
                self._def_event(expr)
            elif 'include' in expr:
                self._def_include(expr)
            else:
                assert False

    def check(self) -> None:
        for ent in self._entity_list:
            ent.check(self)
            ent.connect_doc()
        for ent in self._entity_list:
            ent.set_module(self)
        for doc in self.docs:
            doc.check()

        features = list(self._feature_dict.values())
        if len(features) > 64:
            raise QAPISemError(
                features[64].info,
                "Maximum of 64 schema features is permitted")

    def visit(self, visitor: QAPISchemaVisitor) -> None:
        visitor.visit_begin(self)
        for mod in self._module_dict.values():
            mod.visit(visitor)
        visitor.visit_end()