Fix source checking tools using project_source_info

Moving files in [0] caused the module not to be found.

Also revert Python 3.6 compatibility changes.

[0] 2c9ab53273
This commit is contained in:
Campbell Barton
2025-01-02 16:23:37 +11:00
parent 030f367c33
commit 7516614c2b

View File

@@ -24,10 +24,8 @@ import subprocess
from typing import (
Any,
IO,
List,
Tuple,
Union,
# Proxies for `collections.abc`
)
from collections.abc import (
Callable,
Iterator,
Sequence,
@@ -59,7 +57,7 @@ def is_c_any(filename: str) -> bool:
CMAKE_DIR = "."
def cmake_cache_var_iter() -> Iterator[Tuple[str, str, str]]:
def cmake_cache_var_iter() -> Iterator[tuple[str, str, str]]:
import re
re_cache = re.compile(r'([A-Za-z0-9_\-]+)?:?([A-Za-z0-9_\-]+)?=(.*)$')
with open(join(CMAKE_DIR, "CMakeCache.txt"), 'r', encoding='utf-8') as cache_file:
@@ -70,7 +68,7 @@ def cmake_cache_var_iter() -> Iterator[Tuple[str, str, str]]:
yield (var, type_ or "", val)
def cmake_cache_var(var: str) -> Union[str, None]:
def cmake_cache_var(var: str) -> str | None:
for var_iter, _type_iter, value_iter in cmake_cache_var_iter():
if var == var_iter:
return value_iter
@@ -85,7 +83,7 @@ def cmake_cache_var_or_exit(var: str) -> str:
return value
def do_ignore(filepath: str, ignore_prefix_list: Union[Sequence[str], None]) -> bool:
def do_ignore(filepath: str, ignore_prefix_list: Sequence[str] | None) -> bool:
if ignore_prefix_list is None:
return False
@@ -93,7 +91,7 @@ def do_ignore(filepath: str, ignore_prefix_list: Union[Sequence[str], None]) ->
return any([relpath.startswith(prefix) for prefix in ignore_prefix_list])
def makefile_log() -> List[str]:
def makefile_log() -> list[str]:
import subprocess
import time
@@ -133,8 +131,8 @@ def makefile_log() -> List[str]:
def build_info(
use_c: bool = True,
use_cxx: bool = True,
ignore_prefix_list: Union[List[str], None] = None,
) -> List[Tuple[str, List[str], List[str]]]:
ignore_prefix_list: list[str] | None = None,
) -> list[tuple[str, list[str], list[str]]]:
makelog = makefile_log()
source = []
@@ -152,7 +150,7 @@ def build_info(
print("parsing make log ...")
for line in makelog:
args_orig: Union[str, List[str]] = line.split()
args_orig: str | list[str] = line.split()
args = [fake_compiler if c in compilers else c for c in args_orig]
if args == args_orig:
# No compilers in the command, skip.
@@ -219,7 +217,7 @@ def build_defines_as_source() -> str:
return stdout.read().strip().decode('ascii')
def build_defines_as_args() -> List[str]:
def build_defines_as_args() -> list[str]:
return [
("-D" + "=".join(l.split(maxsplit=2)[1:]))
for l in build_defines_as_source().split("\n")
@@ -227,7 +225,7 @@ def build_defines_as_args() -> List[str]:
]
def process_make_non_blocking(proc: subprocess.Popen) -> subprocess.Popen:
def process_make_non_blocking(proc: subprocess.Popen[Any]) -> subprocess.Popen[Any]:
import fcntl
for fh in (proc.stderr, proc.stdout):
if fh is None:
@@ -241,11 +239,11 @@ def process_make_non_blocking(proc: subprocess.Popen) -> subprocess.Popen:
# could be moved elsewhere!, this just happens to be used by scripts that also
# use this module.
def queue_processes(
process_funcs: Sequence[Tuple[Callable[..., subprocess.Popen], Tuple[Any, ...]]],
process_funcs: Sequence[tuple[Callable[..., subprocess.Popen[Any]], tuple[Any, ...]]],
*,
job_total: int = -1,
sleep: float = 0.1,
process_finalize: Union[Callable[[subprocess.Popen, bytes, bytes], Union[int, None]], None] = None,
process_finalize: Callable[[subprocess.Popen[Any], bytes, bytes], int | None] | None = None,
) -> None:
""" Takes a list of function arg pairs, each function must return a process
"""
@@ -269,21 +267,18 @@ def queue_processes(
if process_finalize is not None:
def poll_and_finalize(
p: subprocess.Popen,
stdout: List[bytes],
stderr: List[bytes],
) -> Union[int, None]:
p: subprocess.Popen[Any],
stdout: list[bytes],
stderr: list[bytes],
) -> int | None:
assert p.stdout is not None
data = p.stdout.read()
if data:
if data := p.stdout.read():
stdout.append(data)
assert p.stderr is not None
data = p.stderr.read()
if data:
if data := p.stderr.read():
stderr.append(data)
returncode = p.poll()
if returncode is not None:
if (returncode := p.poll()) is not None:
data_stdout, data_stderr = p.communicate()
if data_stdout:
stdout.append(data_stdout)
@@ -293,13 +288,13 @@ def queue_processes(
return returncode
else:
def poll_and_finalize(
p: subprocess.Popen,
stdout: List[bytes],
stderr: List[bytes],
) -> Union[int, None]:
p: subprocess.Popen[Any],
stdout: list[bytes],
stderr: list[bytes],
) -> int | None:
return p.poll()
processes: List[Tuple[subprocess.Popen, List[bytes], List[bytes]]] = []
processes: list[tuple[subprocess.Popen[Any], list[bytes], list[bytes]]] = []
for func, args in process_funcs:
# wait until a thread is free
while 1: