diff --git a/.github/workflows/build_test.yaml b/.github/workflows/build_test.yaml index 43b5370..e707e65 100644 --- a/.github/workflows/build_test.yaml +++ b/.github/workflows/build_test.yaml @@ -30,10 +30,12 @@ jobs: - name: Check code style run: | python3 -m pycodestyle ./lglpy + python3 -m pycodestyle ./generator - name: Check typing run: | python3 -m mypy ./lglpy + python3 -m mypy ./generator build-ubuntu-x64-clang: name: Ubuntu x64 Clang diff --git a/generator/generate_vulkan_common.py b/generator/generate_vulkan_common.py index fc18509..07d6976 100644 --- a/generator/generate_vulkan_common.py +++ b/generator/generate_vulkan_common.py @@ -28,10 +28,11 @@ from datetime import datetime import os import re -import shutil import sys +from typing import Optional, TextIO import xml.etree.ElementTree as ET + # These functions are not part of the layer implementation EXCLUDED_FUNCTIONS = { # Functions exposed by loader not the implementation @@ -114,7 +115,8 @@ 'VkQueue', } -class VersionMapping(dict): + +class VersionInfo(dict): ''' Create a mapping of function => required version. @@ -126,21 +128,24 @@ class VersionMapping(dict): possible extensions. ''' - def __init__(self, root): + def __init__(self, root: ET.Element): ''' Load all functions that are defined in the XML. + + Args: + root: The root of the Vulkan spec XML document. ''' self.load_api_core_versions(root) self.load_api_extensions(root) - def load_api_core_versions(self, root): + def load_api_core_versions(self, root: ET.Element): ''' Load all core API functions that should be added to the map. ''' # Find all the formally supported API feature groups features = root.findall('.//feature') for feature in features: - # Omit commands that are specific to VulkanSC + # Omit commands that are only used in VulkanSC apis = feature.get('api', 'vulkan').split(',') if 'vulkan' not in apis: continue @@ -155,14 +160,18 @@ def load_api_core_versions(self, root): assert fname not in self, fname self[fname] = [(api_version, None)] - def load_api_extensions(self, root): + def load_api_extensions(self, root: ET.Element): ''' Load all extension API functions that should be added to the map. + + Args: + root: The root of the Vulkan spec XML document. ''' # Find all the formally supported API feature groups extensions = root.findall('.//extension') for extension in extensions: ext_name = extension.get('name') + assert ext_name is not None # Omit extensions that are specific to VulkanSC apis = extension.get('supported', 'vulkan').split(',') @@ -170,12 +179,14 @@ def load_api_extensions(self, root): continue # Omit extensions that are not on the Android platform - platforms = extension.get('platform', None) + platform = extension.get('platform', None) is_android = True - if platforms: - platforms = platforms.split(',') - is_android = 'android' in platforms + platform_list = None + if platform: + platform_list = platform.split(',') + is_android = 'android' in platform_list + # TODO: Relax this and allow other operating systems if not is_android: continue @@ -195,15 +206,18 @@ def load_api_extensions(self, root): if fname not in self: self[fname] = [] - self[fname].append((ext_name, platforms)) + self[fname].append((ext_name, platform_list)) - def get_platform_define(self, command): + def get_platform_define(self, command: str) -> Optional[str]: ''' - Determine the platform define for a command mapping. + Determine the platform define needed for a command mapping. + + Args: + command: The name of the command to process. ''' mappings = self[command] - platforms = set() + platforms = set() # type: set[str] for mapping in mappings: # No platform always takes precedence @@ -225,8 +239,32 @@ def get_platform_define(self, command): class Command: + ''' + Parsed specification of a single Vulkan API entry point. + + Attributes: + name: Name of this command. + dispatch_type: Is this 'instance' or 'device' for dispatch? + rtype: Function return type. + params: Function parameter list. + alias: Optional alias that provides the actual specification. For + commands with an alias, the dispatch_type, return type, and params + will be None until the alias is resolved. + ''' + + def __init__(self, mapping: VersionInfo, root: ET.Element): + ''' + Create a new parsed Vulkan command entry. + + Args: + mapping: The version information for that added this command. + root: The XML node root for the command subtree. + ''' + self.alias = None + self.rtype = None + self.params = None + self.dispatch_type = None - def __init__(self, mapping, root): # Omit commands that are specific to VulkanSC apis = root.get('api', 'vulkan').split(',') if 'vulkan' not in apis: @@ -235,12 +273,13 @@ def __init__(self, mapping, root): # Function is an alias if len(root) == 0: self.name = root.get('name') + assert self.name is not None + self.alias = root.get('alias') + assert self.alias is not None # Function is a standalone specification else: - self.alias = None - # Extract the basic function prototype proto = root[0] text = proto.text.strip() if proto.text else '' @@ -250,6 +289,7 @@ def __init__(self, mapping, root): assert text == '' and tail == '', f'`{text}`, `{tail}`' rtype = proto[0] + assert rtype.text text = rtype.text.strip() tail = rtype.tail.strip() if rtype.tail else '' assert rtype.tag == 'type', ET.tostring(proto) @@ -257,6 +297,7 @@ def __init__(self, mapping, root): self.rtype = text fname = proto[1] + assert fname.text text = fname.text.strip() tail = fname.tail.strip() if fname.tail else '' assert fname.tag == 'name', ET.tostring(proto) @@ -281,8 +322,9 @@ def __init__(self, mapping, root): type_prefix = f'{text} ' if text else '' - ptype = node[0] valid_tail = ('', '*', '**', '* const*') + ptype = node[0] + assert ptype.text text = ptype.text.strip() tail = ptype.tail.strip() if ptype.tail else '' assert ptype.tag == 'type', ET.tostring(proto) @@ -290,16 +332,17 @@ def __init__(self, mapping, root): type_postfix = f'{tail}' if tail else '' - ptype = f'{type_prefix}{text}{type_postfix}' + ptype_str = f'{type_prefix}{text}{type_postfix}' pname = node[1] + assert pname.text text = pname.text.strip() tail = pname.tail.strip() if pname.tail else '' assert pname.tag == 'name', ET.tostring(proto) assert text assert tail == '' or re.match(r'^\[\d+]$', tail) - self.params.append((ptype, text, tail)) + self.params.append((ptype_str, text, tail)) # Filter out functions that are not dynamically dispatched if self.name in EXCLUDED_FUNCTIONS: @@ -311,19 +354,27 @@ def __init__(self, mapping, root): # Identify instance vs device functions if not self.alias: - dispatch_type = self.params[0][0] - if dispatch_type in INSTANCE_FUNCTION_PARAM_TYPE: + disp_type = self.params[0][0] + if disp_type in INSTANCE_FUNCTION_PARAM_TYPE: self.dispatch_type = 'instance' - elif dispatch_type in DEVICE_FUNCTION_PARAM_TYPE: + elif disp_type in DEVICE_FUNCTION_PARAM_TYPE: self.dispatch_type = 'device' + elif self.name.startswith('vkEnumerateInstance'): + self.dispatch_type = 'instance' else: - if self.name.startswith('vkEnumerateInstance'): - self.dispatch_type = 'instance' - else: - assert False, f'Unknown dispatch: {dispatch_type} {self.name}' + assert False, f'Unknown dispatch: {disp_type} {self.name}' + + +def load_template(path: str) -> str: + ''' + Load a template from the generator template directory. + Args: + path: The file name in the template directory. -def load_template(path): + Returns: + The loaded text template. + ''' base_dir = os.path.dirname(__file__) template_file = os.path.join(base_dir, 'vk_codegen', path) @@ -334,7 +385,13 @@ def load_template(path): return data -def write_copyright_header(file): +def write_copyright_header(file: TextIO) -> None: + ''' + Write the standard license and copyright header to the file. + + Args: + file: The file handle to write to. + ''' data = load_template('header.txt') start_year = 2024 @@ -350,8 +407,16 @@ def write_copyright_header(file): file.write('\n') -def generate_layer_instance_dispatch_table(file, mapping, commands): +def generate_instance_dispatch_table( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the instance dispatch table. + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -367,29 +432,30 @@ def generate_layer_instance_dispatch_table(file, mapping, commands): if command.dispatch_type != 'instance': continue - tname = command.name - if tname in NO_INTERCEPT_OR_DISPATCH_FUNCTIONS: + assert command.name + + if command.name in NO_INTERCEPT_OR_DISPATCH_FUNCTIONS: continue plat_define = mapping.get_platform_define(command.name) ttype = f'PFN_{command.name}' - if tname not in NO_INTERCEPT_FUNCTIONS: + if command.name not in NO_INTERCEPT_FUNCTIONS: if plat_define: itable_members.append(f'#if defined({plat_define})') - itable_members.append(f' ENTRY({tname}),') + itable_members.append(f' ENTRY({command.name}),') if plat_define: itable_members.append('#endif') - if tname not in NO_DISPATCH_FUNCTIONS: + if command.name not in NO_DISPATCH_FUNCTIONS: if plat_define: dispatch_table_members.append(f'#if defined({plat_define})') dispatch_table_inits.append(f'#if defined({plat_define})') - dispatch_table_members.append(f' {ttype} {tname};') - dispatch_table_inits.append(f' ENTRY({tname});') + dispatch_table_members.append(f' {ttype} {command.name};') + dispatch_table_inits.append(f' ENTRY({command.name});') if plat_define: dispatch_table_members.append('#endif') @@ -401,14 +467,14 @@ def generate_layer_instance_dispatch_table(file, mapping, commands): if command.dispatch_type != 'device': continue + assert command.name plat_define = mapping.get_platform_define(command.name) ttype = f'PFN_{command.name}' - tname = command.name if plat_define: itable_members.append(f'#if defined({plat_define})') - itable_members.append(f' ENTRY({tname}),') + itable_members.append(f' ENTRY({command.name}),') if plat_define: itable_members.append('#endif') @@ -419,8 +485,16 @@ def generate_layer_instance_dispatch_table(file, mapping, commands): file.write(data) -def generate_layer_instance_layer_decls(file, mapping, commands): +def generate_instance_decls( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the instance intercept declarations header. + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -436,6 +510,7 @@ def generate_layer_instance_layer_decls(file, mapping, commands): continue lines = [] + assert command.name plat_define = mapping.get_platform_define(command.name) if plat_define: @@ -444,7 +519,8 @@ def generate_layer_instance_layer_decls(file, mapping, commands): # Declare the default implementation lines.append('/* See Vulkan API for documentation. */') lines.append('/* Default common code pass-through implementation. */') - decl = f'VKAPI_ATTR {command.rtype} VKAPI_CALL layer_{command.name}_default(' + decl = f'VKAPI_ATTR {command.rtype} ' \ + f'VKAPI_CALL layer_{command.name}_default(' lines.append(decl) for i, (ptype, pname, array) in enumerate(command.params): @@ -482,8 +558,16 @@ def generate_layer_instance_layer_decls(file, mapping, commands): file.write('\n') -def generate_layer_instance_layer_defs(file, mapping, commands): +def generate_instance_defs( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the instance intercept definitions. + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -495,11 +579,11 @@ def generate_layer_instance_layer_defs(file, mapping, commands): if command.dispatch_type != 'instance': continue - tname = command.name - if tname in NO_INTERCEPT_FUNCTIONS: + assert command.name + if command.name in NO_INTERCEPT_FUNCTIONS: continue - if tname in CUSTOM_FUNCTIONS: + if command.name in CUSTOM_FUNCTIONS: continue plat_define = mapping.get_platform_define(command.name) @@ -507,7 +591,8 @@ def generate_layer_instance_layer_defs(file, mapping, commands): lines.append(f'#if defined({plat_define})\n') lines.append('/* See Vulkan API for documentation. */') - decl = f'VKAPI_ATTR {command.rtype} VKAPI_CALL layer_{command.name}_default(' + decl = f'VKAPI_ATTR {command.rtype} ' \ + f'VKAPI_CALL layer_{command.name}_default(' lines.append(decl) for i, (ptype, pname, array) in enumerate(command.params): @@ -541,8 +626,16 @@ def generate_layer_instance_layer_defs(file, mapping, commands): file.write(data) -def generate_layer_device_dispatch_table(file, mapping, commands): +def generate_device_dispatch_table( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the device dispatch table. + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -552,12 +645,13 @@ def generate_layer_device_dispatch_table(file, mapping, commands): itable_members = [] dispatch_table_members = [] dispatch_table_inits = [] + for command in commands: if command.dispatch_type != 'device': continue - tname = command.name - if tname in NO_INTERCEPT_OR_DISPATCH_FUNCTIONS: + assert command.name + if command.name in NO_INTERCEPT_OR_DISPATCH_FUNCTIONS: continue plat_define = mapping.get_platform_define(command.name) @@ -568,9 +662,9 @@ def generate_layer_device_dispatch_table(file, mapping, commands): dispatch_table_members.append(f'#if defined({plat_define})') dispatch_table_inits.append(f'#if defined({plat_define})') - itable_members.append(f' ENTRY({tname}),') - dispatch_table_members.append(f' {ttype} {tname};') - dispatch_table_inits.append(f' ENTRY({tname});') + itable_members.append(f' ENTRY({command.name}),') + dispatch_table_members.append(f' {ttype} {command.name};') + dispatch_table_inits.append(f' ENTRY({command.name});') if plat_define: itable_members.append('#endif') @@ -583,7 +677,16 @@ def generate_layer_device_dispatch_table(file, mapping, commands): file.write(data) -def generate_layer_device_layer_decls(file, mapping, commands): +def generate_device_decls( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the device intercept declarations header. + + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -599,15 +702,17 @@ def generate_layer_device_layer_decls(file, mapping, commands): if command.dispatch_type != 'device': continue - lines = [] + assert command.name + lines = [] plat_define = mapping.get_platform_define(command.name) if plat_define: lines.append(f'#if defined({plat_define})\n') lines.append('/* See Vulkan API for documentation. */') lines.append('/* Default common code pass-through implementation. */') - decl = f'VKAPI_ATTR {command.rtype} VKAPI_CALL layer_{command.name}_default(' + decl = f'VKAPI_ATTR {command.rtype} ' \ + f'VKAPI_CALL layer_{command.name}_default(' lines.append(decl) for i, (ptype, pname, array) in enumerate(command.params): @@ -645,8 +750,16 @@ def generate_layer_device_layer_decls(file, mapping, commands): file.write('\n') -def generate_layer_device_layer_defs(file, mapping, commands): +def generate_device_defs( + file: TextIO, mapping: VersionInfo, commands: list[Command]) -> None: + ''' + Generate the device intercept definitions. + Args: + file: The file to write. + mapping: The version mapping information for the commands. + commands: The list of commands read from the spec. + ''' # Write the copyright header to the file write_copyright_header(file) @@ -658,8 +771,8 @@ def generate_layer_device_layer_defs(file, mapping, commands): if command.dispatch_type != 'device': continue - tname = command.name - if tname in CUSTOM_FUNCTIONS: + assert command.name + if command.name in CUSTOM_FUNCTIONS: continue plat_define = mapping.get_platform_define(command.name) @@ -668,7 +781,8 @@ def generate_layer_device_layer_defs(file, mapping, commands): lines.append('/* See Vulkan API for documentation. */') - decl = f'VKAPI_ATTR {command.rtype} VKAPI_CALL layer_{command.name}_default(' + decl = f'VKAPI_ATTR {command.rtype} ' \ + f'VKAPI_CALL layer_{command.name}_default(' lines.append(decl) for i, (ptype, pname, array) in enumerate(command.params): @@ -702,13 +816,7 @@ def generate_layer_device_layer_defs(file, mapping, commands): file.write(data) -def copy_resource(src_dir, out_dir): - out_dir = os.path.abspath(out_dir) - os.makedirs(out_dir, exist_ok=True) - shutil.copytree(src_dir, out_dir, dirs_exist_ok=True) - - -def main(): +def main() -> int: ''' Tool main function. @@ -724,7 +832,7 @@ def main(): root = tree.getroot() # Parse function to API version or extension mapping - mapping = VersionMapping(root) + mapping = VersionInfo(root) # Parse function prototypes nodes = root.findall('.//commands/command') @@ -748,36 +856,33 @@ def main(): command.params = aliases[0].params command.dispatch_type = aliases[0].dispatch_type - # Remove the alias to avoid confusion in future - del command.alias - # Sort functions into alphabetical order - commands.sort(key=lambda x: x.name) + commands.sort(key=lambda x: str(x.name)) # Generate dynamic resources outfile = os.path.join(outdir, 'instance_dispatch_table.hpp') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_instance_dispatch_table(handle, mapping, commands) - - outfile = os.path.join(outdir, 'device_dispatch_table.hpp') - with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_device_dispatch_table(handle, mapping, commands) + generate_instance_dispatch_table(handle, mapping, commands) outfile = os.path.join(outdir, 'instance_functions.hpp') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_instance_layer_decls(handle, mapping, commands) + generate_instance_decls(handle, mapping, commands) outfile = os.path.join(outdir, 'instance_functions.cpp') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_instance_layer_defs(handle, mapping, commands) + generate_instance_defs(handle, mapping, commands) + + outfile = os.path.join(outdir, 'device_dispatch_table.hpp') + with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: + generate_device_dispatch_table(handle, mapping, commands) outfile = os.path.join(outdir, 'device_functions.hpp') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_device_layer_decls(handle, mapping, commands) + generate_device_decls(handle, mapping, commands) outfile = os.path.join(outdir, 'device_functions.cpp') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_device_layer_defs(handle, mapping, commands) + generate_device_defs(handle, mapping, commands) return 0 diff --git a/generator/generate_vulkan_layer.py b/generator/generate_vulkan_layer.py index d833fe5..8242423 100644 --- a/generator/generate_vulkan_layer.py +++ b/generator/generate_vulkan_layer.py @@ -26,284 +26,23 @@ ''' import argparse -from datetime import datetime import os import re import shutil import sys -import xml.etree.ElementTree as ET - -# These functions are manually created as fully exported entry points in the -# layer library, and are not part of the dynamic dispatch behavior -MANUAL_FUNCTIONS = { - # Exposed by loader so not handled by the layer - 'vkEnumerateInstanceVersion', - # Exposed as symbols managed by the loader so not handled by the layer - 'vkGetInstanceProcAddr', - 'vkGetDeviceProcAddr', - # Exposed by layer as explicit entry points - 'vkEnumerateDeviceLayerProperties', - 'vkEnumerateInstanceExtensionProperties', - 'vkEnumerateInstanceLayerProperties', -} - -# These functions are manually exported, but we can use driver forwarding -FORWARD_WITHOUT_INTERCEPT = { - 'vkEnumerateDeviceExtensionProperties', - 'vkEnumerateInstanceExtensionProperties', -} - -# These functions are found via the loader-injected chain info -INTERCEPT_WITHOUT_FORWARD = { - 'vkCreateDevice', - 'vkCreateInstance', -} - -# Filter out extensions from these vendors by default -EXTENSION_VENDOR_FILTER = { - 'AMD', - 'ANDROID', - 'GOOGLE', - 'HUAWEI', - 'IMG', - 'INTEL', - 'LUNARG', - 'MSFT', - 'NV', - 'NVX', - 'QCOM', - 'SEC', - 'VALVE', -} - -# Filter out KHR and EXT extensions that we don't support or need on Android -EXTENSION_NAME_FILTER = { - 'VK_KHR_video_queue', - 'VK_KHR_video_decode_queue', - 'VK_KHR_video_encode_queue', - 'VK_EXT_acquire_drm_display', - 'VK_EXT_headless_surface', -} - -INSTANCE_FUNCTION_PARAM_TYPE = { - 'const VkInstanceCreateInfo*', - 'VkInstance', - 'VkPhysicalDevice', -} - -DEVICE_FUNCTION_PARAM_TYPE = { - 'VkCommandBuffer', - 'VkDevice', - 'VkQueue', -} - -class VersionMapping(dict): - ''' - Create a mapping of function => required version. +from typing import TextIO - The required version is a list of either: - * An API version string ('1.0', '1.1', etc) - * An extension string (e.g. 'VK_KHR_swapchain') - The required version is a list because some APIs can be added by multiple - possible extensions. +def load_template(path: str) -> str: ''' + Load a template from the generator template directory. + + Args: + path: The file name in the template directory. - def __init__(self, root): - ''' - Load all functions that are defined in the XML. - ''' - self.load_api_core_versions(root) - self.load_api_extensions(root) - - def load_api_core_versions(self, root): - ''' - Load all core API functions that should be added to the map. - ''' - # Find all the formally supported API feature groups - features = root.findall('.//feature') - for feature in features: - # Omit commands that are specific to VulkanSC - apis = feature.get('api', 'vulkan').split(',') - if 'vulkan' not in apis: - continue - - # Get the API version - api_version = feature.get('number') - - # Get the commands added in this API version - commands = feature.findall('.//command') - for command in commands: - fname = command.get('name') - assert fname not in self, fname - self[fname] = [(api_version, None)] - - def load_api_extensions(self, root): - ''' - Load all extension API functions that should be added to the map. - ''' - # Find all the formally supported API feature groups - extensions = root.findall('.//extension') - for extension in extensions: - ext_name = extension.get('name') - - # Omit extensions that are specific to VulkanSC - apis = extension.get('supported', 'vulkan').split(',') - if 'vulkan' not in apis: - continue - - # Omit extensions that are not on the Android platform - platforms = extension.get('platform', None) - is_android = True - if platforms: - platforms = platforms.split(',') - is_android = 'android' in platforms - - if not is_android: - continue - - # Omit extensions that are from other vendors - vendor = ext_name.split("_")[1] - if vendor in EXTENSION_VENDOR_FILTER: - continue - - # Omit extensions that are explicitly not supported e.g. video - if ext_name in EXTENSION_NAME_FILTER: - continue - - # Note, not all extensions have commands - commands = extension.findall('.//command') - for command in commands: - fname = command.get('name') - if fname not in self: - self[fname] = [] - - self[fname].append((ext_name, platforms)) - - def get_platform_define(self, command): - ''' - Determine the platform define for a command mapping. - ''' - mappings = self[command] - - platforms = set() - - for mapping in mappings: - # No platform always takes precedence - if mapping[1] is None: - platforms = set() - break - - # Limited platforms accumulate - platforms.update(set(mapping[1])) - - # Currently only handle no define mapping or Android only - if not platforms: - return None - - if len(platforms) == 1 and 'android' in platforms: - return 'VK_USE_PLATFORM_ANDROID_KHR' - - assert False, f'Unhandled mapping: {platforms}' - - -class Command: - - def __init__(self, mapping, root): - # Omit commands that are specific to VulkanSC - apis = root.get('api', 'vulkan').split(',') - if 'vulkan' not in apis: - raise NotImplementedError - - # Function is an alias - if len(root) == 0: - self.name = root.get('name') - self.alias = root.get('alias') - - # Function is a standalone specification - else: - self.alias = None - - # Extract the basic function prototype - proto = root[0] - text = proto.text.strip() if proto.text else '' - tail = proto.tail.strip() if proto.tail else '' - assert proto.tag == 'proto', ET.tostring(proto) - assert len(proto) == 2, ET.tostring(proto) - assert text == '' and tail == '', f'`{text}`, `{tail}`' - - rtype = proto[0] - text = rtype.text.strip() - tail = rtype.tail.strip() if rtype.tail else '' - assert rtype.tag == 'type', ET.tostring(proto) - assert text and tail == '', f'`{text}`, `{tail}`' - self.rtype = text - - fname = proto[1] - text = fname.text.strip() - tail = fname.tail.strip() if fname.tail else '' - assert fname.tag == 'name', ET.tostring(proto) - assert text and tail == '', f'`{text}`, `{tail}`' - self.name = text - self.params = [] - - # Extract the function parameters - parameters = root.findall('./param') - for node in parameters: - assert len(node) == 2, ET.tostring(node) - - # Omit parameters that are specific to VulkanSC - apis = node.get('api', 'vulkan').split(',') - if 'vulkan' not in apis: - continue - - valid_text = ('', 'const', 'struct', 'const struct') - text = node.text.strip() if node.text else '' - tail = node.tail.strip() if node.tail else '' - assert text in valid_text and tail == '', f'`{text}`, `{tail}`' - - type_prefix = f'{text} ' if text else '' - - ptype = node[0] - valid_tail = ('', '*', '**', '* const*') - text = ptype.text.strip() - tail = ptype.tail.strip() if ptype.tail else '' - assert ptype.tag == 'type', ET.tostring(proto) - assert text and tail in valid_tail, f'`{text}`, `{tail}`' - - type_postfix = f'{tail}' if tail else '' - - ptype = f'{type_prefix}{text}{type_postfix}' - - pname = node[1] - text = pname.text.strip() - tail = pname.tail.strip() if pname.tail else '' - assert pname.tag == 'name', ET.tostring(proto) - assert text - assert tail == '' or re.match(r'^\[\d+]$', tail) - - self.params.append((ptype, text, tail)) - - # Filter out functions that are not dynamically dispatched - if self.name in MANUAL_FUNCTIONS: - raise NotImplementedError - - # Filter out functions that are not in our supported mapping - if self.name not in mapping: - raise NotImplementedError - - # Identify instance vs device functions - if not self.alias: - dispatch_type = self.params[0][0] - if dispatch_type in INSTANCE_FUNCTION_PARAM_TYPE: - self.dispatch_type = 'instance' - elif dispatch_type in DEVICE_FUNCTION_PARAM_TYPE: - self.dispatch_type = 'device' - else: - assert False, f'Unknown dispatch: {dispatch_type} {self.name}' - - -def load_template(path): + Returns: + The loaded text template. + ''' base_dir = os.path.dirname(__file__) template_file = os.path.join(base_dir, 'vk_codegen', path) @@ -314,75 +53,110 @@ def load_template(path): return data -def write_copyright_header(file): - data = load_template('header.txt') - - start_year = 2024 - end_year = datetime.now().year - if start_year == end_year: - date_string = f'{start_year}' - else: - date_string = f'{start_year}-{end_year}' +def get_layer_api_name(vendor: str, layer: str) -> str: + ''' + Generate the layer string name for the new layer used by the loader. + Names are of the form "VK_LAYER__" in upper case. - data = data.replace('{COPYRIGHT_YEARS}', date_string) + Args: + vendor: The layer vendor tag. + layer: The name of the layer. - file.write(data) - file.write('\n') + Returns: + The layer string name used by the loader. + ''' + pattern = re.compile(r'^VkLayer(\w+)$') + match = pattern.match(layer) + assert match + name_parts = ('VK_LAYER', vendor.upper(), match.group(1).upper()) + return '_'.join(name_parts) -def generate_layer_root_cmake(file, project_name, layer_name): - data = load_template('root_CMakeLists.txt') - data = data.replace('{PROJECT_NAME}', project_name) - data = data.replace('{LAYER_NAME}', layer_name) - file.write(data) +def get_layer_api_description(vendor: str, layer: str) -> str: + ''' + Generate the layer description placeholder to return from the layer + enumeration query in the API. We expect developers to replace this with + something more useful for long-lived layers. -def get_layer_api_name(vendor_name, layer_name): + Args: + vendor: The layer vendor tag. + layer: The name of the layer. - pattern = re.compile(r'^VkLayer(\w+)$') - match = pattern.match(layer_name) - assert match + Returns: + The layer string description returned by the API queries. + ''' + return f'{layer} by {vendor}' - name_parts = ('VK_LAYER', vendor_name.upper(), match.group(1).upper()) - return '_'.join(name_parts) +def generate_root_cmake(file: TextIO, project: str, layer: str) -> None: + ''' + Generate the root CMake file for a new layer. -def get_layer_api_description(vendor_name, layer_name): - return f'{layer_name} by {vendor_name}' + Args: + file: The file handle to write to. + project: The name of the CMake project. + layer: The name of the layer used as the Android log tag. + ''' + data = load_template('root_CMakeLists.txt') + data = data.replace('{PROJECT_NAME}', project) + data = data.replace('{LAYER_NAME}', layer) + file.write(data) -def generate_layer_source_cmake(file, vendor_name, layer_name): +def generate_source_cmake(file: TextIO, vendor: str, layer: str) -> None: + ''' + Generate the layer source directory CMake file for a new layer. + Args: + file: The file handle to write to. + vendor: The layer vendor tag. + layer: The name of the layer. + ''' data = load_template('source_CMakeLists.txt') - data = data.replace('{LAYER_NAME}', layer_name) + data = data.replace('{LAYER_NAME}', layer) - name = get_layer_api_name(vendor_name, layer_name) + name = get_layer_api_name(vendor, layer) data = data.replace('{LGL_LAYER_NAME}', name) - desc = get_layer_api_description(vendor_name, layer_name) + desc = get_layer_api_description(vendor, layer) data = data.replace('{LGL_LAYER_DESC}', desc) file.write(data) -def generate_layer_install_helper(file, vendor_name, layer_name): +def generate_install_helper(file: TextIO, vendor: str, layer: str) -> None: + ''' + Generate the Android installer helper with placeholders replaced. + Args: + file: The file handle to write to. + vendor: The layer vendor tag. + layer: The name of the layer. + ''' data = load_template('android_install.py') - data = data.replace('{LAYER_NAME}', layer_name) + data = data.replace('{LAYER_NAME}', layer) - name = get_layer_api_name(vendor_name, layer_name) + name = get_layer_api_name(vendor, layer) data = data.replace('{LGL_LAYER_NAME}', name) file.write(data) -def copy_resource(src_dir, out_dir): - out_dir = os.path.abspath(out_dir) - os.makedirs(out_dir, exist_ok=True) - shutil.copytree(src_dir, out_dir, dirs_exist_ok=True) +def copy_resource(src_dir: str, dst_dir: str) -> None: + ''' + Copy a directory of resources to another location. + + Args: + src_dir: The source directory location. + out_dir: The destination directory location. + ''' + dst_dir = os.path.abspath(dst_dir) + os.makedirs(dst_dir, exist_ok=True) + shutil.copytree(src_dir, dst_dir, dirs_exist_ok=True) -def parse_command_line(): +def parse_command_line() -> argparse.Namespace: ''' Parse the command line. @@ -395,10 +169,10 @@ def parse_command_line(): help='CMake project name') parser.add_argument('--vendor-name', default='LGL', - help='Layer vendor name (optional)') + help='layer vendor name (optional)') parser.add_argument('--layer-name', default=None, - help='Layer name (optional)') + help='layer name (optional)') parser.add_argument('--output', metavar='DIR', required=True, help='output directory for layer project') @@ -414,7 +188,7 @@ def parse_command_line(): return args -def main(): +def main() -> int: ''' Tool main function. @@ -426,7 +200,7 @@ def main(): # Validate the layer name is well formed pattern = re.compile(r'^VkLayer\w+$') if not pattern.match(args.layer_name): - print(f'ERROR: Layer name "{args.layer_name}" is invalid') + print(f'ERROR: Layer name "{args.layer_name}" must start with VkLayer') return 1 # Check that output directory is either empty or over-writable @@ -435,64 +209,31 @@ def main(): if not os.path.isdir(outdir): print(f'ERROR: Output location "{outdir}" is not a directory') return 1 + if len(os.listdir(outdir)) != 0 and not args.overwrite: print(f'ERROR: Output directory "{outdir}" is not empty') return 1 - # Parse the XML headers - tree = ET.parse('./khronos/vulkan/registry/vk.xml') - root = tree.getroot() - - # Parse function to API version or extension mapping - mapping = VersionMapping(root) - - # Parse function prototypes - nodes = root.findall('.//commands/command') - - commands = [] - for node in nodes: - # Parse, skipping commands for sibling APIs - try: - commands.append(Command(mapping, node)) - except NotImplementedError: - pass - - # Flatten out command alias cross-references - for command in commands: - alias_name = command.alias - if alias_name: - aliases = [x for x in commands if x.name == alias_name] - assert len(aliases) == 1 - - command.rtype = aliases[0].rtype - command.params = aliases[0].params - command.dispatch_type = aliases[0].dispatch_type - - # Remove the alias to avoid confusion in future - del command.alias - - # Sort functions into alphabetical order - commands.sort(key=lambda x: x.name) - # Generate static resources base_dir = os.path.dirname(__file__) source_dir = os.path.join(base_dir, 'vk_layer') copy_resource(source_dir, outdir) - # Generate the layer skeleton + # Generate templated resources outfile = os.path.join(outdir, 'CMakeLists.txt') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_root_cmake(handle, args.project_name, args.layer_name) + generate_root_cmake(handle, args.project_name, args.layer_name) outfile = os.path.join(outdir, 'source/CMakeLists.txt') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_source_cmake(handle, args.vendor_name, args.layer_name) + generate_source_cmake(handle, args.vendor_name, args.layer_name) outfile = os.path.join(outdir, 'android_install.py') with open(outfile, 'w', encoding='utf-8', newline='\n') as handle: - generate_layer_install_helper(handle, args.vendor_name, args.layer_name) + generate_install_helper(handle, args.vendor_name, args.layer_name) return 0 + if __name__ == '__main__': sys.exit(main()) diff --git a/generator/vk_codegen/android_install.py b/generator/vk_codegen/android_install.py index 1cd3d03..22d8ce0 100644 --- a/generator/vk_codegen/android_install.py +++ b/generator/vk_codegen/android_install.py @@ -30,6 +30,7 @@ import shlex import subprocess as sp import sys +from typing import Any, Optional # Android temp directory ANDROID_TMP_DIR = '/data/local/tmp/' @@ -38,6 +39,7 @@ EXPECTED_VULKAN_LAYER_NAME = '{LGL_LAYER_NAME}' EXPECTED_VULKAN_LAYER_FILE = 'lib{LAYER_NAME}.so' + class Device: ''' A basic wrapper around adb, allowing a specific device to be registered. @@ -46,7 +48,7 @@ class Device: device: The name of the device to call, or None for non-specific use. ''' - def adb_quiet(self, *args): + def adb_quiet(self, *args: str) -> None: ''' Call `adb` to run a command, but ignore output and errors. @@ -57,7 +59,7 @@ def adb_quiet(self, *args): commands.extend(args) sp.run(commands, stdout=sp.DEVNULL, stderr=sp.DEVNULL, check=False) - def adb(self, *args, **kwargs): + def adb(self, *args: str, **kwargs: Any) -> str: ''' Call `adb` to run command, and capture output and results. @@ -73,7 +75,7 @@ def adb(self, *args, **kwargs): Raises: CalledProcessError: The subprocess was not successfully executed. ''' - commands = ['adb'] + commands = ['adb'] # type: Any commands.extend(args) text = kwargs.get('text', True) @@ -84,12 +86,12 @@ def adb(self, *args, **kwargs): if shell: # Unix shells need a flattened command for shell commands if os.name != 'nt': - quotedCommands = [] + quoted_commands = [] for command in commands: if command != '>': command = shlex.quote(command) - quotedCommands.append(command) - commands = ' '.join(quotedCommands) + quoted_commands.append(command) + commands = ' '.join(quoted_commands) # Run on the device but with shell argument quoting if quote: @@ -101,11 +103,13 @@ def adb(self, *args, **kwargs): return rep.stdout - def adb_run_as(self, package, *args, quiet=False): + def adb_run_as(self, package: str, + *args: str, quiet: bool = False) -> Optional[str]: ''' Call `adb` to run command as a package using `run-as` or as root, if root is accessible. If command will be run as root, this function - will change CWD to the package data directory before executing the command. + will change CWD to the package data directory before executing the + command. Args: package: Package name to run-as or change CWD to. @@ -113,7 +117,7 @@ def adb_run_as(self, package, *args, quiet=False): quiet: If True, ignores output from adb. Returns: - The contents of stdout or Nothing, if quiet=True. + The contents of stdout or None if quiet=True. Raises: CalledProcessError: The subprocess was not successfully executed. @@ -122,12 +126,14 @@ def adb_run_as(self, package, *args, quiet=False): command.extend(args) if quiet: - return self.adb_quiet(*command) + self.adb_quiet(*command) + return None return self.adb(*command) -def enable_vulkan_debug_layer(device, package, layer): +def enable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Args: device: The device instance. @@ -138,16 +144,24 @@ def enable_vulkan_debug_layer(device, package, layer): print('\nInstalling Vulkan debug layer') layer = os.path.normpath(layer) - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) device.adb('push', layer, ANDROID_TMP_DIR) - device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layerBase, '.') - device.adb('shell', 'settings', 'put', 'global', 'enable_gpu_debug_layers', '1') - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_app', package) - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layer_base, '.') + + device.adb('shell', 'settings', 'put', 'global', + 'enable_gpu_debug_layers', '1') + + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_app', package) -def disable_vulkan_debug_layer(device, package, layer): + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + + +def disable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Clean up the Vulkan layer installation. @@ -157,35 +171,44 @@ def disable_vulkan_debug_layer(device, package, layer): ''' print('\nRemoving Vulkan debug layer') - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) + + device.adb('shell', 'settings', 'delete', 'global', + 'enable_gpu_debug_layers') + + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global', 'enable_gpu_debug_layers') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_layers') - device.adb_run_as(package, 'rm', layerBase, quiet=True) + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_layers') + device.adb_run_as(package, 'rm', layer_base, quiet=True) -def get_layer(): + +def get_layer() -> Optional[str]: ''' Find the debug layer to use in the build directory. + + Returns: + The part to the library to use. ''' base_dir = './build_arm64/source/' - sym_lib = None + # TODO: If we want to use symbolized layer we need to rename it lib = None for path in os.listdir(base_dir): + # Match symbolized library first so we don't use it if path.endswith('_sym.so'): - sym_lib = os.path.join(base_dir, path) + _ = os.path.join(base_dir, path) elif path.endswith('.so'): lib = os.path.join(base_dir, path) - # TODO: If we want to use symbolized layer we need to rename it return lib -def parse_command_line(): +def parse_command_line() -> argparse.Namespace: ''' Parse the command line. @@ -200,7 +223,7 @@ def parse_command_line(): return parser.parse_args() -def main(): +def main() -> int: ''' Script main function. @@ -211,6 +234,9 @@ def main(): device = Device() layer = get_layer() + if not layer: + print('ERROR: Layer binary not found') + return 1 enable_vulkan_debug_layer(device, args.package, layer) @@ -218,6 +244,9 @@ def main(): disable_vulkan_debug_layer(device, args.package, layer) + return 0 + + if __name__ == '__main__': try: sys.exit(main()) diff --git a/layer_example/android_install.py b/layer_example/android_install.py index 7ae37eb..a8a300f 100644 --- a/layer_example/android_install.py +++ b/layer_example/android_install.py @@ -30,6 +30,7 @@ import shlex import subprocess as sp import sys +from typing import Any, Optional # Android temp directory ANDROID_TMP_DIR = '/data/local/tmp/' @@ -38,6 +39,7 @@ EXPECTED_VULKAN_LAYER_NAME = 'VK_LAYER_LGL_EXAMPLE' EXPECTED_VULKAN_LAYER_FILE = 'libVkLayerExample.so' + class Device: ''' A basic wrapper around adb, allowing a specific device to be registered. @@ -46,7 +48,7 @@ class Device: device: The name of the device to call, or None for non-specific use. ''' - def adb_quiet(self, *args): + def adb_quiet(self, *args: str) -> None: ''' Call `adb` to run a command, but ignore output and errors. @@ -57,7 +59,7 @@ def adb_quiet(self, *args): commands.extend(args) sp.run(commands, stdout=sp.DEVNULL, stderr=sp.DEVNULL, check=False) - def adb(self, *args, **kwargs): + def adb(self, *args: str, **kwargs: Any) -> str: ''' Call `adb` to run command, and capture output and results. @@ -73,7 +75,7 @@ def adb(self, *args, **kwargs): Raises: CalledProcessError: The subprocess was not successfully executed. ''' - commands = ['adb'] + commands = ['adb'] # type: Any commands.extend(args) text = kwargs.get('text', True) @@ -84,12 +86,12 @@ def adb(self, *args, **kwargs): if shell: # Unix shells need a flattened command for shell commands if os.name != 'nt': - quotedCommands = [] + quoted_commands = [] for command in commands: if command != '>': command = shlex.quote(command) - quotedCommands.append(command) - commands = ' '.join(quotedCommands) + quoted_commands.append(command) + commands = ' '.join(quoted_commands) # Run on the device but with shell argument quoting if quote: @@ -101,11 +103,13 @@ def adb(self, *args, **kwargs): return rep.stdout - def adb_run_as(self, package, *args, quiet=False): + def adb_run_as(self, package: str, + *args: str, quiet: bool = False) -> Optional[str]: ''' Call `adb` to run command as a package using `run-as` or as root, if root is accessible. If command will be run as root, this function - will change CWD to the package data directory before executing the command. + will change CWD to the package data directory before executing the + command. Args: package: Package name to run-as or change CWD to. @@ -113,7 +117,7 @@ def adb_run_as(self, package, *args, quiet=False): quiet: If True, ignores output from adb. Returns: - The contents of stdout or Nothing, if quiet=True. + The contents of stdout or None if quiet=True. Raises: CalledProcessError: The subprocess was not successfully executed. @@ -122,12 +126,14 @@ def adb_run_as(self, package, *args, quiet=False): command.extend(args) if quiet: - return self.adb_quiet(*command) + self.adb_quiet(*command) + return None return self.adb(*command) -def enable_vulkan_debug_layer(device, package, layer): +def enable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Args: device: The device instance. @@ -138,16 +144,24 @@ def enable_vulkan_debug_layer(device, package, layer): print('\nInstalling Vulkan debug layer') layer = os.path.normpath(layer) - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) device.adb('push', layer, ANDROID_TMP_DIR) - device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layerBase, '.') - device.adb('shell', 'settings', 'put', 'global', 'enable_gpu_debug_layers', '1') - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_app', package) - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layer_base, '.') + + device.adb('shell', 'settings', 'put', 'global', + 'enable_gpu_debug_layers', '1') + + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_app', package) -def disable_vulkan_debug_layer(device, package, layer): + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + + +def disable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Clean up the Vulkan layer installation. @@ -157,35 +171,44 @@ def disable_vulkan_debug_layer(device, package, layer): ''' print('\nRemoving Vulkan debug layer') - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) + + device.adb('shell', 'settings', 'delete', 'global', + 'enable_gpu_debug_layers') + + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global', 'enable_gpu_debug_layers') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_layers') - device.adb_run_as(package, 'rm', layerBase, quiet=True) + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_layers') + device.adb_run_as(package, 'rm', layer_base, quiet=True) -def get_layer(): + +def get_layer() -> Optional[str]: ''' Find the debug layer to use in the build directory. + + Returns: + The part to the library to use. ''' base_dir = './build_arm64/source/' - sym_lib = None + # TODO: If we want to use symbolized layer we need to rename it lib = None for path in os.listdir(base_dir): + # Match symbolized library first so we don't use it if path.endswith('_sym.so'): - sym_lib = os.path.join(base_dir, path) + _ = os.path.join(base_dir, path) elif path.endswith('.so'): lib = os.path.join(base_dir, path) - # TODO: If we want to use symbolized layer we need to rename it return lib -def parse_command_line(): +def parse_command_line() -> argparse.Namespace: ''' Parse the command line. @@ -200,7 +223,7 @@ def parse_command_line(): return parser.parse_args() -def main(): +def main() -> int: ''' Script main function. @@ -211,6 +234,9 @@ def main(): device = Device() layer = get_layer() + if not layer: + print('ERROR: Layer binary not found') + return 1 enable_vulkan_debug_layer(device, args.package, layer) @@ -218,6 +244,9 @@ def main(): disable_vulkan_debug_layer(device, args.package, layer) + return 0 + + if __name__ == '__main__': try: sys.exit(main()) diff --git a/layer_gpu_timeline/android_install.py b/layer_gpu_timeline/android_install.py index dc1e062..e92abf7 100644 --- a/layer_gpu_timeline/android_install.py +++ b/layer_gpu_timeline/android_install.py @@ -30,6 +30,7 @@ import shlex import subprocess as sp import sys +from typing import Any, Optional # Android temp directory ANDROID_TMP_DIR = '/data/local/tmp/' @@ -38,6 +39,7 @@ EXPECTED_VULKAN_LAYER_NAME = 'VK_LAYER_LGL_GPUTIMELINE' EXPECTED_VULKAN_LAYER_FILE = 'libVkLayerGPUTimeline.so' + class Device: ''' A basic wrapper around adb, allowing a specific device to be registered. @@ -46,7 +48,7 @@ class Device: device: The name of the device to call, or None for non-specific use. ''' - def adb_quiet(self, *args): + def adb_quiet(self, *args: str) -> None: ''' Call `adb` to run a command, but ignore output and errors. @@ -57,7 +59,7 @@ def adb_quiet(self, *args): commands.extend(args) sp.run(commands, stdout=sp.DEVNULL, stderr=sp.DEVNULL, check=False) - def adb(self, *args, **kwargs): + def adb(self, *args: str, **kwargs: Any) -> str: ''' Call `adb` to run command, and capture output and results. @@ -73,7 +75,7 @@ def adb(self, *args, **kwargs): Raises: CalledProcessError: The subprocess was not successfully executed. ''' - commands = ['adb'] + commands = ['adb'] # type: Any commands.extend(args) text = kwargs.get('text', True) @@ -84,12 +86,12 @@ def adb(self, *args, **kwargs): if shell: # Unix shells need a flattened command for shell commands if os.name != 'nt': - quotedCommands = [] + quoted_commands = [] for command in commands: if command != '>': command = shlex.quote(command) - quotedCommands.append(command) - commands = ' '.join(quotedCommands) + quoted_commands.append(command) + commands = ' '.join(quoted_commands) # Run on the device but with shell argument quoting if quote: @@ -101,11 +103,13 @@ def adb(self, *args, **kwargs): return rep.stdout - def adb_run_as(self, package, *args, quiet=False): + def adb_run_as(self, package: str, + *args: str, quiet: bool = False) -> Optional[str]: ''' Call `adb` to run command as a package using `run-as` or as root, if root is accessible. If command will be run as root, this function - will change CWD to the package data directory before executing the command. + will change CWD to the package data directory before executing the + command. Args: package: Package name to run-as or change CWD to. @@ -113,7 +117,7 @@ def adb_run_as(self, package, *args, quiet=False): quiet: If True, ignores output from adb. Returns: - The contents of stdout or Nothing, if quiet=True. + The contents of stdout or None if quiet=True. Raises: CalledProcessError: The subprocess was not successfully executed. @@ -122,12 +126,14 @@ def adb_run_as(self, package, *args, quiet=False): command.extend(args) if quiet: - return self.adb_quiet(*command) + self.adb_quiet(*command) + return None return self.adb(*command) -def enable_vulkan_debug_layer(device, package, layer): +def enable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Args: device: The device instance. @@ -138,16 +144,24 @@ def enable_vulkan_debug_layer(device, package, layer): print('\nInstalling Vulkan debug layer') layer = os.path.normpath(layer) - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) device.adb('push', layer, ANDROID_TMP_DIR) - device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layerBase, '.') - device.adb('shell', 'settings', 'put', 'global', 'enable_gpu_debug_layers', '1') - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_app', package) - device.adb('shell', 'settings', 'put', 'global', 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + device.adb_run_as(package, 'cp', ANDROID_TMP_DIR + layer_base, '.') + + device.adb('shell', 'settings', 'put', 'global', + 'enable_gpu_debug_layers', '1') + + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_app', package) -def disable_vulkan_debug_layer(device, package, layer): + device.adb('shell', 'settings', 'put', 'global', + 'gpu_debug_layers', EXPECTED_VULKAN_LAYER_NAME) + + +def disable_vulkan_debug_layer( + device: Device, package: str, layer: str) -> None: ''' Clean up the Vulkan layer installation. @@ -157,35 +171,44 @@ def disable_vulkan_debug_layer(device, package, layer): ''' print('\nRemoving Vulkan debug layer') - layerBase = os.path.basename(os.path.normpath(layer)) + layer_base = os.path.basename(os.path.normpath(layer)) + + device.adb('shell', 'settings', 'delete', 'global', + 'enable_gpu_debug_layers') + + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global', 'enable_gpu_debug_layers') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_app') - device.adb('shell', 'settings', 'delete', 'global gpu_debug_layers') - device.adb_run_as(package, 'rm', layerBase, quiet=True) + device.adb('shell', 'settings', 'delete', 'global', + 'gpu_debug_layers') + device.adb_run_as(package, 'rm', layer_base, quiet=True) -def get_layer(): + +def get_layer() -> Optional[str]: ''' Find the debug layer to use in the build directory. + + Returns: + The part to the library to use. ''' base_dir = './build_arm64/source/' - sym_lib = None + # TODO: If we want to use symbolized layer we need to rename it lib = None for path in os.listdir(base_dir): + # Match symbolized library first so we don't use it if path.endswith('_sym.so'): - sym_lib = os.path.join(base_dir, path) + _ = os.path.join(base_dir, path) elif path.endswith('.so'): lib = os.path.join(base_dir, path) - # TODO: If we want to use symbolized layer we need to rename it return lib -def parse_command_line(): +def parse_command_line() -> argparse.Namespace: ''' Parse the command line. @@ -200,7 +223,7 @@ def parse_command_line(): return parser.parse_args() -def main(): +def main() -> int: ''' Script main function. @@ -211,6 +234,9 @@ def main(): device = Device() layer = get_layer() + if not layer: + print('ERROR: Layer binary not found') + return 1 enable_vulkan_debug_layer(device, args.package, layer) @@ -218,6 +244,9 @@ def main(): disable_vulkan_debug_layer(device, args.package, layer) + return 0 + + if __name__ == '__main__': try: sys.exit(main())