import yaml as pyyaml
-# To be loaded dynamically as needed
-jsonschema = None
-
-
class SpecException(Exception):
"""Netlink spec exception.
"""
mcast_groups dict of all multicast groups (index by name)
kernel_family dict of kernel family attributes
"""
+
+ # To be loaded dynamically as needed
+ jsonschema = None
+
def __init__(self, spec_path, schema_path=None, exclude_ops=None):
with open(spec_path, "r", encoding='utf-8') as stream:
prefix = '# SPDX-License-Identifier: '
if schema_path is None:
schema_path = os.path.dirname(os.path.dirname(spec_path)) + f'/{self.proto}.yaml'
if schema_path:
- global jsonschema
-
with open(schema_path, "r", encoding='utf-8') as stream:
schema = pyyaml.safe_load(stream)
- if jsonschema is None:
- jsonschema = importlib.import_module("jsonschema")
+ if SpecFamily.jsonschema is None:
+ SpecFamily.jsonschema = importlib.import_module("jsonschema")
- jsonschema.validate(self.yaml, schema)
+ SpecFamily.jsonschema.validate(self.yaml, schema)
self.attr_sets = collections.OrderedDict()
self.sub_msgs = collections.OrderedDict()
yield from self.msgs
-genl_family_name_to_id = None
-
-
def _genl_msg(nl_type, nl_flags, genl_cmd, genl_version, seq=None):
# we prepend length in _genl_msg_finalize()
if seq is None:
# pylint: disable=too-many-nested-blocks
def _genl_load_families():
+ genl_family_name_to_id = {}
+
with socket.socket(socket.AF_NETLINK, socket.SOCK_RAW, Netlink.NETLINK_GENERIC) as sock:
sock.setsockopt(Netlink.SOL_NETLINK, Netlink.NETLINK_CAP_ACK, 1)
sock.send(msg, 0)
- global genl_family_name_to_id
- genl_family_name_to_id = dict()
-
while True:
reply = sock.recv(128 * 1024)
nms = NlMsgs(reply)
for nl_msg in nms:
if nl_msg.error:
- print("Netlink error:", nl_msg.error)
- return
+ raise YnlException(f"Netlink error: {nl_msg.error}")
if nl_msg.done:
- return
+ return genl_family_name_to_id
gm = GenlMsg(nl_msg)
fam = {}
class GenlProtocol(NetlinkProtocol):
+ genl_family_name_to_id = {}
+
def __init__(self, family_name):
super().__init__(family_name, Netlink.NETLINK_GENERIC)
- global genl_family_name_to_id
- if genl_family_name_to_id is None:
- _genl_load_families()
+ if not GenlProtocol.genl_family_name_to_id:
+ GenlProtocol.genl_family_name_to_id = _genl_load_families()
- self.genl_family = genl_family_name_to_id[family_name]
- self.family_id = genl_family_name_to_id[family_name]['id']
+ self.genl_family = GenlProtocol.genl_family_name_to_id[family_name]
+ self.family_id = GenlProtocol.genl_family_name_to_id[family_name]['id']
def message(self, flags, command, version, seq=None):
nlmsg = self._message(self.family_id, flags, seq)