Initial rework
This commit is contained in:
@@ -0,0 +1,62 @@
|
||||
import yaml
|
||||
|
||||
from argparse import ArgumentParser
|
||||
from argparse import Namespace
|
||||
from pathlib import Path
|
||||
from semantic_version import Version
|
||||
|
||||
|
||||
# Load application configurations.
|
||||
cfgdir = Path.home() / ".config" / "chinook" / "remotes.yaml"
|
||||
remotes = yaml.full_load(cfgdir) if Path.exists(cfgdir) else []
|
||||
|
||||
def _build(args: Namespace) -> None:
|
||||
from chinook import management
|
||||
from chinook import pipeline
|
||||
|
||||
from chinook._config import Config
|
||||
from chinook._logger import logger
|
||||
|
||||
try:
|
||||
projpath = Path(args.project)
|
||||
cfgpath = projpath / "chinookconfig"
|
||||
return pipeline.build(
|
||||
Config.create_default_or_load(cfgpath),
|
||||
management.load(projpath)
|
||||
)
|
||||
except Exception as err:
|
||||
logger.exception(str(err))
|
||||
logger.critical("Failed to build project")
|
||||
|
||||
|
||||
def main() -> None:
|
||||
print("\n".join([
|
||||
"┏┓┓ • ┓ ",
|
||||
"┃ ┣┓┓┏┓┏┓┏┓┃┏",
|
||||
"┗┛┛┗┗┛┗┗┛┗┛┛┗ v" + str(Version(major=0,minor=1,patch=0))
|
||||
]))
|
||||
|
||||
# Parent parser to all subcommands.
|
||||
parser = ArgumentParser(
|
||||
prog = "chinook",
|
||||
description = "Opinionated build tool for C/C++!"
|
||||
)
|
||||
|
||||
commands = parser.add_subparsers(required=True)
|
||||
buildcmd = commands.add_parser("build",help="Builds your Chinook project")
|
||||
buildcmd.set_defaults(func=_build)
|
||||
buildcmd.add_argument(
|
||||
"project",
|
||||
nargs="?",
|
||||
default=".",
|
||||
type=str,
|
||||
help="The path to the project to build",
|
||||
metavar="PROJECT"
|
||||
)
|
||||
|
||||
arguments = parser.parse_args()
|
||||
arguments.func(arguments)
|
||||
|
||||
|
||||
# Auto execute.
|
||||
if __name__ == "__main__": main()
|
||||
@@ -0,0 +1,119 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from functools import cache
|
||||
from pathlib import Path
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from .models import Abi
|
||||
from .models import Arch
|
||||
from .models import Sys
|
||||
from .models import Tooling
|
||||
from .models import Triple
|
||||
|
||||
|
||||
@dataclass
|
||||
class Config(yaml.YAMLObject):
|
||||
_DEFAULT_TOOLS: ClassVar[list[Tooling]] = [
|
||||
Tooling(
|
||||
name = "gnuc-generic",
|
||||
when = [
|
||||
Triple(arch=Arch.X64, sys=Sys.LINUX, abi=Abi.GNU),
|
||||
Triple(arch=Arch.X64, sys=Sys.WINDOWS, abi=Abi.MINGW)
|
||||
],
|
||||
cc = "gcc",
|
||||
cxx = "g++",
|
||||
ld = "ld",
|
||||
ar = "ar"
|
||||
)
|
||||
]
|
||||
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!config"
|
||||
|
||||
toolings: list[Tooling] = field(default_factory=list)
|
||||
|
||||
|
||||
@property
|
||||
def native_tooling(self) -> Tooling:
|
||||
candidates = self.toolings + self._DEFAULT_TOOLS
|
||||
return next((t for t in candidates if t.native), None)
|
||||
|
||||
@classmethod
|
||||
@property
|
||||
@cache
|
||||
def config_directory(cls) -> Path:
|
||||
return Path.home() / ".config" / "chinook"
|
||||
|
||||
@classmethod
|
||||
def create_default_or_load(cls, cfgpath: Path) -> Self:
|
||||
if Path.exists(cfgpath):
|
||||
return cls._load_config(cfgpath)
|
||||
|
||||
cfgpath = cls.config_directory / "config"
|
||||
if Path.exists(cfgpath):
|
||||
return cls._load_config(cfgpath)
|
||||
|
||||
default = cls(toolings=cls._DEFAULT_TOOLS)
|
||||
cfgpath.parent.mkdir(parents=True, exist_ok=True)
|
||||
with open(cfgpath, 'w') as file:
|
||||
file.write(yaml.dump(default, Dumper=cls.yaml_dumper, sort_keys=False))
|
||||
return default
|
||||
|
||||
@classmethod
|
||||
def _load_config(cls, cfgpath: Path) -> Self:
|
||||
try:
|
||||
with open(cfgpath, 'r') as file:
|
||||
data = yaml.load(file.read(), Loader=cls.yaml_loader)
|
||||
return data if isinstance(data, Config) else cls.from_dict(data)
|
||||
except Exception as rethrowme:
|
||||
raise rethrowme
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"toolings": [Tooling.to_dict(t) for t in data.toolings]
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
if not isinstance(data, Config):
|
||||
raise ValueError
|
||||
return dumper.represent_mapping(
|
||||
"tag:yaml.org,2002:map",
|
||||
cls.to_dict(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
return cls(**{
|
||||
"toolings": Tooling.from_list(data.get("toolings", []))
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_dict(
|
||||
loader.construct_mapping(node, deep=True)
|
||||
)
|
||||
|
||||
|
||||
Config.yaml_dumper.add_representer(
|
||||
Config.yaml_tag,
|
||||
Config.to_yaml
|
||||
)
|
||||
|
||||
Config.yaml_loader.add_constructor(
|
||||
Config.yaml_tag,
|
||||
Config.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,10 @@
|
||||
import logging
|
||||
|
||||
_formatter = logging.Formatter("%(asctime)s %(levelname)s %(message)s")
|
||||
_console = logging.StreamHandler()
|
||||
_console.setLevel(logging.DEBUG)
|
||||
_console.setFormatter(_formatter)
|
||||
|
||||
logger = logging.getLogger("chinook")
|
||||
logger.addHandler(_console)
|
||||
logger.setLevel(logging.INFO)
|
||||
@@ -0,0 +1,2 @@
|
||||
from ._compiler import Compiler
|
||||
from ._options import CompilerOptions
|
||||
@@ -0,0 +1,36 @@
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Result
|
||||
from ._driver import CompilerDriver
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
class ClangDriver(CompilerDriver):
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
@@ -0,0 +1,67 @@
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Abi
|
||||
from ..models import Result
|
||||
from ..models import Tooling
|
||||
from ._clang import ClangDriver
|
||||
from ._driver import CompilerDriver
|
||||
from ._gnuc import GnucDriver
|
||||
from ._mingw import MingwDriver
|
||||
from ._msvc import MsvcDriver
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
@dataclass
|
||||
class Compiler:
|
||||
__COMPILER_MAPPING = {
|
||||
Abi.GNU: GnucDriver,
|
||||
Abi.MSVC: MsvcDriver,
|
||||
Abi.MINGW: MingwDriver,
|
||||
Abi.ANDROID: ClangDriver,
|
||||
Abi.ELF: GnucDriver,
|
||||
}
|
||||
|
||||
tooling: Tooling
|
||||
_driver: CompilerDriver = field(init=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self._driver = Compiler._driver_for(self.tooling)
|
||||
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
return self._driver.compile_program(options, dstpath)
|
||||
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
return self._driver.compile_archive(options, dstpath)
|
||||
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
return self._driver.compile_library(options, dstpath)
|
||||
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstPath: Path
|
||||
) -> Result:
|
||||
return self._driver.compile_object(options, srcpath, dstPath)
|
||||
|
||||
@classmethod
|
||||
def _driver_for(cls, tooling: Tooling) -> CompilerDriver:
|
||||
from ..models import Triple
|
||||
triples = tooling.when or [Triple.current()]
|
||||
driver = cls.__COMPILER_MAPPING.get(triples[0].abi, GnucDriver)
|
||||
return driver(tooling)
|
||||
@@ -0,0 +1,48 @@
|
||||
from abc import ABC
|
||||
from abc import abstractmethod
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Tooling
|
||||
from ..models import Result
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
class CompilerDriver(ABC):
|
||||
def __init__(self, tooling: Tooling) -> None:
|
||||
if tooling is None:
|
||||
raise ValueError
|
||||
self.tooling = tooling
|
||||
|
||||
@abstractmethod
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
...
|
||||
|
||||
@abstractmethod
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
...
|
||||
@@ -0,0 +1,193 @@
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Optimization
|
||||
from ..models import Result
|
||||
from ..models import StdC
|
||||
from ..models import StdCpp
|
||||
from ..shell import execute
|
||||
from .._logger import logger
|
||||
from ._driver import CompilerDriver
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
class GnucDriver(CompilerDriver):
|
||||
__OPTLEVEL_MAPPING = {
|
||||
Optimization.NONE: "-O0",
|
||||
Optimization.LOW: "-O1",
|
||||
Optimization.HIGH: "-O2",
|
||||
Optimization.DEBUG: "-Og",
|
||||
Optimization.SMALL: "-Os",
|
||||
Optimization.SPEED: "-O3",
|
||||
Optimization.TINY: "-Oz",
|
||||
Optimization.FAST: "-OFast"
|
||||
}
|
||||
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
assert options is not None, "Compiler options are required"
|
||||
assert len(options.objfiles) != 0, "Object files are required"
|
||||
|
||||
compiler = self.tooling.cc
|
||||
cppext = [".cpp", ".cxx", ".cc"]
|
||||
for obj in options.objfiles:
|
||||
if obj.suffixes[0] in cppext:
|
||||
compiler = self.tooling.cxx
|
||||
break
|
||||
|
||||
args = [f"{compiler} -o {str(dstpath)}"]
|
||||
args.extend(sorted(
|
||||
[str(v) for v in options.objfiles],
|
||||
key=lambda f: f.endswith(".so")
|
||||
))
|
||||
|
||||
args.extend(options.flags)
|
||||
if len(options.archives) > 0:
|
||||
for directory in options.libdirs: args.append(f"-L{str(directory)}")
|
||||
for archive in options.archives: args.append(f"-l{archive}")
|
||||
cmdstr = ' '.join(args)
|
||||
# logger.info(cmdstr)
|
||||
|
||||
result: list[Result] = [None]
|
||||
execute(
|
||||
cmdstr,
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_data(result, str(dstpath), cmdstr, stdout, stderr),
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_error(result, str(dstpath), cmdstr, stdout, stderr)
|
||||
)
|
||||
return result[0]
|
||||
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
args = [self.tooling.ar, f"rcs {str(dstpath)}"]
|
||||
for obj in options.objfiles:
|
||||
args.append(str(obj))
|
||||
cmdstr = ' '.join(args)
|
||||
|
||||
result: list[Result] = [None]
|
||||
execute(
|
||||
cmdstr,
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_data(result, str(dstpath), cmdstr, stdout, stderr),
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_error(result, str(dstpath), cmdstr, stdout, stderr)
|
||||
)
|
||||
return result[0]
|
||||
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
assert options is not None, "Compiler options are required"
|
||||
assert len(options.objfiles) != 0, "Object files are required"
|
||||
|
||||
compiler = self.tooling.cc
|
||||
cppext = [".cpp", ".cxx", ".cc"]
|
||||
for obj in options.objfiles:
|
||||
if obj.suffixes[0] in cppext:
|
||||
compiler = self.tooling.cxx
|
||||
break
|
||||
|
||||
args = [f"{compiler} -shared -o {str(dstpath)}"]
|
||||
args.extend(sorted(
|
||||
[str(v) for v in options.objfiles],
|
||||
key=lambda f: f.endswith(".so")
|
||||
))
|
||||
|
||||
args.extend(options.flags)
|
||||
if len(options.archives) > 0:
|
||||
for directory in options.libdirs: args.append(f"-L{str(directory)}")
|
||||
for archive in options.archives: args.append(f"-l{archive}")
|
||||
cmdstr = ' '.join(args)
|
||||
# logger.info(cmdstr)
|
||||
|
||||
result: list[Result] = [None]
|
||||
execute(
|
||||
cmdstr,
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_data(result, str(dstpath), cmdstr, stdout, stderr),
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_error(result, str(dstpath), cmdstr, stdout, stderr)
|
||||
)
|
||||
return result[0]
|
||||
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
assert options is not None, "Compiler options are required"
|
||||
|
||||
compiler = self.tooling.cc
|
||||
standard = options.stdc
|
||||
optlevel = self.__OPTLEVEL_MAPPING.get(options.optlevel)
|
||||
|
||||
cppext = [".cpp", ".cxx", ".cc"]
|
||||
if srcpath.suffix in cppext:
|
||||
compiler = self.tooling.cxx
|
||||
standard = options.stdcpp
|
||||
|
||||
args = [
|
||||
f"{compiler}",
|
||||
f"-c {str(srcpath)}",
|
||||
f"-o {str(dstpath)}"
|
||||
]
|
||||
if options.pic:
|
||||
args.append("-fPIC")
|
||||
args.append(f"-std={standard}")
|
||||
args.append(optlevel)
|
||||
args.extend(options.flags)
|
||||
for directory in options.incdirs: args.append(f"-I{str(directory)}")
|
||||
for directory in options.libdirs: args.append(f"-L{str(directory)}")
|
||||
for archive in options.archives: args.append(f"-l{archive}")
|
||||
for define in options.defines: args.append(f"-D{define}")
|
||||
cmdstr = ' '.join(args)
|
||||
# logger.info(cmdstr)
|
||||
|
||||
result: list[Result] = [None]
|
||||
execute(
|
||||
cmdstr,
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_data(result, str(dstpath), cmdstr, stdout, stderr),
|
||||
lambda cmdstr, stdout, stderr:
|
||||
self.handle_on_error(result, str(dstpath), cmdstr, stdout, stderr)
|
||||
)
|
||||
return result[0]
|
||||
|
||||
def handle_on_data(
|
||||
self,
|
||||
result: list[Result],
|
||||
target: str,
|
||||
cmdstr: str,
|
||||
stdout: str,
|
||||
stderr: str
|
||||
) -> None:
|
||||
result[0] = Result(
|
||||
success = True,
|
||||
target = target,
|
||||
command = cmdstr
|
||||
)
|
||||
result[0].diagnostics.append(stdout)
|
||||
|
||||
def handle_on_error(
|
||||
self,
|
||||
result: list[Result],
|
||||
target: str,
|
||||
cmdstr: str,
|
||||
stdout: str,
|
||||
stderr: str
|
||||
) -> None:
|
||||
result[0] = Result(
|
||||
success = False,
|
||||
target = target,
|
||||
command = cmdstr
|
||||
)
|
||||
result[0].diagnostics.append(stderr)
|
||||
@@ -0,0 +1,36 @@
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Result
|
||||
from ._driver import CompilerDriver
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
class MingwDriver(CompilerDriver):
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
@@ -0,0 +1,36 @@
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Result
|
||||
from ._driver import CompilerDriver
|
||||
from ._options import CompilerOptions
|
||||
|
||||
|
||||
class MsvcDriver(CompilerDriver):
|
||||
def compile_program(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_archive(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_library(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
|
||||
def compile_object(
|
||||
self,
|
||||
options: CompilerOptions,
|
||||
srcpath: Path,
|
||||
dstpath: Path
|
||||
) -> Result:
|
||||
pass
|
||||
@@ -0,0 +1,39 @@
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from pathlib import Path
|
||||
from typing import Self
|
||||
|
||||
from ..models import Optimization
|
||||
from ..models import StdC
|
||||
from ..models import StdCpp
|
||||
|
||||
|
||||
@dataclass
|
||||
class CompilerOptions:
|
||||
pic: bool = False
|
||||
debug: bool = False
|
||||
stdc: StdC = StdC.C11
|
||||
stdcpp: StdCpp = StdCpp.C14
|
||||
optlevel: Optimization = Optimization.NONE
|
||||
|
||||
incdirs: set[Path] = field(default_factory=set)
|
||||
libdirs: set[Path] = field(default_factory=set)
|
||||
objfiles: set[Path] = field(default_factory=set)
|
||||
archives: set[str] = field(default_factory=set)
|
||||
defines: set[str] = field(default_factory=set)
|
||||
flags: set[str] = field(default_factory=set)
|
||||
|
||||
def __deepcopy__(self, memo) -> Self:
|
||||
return CompilerOptions(
|
||||
self.pic,
|
||||
self.debug,
|
||||
self.stdc,
|
||||
self.stdcpp,
|
||||
self.optlevel,
|
||||
set(self.incdirs),
|
||||
set(self.libdirs),
|
||||
set(self.objfiles),
|
||||
set(self.archives),
|
||||
set(self.defines),
|
||||
set(self.flags)
|
||||
)
|
||||
@@ -0,0 +1,21 @@
|
||||
from typing import Optional
|
||||
|
||||
from ..models import Project
|
||||
from ..models import Target
|
||||
from ._clone import clone
|
||||
from ._except import NoProjectFoundError
|
||||
from ._latest import latest
|
||||
from ._load import load
|
||||
|
||||
|
||||
def find_project_by_name(name: str) -> Optional[Project]:
|
||||
from ._shared import projects_by_name_mapping
|
||||
return projects_by_name_mapping.get(name)
|
||||
|
||||
def find_target_by_name(name: str) -> Optional[Target]:
|
||||
from ._shared import targets_by_name_mapping
|
||||
return targets_by_name_mapping.get(name)
|
||||
|
||||
def find_project_by_target(name: str) -> Optional[Project]:
|
||||
from ._shared import project_by_targets_mapping
|
||||
return project_by_targets_mapping.get(name)
|
||||
@@ -0,0 +1,11 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class NoProjectFoundError(Exception):
|
||||
@property
|
||||
def path(self) -> Path:
|
||||
return self._path
|
||||
|
||||
def __init__(self, path: Path) -> None:
|
||||
super().__init__(f"Project could not be found: {path}")
|
||||
self._path = path
|
||||
@@ -0,0 +1,10 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
def latest(src: Path, dst: Path) -> bool:
|
||||
if not Path.exists(dst):
|
||||
return False
|
||||
|
||||
srcmtime = src.stat().st_mtime
|
||||
dstmtime = dst.stat().st_mtime
|
||||
return srcmtime < dstmtime
|
||||
@@ -0,0 +1,126 @@
|
||||
import yaml
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from ..models import Access
|
||||
from ..models import Accessor
|
||||
from ..models import Dependency
|
||||
from ..models import Destination
|
||||
from ..models import Origination
|
||||
from ..models import Output
|
||||
from ..models import Project
|
||||
from ..models import Target
|
||||
from ..shell import execute
|
||||
from .._config import Config
|
||||
from ._except import NoProjectFoundError
|
||||
from ._shared import projects_by_name_mapping
|
||||
from ._shared import targets_by_name_mapping
|
||||
from ._shared import project_by_targets_mapping
|
||||
|
||||
|
||||
def load(path: Path) -> Project:
|
||||
if path in projects_by_name_mapping:
|
||||
return projects_by_name_mapping[path]
|
||||
|
||||
fullpath = path / "chinookfile"
|
||||
if not Path.exists(fullpath):
|
||||
raise NoProjectFoundError(fullpath)
|
||||
|
||||
try:
|
||||
with open(fullpath, 'r') as file:
|
||||
yamlobj = yaml.load(file.read(), Loader=Project.yaml_loader)
|
||||
project = Project.from_dict(yamlobj)
|
||||
project.origination = Origination(path)
|
||||
project.destination = Destination(path)
|
||||
_collect_required(project)
|
||||
_collect_targets(project)
|
||||
_flatten_targets(project)
|
||||
return project
|
||||
except Exception as rethrowme:
|
||||
raise rethrowme
|
||||
|
||||
def _collect_required(project: Project) -> None:
|
||||
for required in project.requires:
|
||||
_clone_then_load_required(required)
|
||||
|
||||
def _collect_targets(project: Project) -> None:
|
||||
for target in project.targets:
|
||||
if target.name not in targets_by_name_mapping:
|
||||
targets_by_name_mapping[target.name] = target
|
||||
project_by_targets_mapping[target.name] = project
|
||||
|
||||
def _flatten_targets(project: Project) -> None:
|
||||
for target in project.targets:
|
||||
_flatten_target(target)
|
||||
target.incs.includes.add(Accessor(
|
||||
level = Access.PUBLIC \
|
||||
if target.name in project.exports \
|
||||
else Access.PRIVATE,
|
||||
value = project.origination.src("include")
|
||||
))
|
||||
target.incs.libraries.add(Accessor(
|
||||
level = Access.PUBLIC,
|
||||
value = project.destination.libpath
|
||||
))
|
||||
|
||||
def _flatten_target(target: Target) -> None:
|
||||
for dependency in target.deps:
|
||||
if (parent:=targets_by_name_mapping.get(dependency.value)) == None:
|
||||
raise ValueError(f"No such dependable target: {dependency}")
|
||||
_inherit_target(target, parent)
|
||||
|
||||
def _inherit_target(target: Target, parent: Target) -> None:
|
||||
_flatten_target(parent)
|
||||
target.incs.includes.update([v for v in parent.incs.includes if v.level != Access.PRIVATE])
|
||||
target.incs.libraries.update([v for v in parent.incs.libraries if v.level != Access.PRIVATE])
|
||||
|
||||
parent_project = project_by_targets_mapping.get(parent.name)
|
||||
assert parent_project is not None, "Parent project not mapped!"
|
||||
target.incs.includes.add(Accessor(
|
||||
level = Access.PUBLIC,
|
||||
value = parent_project.origination.src("include")
|
||||
))
|
||||
target.incs.libraries.add(Accessor(
|
||||
level = Access.PUBLIC,
|
||||
value = parent_project.destination.libpath
|
||||
))
|
||||
|
||||
target.opts.update([v for v in parent.opts if v.level != Access.PRIVATE])
|
||||
target.defs.update([v for v in parent.defs if v.level != Access.PRIVATE])
|
||||
target.deps.update([v for v in parent.deps if v.level != Access.PRIVATE])
|
||||
target.libs.update([v for v in parent.libs if v.level != Access.PRIVATE])
|
||||
target.libs.add(Accessor(
|
||||
level = Access.PUBLIC,
|
||||
value = parent.name
|
||||
))
|
||||
|
||||
if parent.type == Output.INT:
|
||||
target.srcs.update([v for v in parent.srcs if v.level != Access.PRIVATE])
|
||||
|
||||
def _clone_then_load_required(required: Dependency) -> None:
|
||||
reqpath = Config.config_directory / required.cid
|
||||
if not Path.exists(reqpath):
|
||||
_clone_required(required, reqpath)
|
||||
load(reqpath)
|
||||
|
||||
def _clone_required(
|
||||
required: Dependency,
|
||||
reqpath: Path
|
||||
) -> None:
|
||||
# TODO: Clean this up.
|
||||
def handle_success(cmdstr: str, stdout: str, stderr: str) -> None:
|
||||
return
|
||||
|
||||
errstr = None
|
||||
def handle_error(cmdstr: str, stdout: str, stderr: str) -> None:
|
||||
global errstr
|
||||
errstr = stderr
|
||||
|
||||
execute(
|
||||
f"git clone -b {required.branch} --depth 1 {required.remote} {reqpath}",
|
||||
handle_success,
|
||||
handle_error
|
||||
)
|
||||
|
||||
if errstr != None:
|
||||
raise Exception(errstr)
|
||||
@@ -0,0 +1,8 @@
|
||||
from ..models import Project
|
||||
from ..models import Target
|
||||
|
||||
|
||||
projects_by_name_mapping: dict[str, Project] = {}
|
||||
targets_by_name_mapping: dict[str, Target] = {}
|
||||
|
||||
project_by_targets_mapping: dict[str, Project] = {}
|
||||
@@ -0,0 +1,19 @@
|
||||
from ._access import Access
|
||||
from ._accessor import Accessor
|
||||
from ._branch import Branch
|
||||
from ._dependency import Dependency
|
||||
from ._destination import Destination
|
||||
from ._identity import Identity
|
||||
from ._optimization import Optimization
|
||||
from ._origination import Origination
|
||||
from ._output import Output
|
||||
from ._project import Project
|
||||
from ._standards import StdC
|
||||
from ._standards import StdCpp
|
||||
from ._result import Result
|
||||
from ._target import Target
|
||||
from ._tooling import Tooling
|
||||
from ._triple import Arch
|
||||
from ._triple import Sys
|
||||
from ._triple import Abi
|
||||
from ._triple import Triple
|
||||
@@ -0,0 +1,8 @@
|
||||
from enum import auto
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class Access(StrEnum):
|
||||
PUBLIC = "!public"
|
||||
PROTECTED = "!protected"
|
||||
PRIVATE = "!private"
|
||||
@@ -0,0 +1,79 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from ._access import Access
|
||||
|
||||
|
||||
@dataclass
|
||||
class Accessor(yaml.YAMLObject):
|
||||
yaml_dumper: ClassVar[Any] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[Any] = yaml.SafeLoader
|
||||
|
||||
level: Access
|
||||
value: str
|
||||
|
||||
def __hash__(self) -> int:
|
||||
return hash((self.level, self.value))
|
||||
|
||||
@classmethod
|
||||
def to_str(cls, data: Self) -> str:
|
||||
return f"{data.level} {data.value}"
|
||||
|
||||
@classmethod
|
||||
def to_list(cls, data: list[Self]) -> list[str]:
|
||||
return [cls.to_str(d) for d in data]
|
||||
|
||||
@classmethod
|
||||
def to_yaml(cls, dumper: yaml.SafeDumper, data: Self) -> yaml.Node:
|
||||
if not isinstance(data, Accessor):
|
||||
raise ValueError
|
||||
return dumper.represent_scalar(data.level, data.value)
|
||||
|
||||
@classmethod
|
||||
def from_value(cls, value: Any) -> Self:
|
||||
if isinstance(value, Accessor):
|
||||
return value
|
||||
if isinstance(value, str):
|
||||
return cls.from_dict({
|
||||
"level": Access.PUBLIC,
|
||||
"value": value
|
||||
})
|
||||
raise ValueError
|
||||
|
||||
@staticmethod
|
||||
def from_list(value: list[Any]) -> list[Self]:
|
||||
return [Accessor.from_value(v) for v in value]
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
return cls(**{
|
||||
"level": data.get("level", Access.PUBLIC),
|
||||
"value": data.get("value", "")
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, loader: yaml.SafeLoader, node: yaml.Node) -> Self:
|
||||
return cls.from_dict({
|
||||
"level": node.tag,
|
||||
"value": loader.construct_scalar(node)
|
||||
})
|
||||
|
||||
|
||||
Accessor.yaml_loader.add_constructor(
|
||||
Access.PUBLIC,
|
||||
Accessor.from_yaml
|
||||
)
|
||||
|
||||
Accessor.yaml_loader.add_constructor(
|
||||
Access.PROTECTED,
|
||||
Accessor.from_yaml
|
||||
)
|
||||
|
||||
Accessor.yaml_loader.add_constructor(
|
||||
Access.PRIVATE,
|
||||
Accessor.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,40 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
|
||||
from semantic_version import validate
|
||||
from semantic_version import Version
|
||||
|
||||
from typing import ClassVar
|
||||
from typing import Optional
|
||||
|
||||
|
||||
@dataclass
|
||||
class Branch(yaml.YAMLObject):
|
||||
yaml_tag: ClassVar[str] = u"!branch"
|
||||
yaml_loader: ClassVar[any] = yaml.SafeLoader
|
||||
|
||||
value: str
|
||||
_valid: Optional[bool] = field(init=False, default=None)
|
||||
|
||||
@property
|
||||
def as_version(self) -> Version:
|
||||
if not self.is_version:
|
||||
return Version(0, 0, 0, self.value)
|
||||
return Version.parse(self.value)
|
||||
|
||||
@property
|
||||
def is_version(self) -> bool:
|
||||
if self._valid == None:
|
||||
self._valid = validate(self.value)
|
||||
return self._valid
|
||||
|
||||
def __str__(self) -> str:
|
||||
return self.value
|
||||
|
||||
|
||||
Branch.yaml_loader.add_constructor(
|
||||
Branch.yaml_tag,
|
||||
Branch.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,70 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from ._branch import Branch
|
||||
|
||||
|
||||
@dataclass
|
||||
class Dependency(yaml.YAMLError):
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!dependency"
|
||||
|
||||
remote: str
|
||||
branch: Branch
|
||||
|
||||
@property
|
||||
def cid(self) -> str:
|
||||
from urllib.parse import urlparse
|
||||
if self.remote.startswith("git@"):
|
||||
url = self.remote.replace(":", "/", 1)
|
||||
url = url.replace("git@", "https://")
|
||||
parsed = urlparse(url)
|
||||
path = parsed.path.strip("/")
|
||||
if path.endswith(".git"):
|
||||
path = path[:-4]
|
||||
return path + str(self.branch)
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"remote": data.remote,
|
||||
"branch": data.branch
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_list(cls, data: list[Self]) -> list[dict]:
|
||||
return [cls.to_dict(v) for v in data]
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
if not isinstance(data, Dependency):
|
||||
raise ValueError
|
||||
return dumper.represent_scalar(
|
||||
"tag:yaml.org,2002:str",
|
||||
cls.to_cid(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
return cls(**data)
|
||||
|
||||
@classmethod
|
||||
def from_list(cls, data: list[str]) -> list[Self]:
|
||||
return [cls.from_dict(v) for v in data]
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_cid(loader.construct_scalar(node))
|
||||
@@ -0,0 +1,53 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
# TODO: Support custom output file extension based on user
|
||||
# or platform specification.
|
||||
class Destination:
|
||||
@property
|
||||
def root(self) -> Path:
|
||||
return self._root
|
||||
|
||||
@property
|
||||
def binpath(self) -> Path:
|
||||
return self._binpath
|
||||
|
||||
@property
|
||||
def dllpath(self) -> Path:
|
||||
return self._binpath
|
||||
|
||||
@property
|
||||
def libpath(self) -> Path:
|
||||
return self._libpath
|
||||
|
||||
@property
|
||||
def objpath(self) -> Path:
|
||||
return self._objpath
|
||||
|
||||
def __init__(self, outpath: Path, mkdirs=True) -> None:
|
||||
self._root = outpath / ".chinook"
|
||||
self._binpath = self._root / "bin"
|
||||
self._libpath = self._root / "lib"
|
||||
self._objpath = self._root / "obj"
|
||||
|
||||
if mkdirs:
|
||||
self._root.mkdir(parents=True, exist_ok=True)
|
||||
self._binpath.mkdir(parents=True, exist_ok=True)
|
||||
self._libpath.mkdir(parents=True, exist_ok=True)
|
||||
self._objpath.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
@staticmethod
|
||||
def cwd() -> 'Destination':
|
||||
return Destination(Path.cwd(), mkdirs=False)
|
||||
|
||||
def bin(self, file: str) -> Path:
|
||||
return self.binpath / file
|
||||
|
||||
def dll(self, file: str) -> Path:
|
||||
return self.dllpath / f"{file}.so"
|
||||
|
||||
def lib(self, file: str) -> Path:
|
||||
return self.libpath / f"lib{file}.a"
|
||||
|
||||
def obj(self, file: Path) -> Path:
|
||||
return self.objpath / f"{file}.o"
|
||||
@@ -0,0 +1,43 @@
|
||||
import re
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Self
|
||||
|
||||
from ._branch import Branch
|
||||
|
||||
|
||||
@dataclass
|
||||
class Identity(yaml.YAMLObject):
|
||||
name: str
|
||||
gpid: str
|
||||
semv: Branch
|
||||
|
||||
@property
|
||||
def cid(self) -> str:
|
||||
return f"{self.gpid}/{self.name}@{self.semv}"
|
||||
|
||||
@classmethod
|
||||
def to_cid(cls, data: Self) -> str:
|
||||
return f"{data.gpid}/{data.name}@{data.semv}"
|
||||
|
||||
@staticmethod
|
||||
def from_cid(cid: str) -> dict[str, str]:
|
||||
pattern = r"^(?P<gpid>[^/]+)/(?P<name>[^@]+)@(?P<semv>.+)$"
|
||||
match = re.match(pattern, cid)
|
||||
return match.groupdict() if match else {}
|
||||
|
||||
@staticmethod
|
||||
def from_fields(fields: dict) -> Self:
|
||||
if (name:=fields.get("name")) == None:
|
||||
raise AttributeError(name="name",obj=fields)
|
||||
if (gpid:=fields.get("gpid")) == None:
|
||||
raise AttributeError(name="gpid",obj=fields)
|
||||
if (semv:=fields.get("semv")) == None:
|
||||
raise AttributeError(name="semv",obj=fields)
|
||||
|
||||
return Identity(
|
||||
name = name,
|
||||
gpid = gpid,
|
||||
semv = Branch(semv)
|
||||
)
|
||||
@@ -0,0 +1,47 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
|
||||
from typing import Any
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from ._accessor import Accessor
|
||||
|
||||
|
||||
@dataclass
|
||||
class Includes(yaml.YAMLObject):
|
||||
yaml_dumper: ClassVar[Any] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[Any] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!includes"
|
||||
|
||||
includes: set[Accessor] = field(default_factory=set)
|
||||
libraries: set[Accessor] = field(default_factory=set)
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"includes": Accessor.to_list(data.includes),
|
||||
"libraries": Accessor.to_list(data.libraries)
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_yaml(cls, dumper: yaml.SafeDumper, data: Self) -> yaml.Node:
|
||||
if not isinstance(data, Includes):
|
||||
raise ValueError
|
||||
return dumper.represent_mapping(
|
||||
"tag:yaml.org,2002:map",
|
||||
cls.to_dict(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
return cls(**{
|
||||
"includes": set(Accessor.from_list(data.get("includes", []))),
|
||||
"libraries": set(Accessor.from_list(data.get("libraries", [])))
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_yaml(cls, loader: yaml.SafeLoader, data: yaml.Node) -> Self:
|
||||
return cls.from_dict(loader.construct_mapping(data))
|
||||
@@ -0,0 +1,13 @@
|
||||
from enum import auto
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class Optimization(StrEnum):
|
||||
NONE = "none"
|
||||
LOW = "low"
|
||||
HIGH = "high"
|
||||
DEBUG = "debug"
|
||||
SMALL = "small"
|
||||
SPEED = "speed"
|
||||
TINY = "tiny"
|
||||
FAST = "fast"
|
||||
@@ -0,0 +1,19 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
class Origination:
|
||||
@property
|
||||
def root(self) -> Path:
|
||||
return self._root
|
||||
|
||||
def __init__(self, inpath: Path) -> None:
|
||||
self._root = inpath
|
||||
self._srcpath = self._root / "source"
|
||||
self._incpath = self._root / "include"
|
||||
|
||||
@staticmethod
|
||||
def cwd() -> 'Origination':
|
||||
return Origination(Path.cwd())
|
||||
|
||||
def src(self, file: Path) -> Path:
|
||||
return self._root / file
|
||||
@@ -0,0 +1,10 @@
|
||||
from enum import StrEnum
|
||||
|
||||
|
||||
class Output(StrEnum):
|
||||
IMP = "imported"
|
||||
INT = "interface"
|
||||
EXE = "program"
|
||||
LIB = "archive"
|
||||
DLL = "dynlib"
|
||||
OBJ = "objfile"
|
||||
@@ -0,0 +1,69 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from ._dependency import Dependency
|
||||
from ._destination import Destination
|
||||
from ._identity import Identity
|
||||
from ._origination import Origination
|
||||
from ._target import Target
|
||||
|
||||
|
||||
@dataclass
|
||||
class Project(Identity):
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!project"
|
||||
|
||||
requires: list[Dependency] = field(default_factory=list)
|
||||
targets: list[Target] = field(default_factory=list)
|
||||
exports: list[str] = field(default_factory=list)
|
||||
|
||||
origination: Origination = field(default=Origination.cwd())
|
||||
destination: Destination = field(default=Destination.cwd())
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"requires": Dependency.to_list(data.requires),
|
||||
"targets": Target.to_list(data.targets),
|
||||
"exports": data.exports
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
return dumper.represent_mapping(
|
||||
"tag:yaml.org,2002:map",
|
||||
cls.to_dict(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
def bad_attr(name: str, obj: dict) -> None:
|
||||
raise AttributeError(name=name,obj=obj)
|
||||
|
||||
return cls(**{
|
||||
"name": data.get("name") or bad_attr("name", data),
|
||||
"gpid": data.get("gpid") or bad_attr("gpid", data),
|
||||
"semv": data.get("semv") or bad_attr("semv", data),
|
||||
"requires": Dependency.from_list(data.get("requires", [])),
|
||||
"targets": Target.from_list(data.get("targets", [])),
|
||||
"exports": data.get("exports", [])
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_dict(
|
||||
loader.construct_mapping(node, deep=True)
|
||||
)
|
||||
@@ -0,0 +1,10 @@
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
|
||||
|
||||
@dataclass
|
||||
class Result:
|
||||
success: bool
|
||||
target: str
|
||||
command: str
|
||||
diagnostics: list[str] = field(default_factory=list)
|
||||
@@ -0,0 +1,21 @@
|
||||
from enum import StrEnum
|
||||
from typing import Union
|
||||
|
||||
|
||||
class StdC(StrEnum):
|
||||
C89 = "c89"
|
||||
C99 = "c99"
|
||||
C11 = "c11"
|
||||
C17 = "c17"
|
||||
C23 = "c23"
|
||||
|
||||
class StdCpp(StrEnum):
|
||||
C98 = "c++98"
|
||||
C11 = "c++11"
|
||||
C14 = "c++14"
|
||||
C17 = "c++17"
|
||||
C20 = "c++20"
|
||||
C23 = "c++23"
|
||||
C26 = "c++26"
|
||||
|
||||
Standards = Union[StdC, StdCpp]
|
||||
@@ -0,0 +1,103 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
from ._accessor import Accessor
|
||||
from ._includes import Includes
|
||||
from ._output import Output
|
||||
|
||||
|
||||
# TODO: Handle C/C++ standards.
|
||||
@dataclass
|
||||
class Target(yaml.YAMLObject):
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!target"
|
||||
|
||||
name: str
|
||||
type: Output
|
||||
incs: Includes
|
||||
opts: set[Accessor] = field(default_factory=set)
|
||||
defs: set[Accessor] = field(default_factory=set)
|
||||
deps: set[Accessor] = field(default_factory=set)
|
||||
libs: set[Accessor] = field(default_factory=set)
|
||||
srcs: set[Accessor] = field(default_factory=set)
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"name": data.name,
|
||||
"type": data.type,
|
||||
"incs": Includes.to_dict(data.incs),
|
||||
"opts": Accessor.to_list(data.opts),
|
||||
"defs": Accessor.to_list(data.defs),
|
||||
"deps": Accessor.to_list(data.deps),
|
||||
"libs": Accessor.to_list(data.libs),
|
||||
"srcs": Accessor.to_list(data.srcs)
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_list(cls, data: list[Self]) -> list[dict]:
|
||||
return [Target.to_dict(v) for v in data]
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
if not isinstance(data, Target):
|
||||
raise ValueError
|
||||
return dumper.represent_mapping(
|
||||
"tag:yaml.org,2002:map",
|
||||
cls.to_dict(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
def bad_attr(name: str, obj: dict) -> None:
|
||||
raise AttributeError(name=name,obj=obj)
|
||||
|
||||
return cls(**{
|
||||
"name": data.get("name") or bad_attr("name", data),
|
||||
"type": data.get("type") or bad_attr("type", data),
|
||||
"incs": Includes.from_dict(data.get("incs", {})),
|
||||
"opts": set(Accessor.from_list(data.get("opts", []))),
|
||||
"defs": set(Accessor.from_list(data.get("defs", []))),
|
||||
"deps": set(Accessor.from_list(data.get("deps", []))),
|
||||
"libs": set(Accessor.from_list(data.get("libs", []))),
|
||||
"srcs": set(Accessor.from_list(data.get("srcs", [])))
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_list(cls, data: list[dict]) -> list[Self]:
|
||||
return [
|
||||
v if isinstance(v, Target) \
|
||||
else Target.from_dict(v)
|
||||
for v in data
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_dict(
|
||||
loader.construct_mapping(node, deep=True)
|
||||
)
|
||||
|
||||
|
||||
Target.yaml_dumper.add_representer(
|
||||
Target.yaml_tag,
|
||||
Target.to_yaml
|
||||
)
|
||||
|
||||
Target.yaml_loader.add_constructor(
|
||||
Target.yaml_tag,
|
||||
Target.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,86 @@
|
||||
import os
|
||||
import shutil
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from pathlib import Path
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
|
||||
@dataclass
|
||||
class Tool(yaml.YAMLObject):
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!tool"
|
||||
|
||||
name: str
|
||||
path: Path = field(init=False,repr=False)
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.path = self._resolve(self.name)
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
if not isinstance(data, Tool):
|
||||
raise ValueError
|
||||
return dumper.add_representer(None, data.name)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls(loader.construct_scalar(node))
|
||||
|
||||
@staticmethod
|
||||
def _looks_like_path(value: str) -> bool:
|
||||
return (
|
||||
os.sep in value
|
||||
or "/" in value
|
||||
or value.startswith(".")
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _resolve(cls, raw: str) -> Path:
|
||||
return cls._resolve_system_executable(raw) \
|
||||
if not cls._looks_like_path(raw) \
|
||||
else cls._resolve_local_executable(raw)
|
||||
|
||||
@classmethod
|
||||
def _resolve_system_executable(cls, raw: str) -> Path:
|
||||
found = shutil.which(raw)
|
||||
if found is not None:
|
||||
return Path(found)
|
||||
|
||||
raise AssertionError(
|
||||
f"Tool {raw!r} was not found on PATH. "
|
||||
f"Install it or supply an explicit path in the config."
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def _resolve_local_executable(cls, raw: str) -> Path:
|
||||
p = Path(raw)
|
||||
if p.exists():
|
||||
return p.resolve()
|
||||
|
||||
raise FileNotFoundError(
|
||||
f"Tool path {raw!r} does not exist on the filesystem."
|
||||
)
|
||||
|
||||
|
||||
Tool.yaml_dumper.add_representer(
|
||||
Tool.yaml_tag,
|
||||
Tool.to_yaml
|
||||
)
|
||||
|
||||
Tool.yaml_loader.add_constructor(
|
||||
Tool.yaml_tag,
|
||||
Tool.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,91 @@
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import field
|
||||
from typing import Any
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
from ._tool import Tool
|
||||
from ._triple import Triple
|
||||
|
||||
|
||||
@dataclass
|
||||
class Tooling(yaml.YAMLObject):
|
||||
yaml_tag: ClassVar[str] = u"!tooling"
|
||||
yaml_dumper: ClassVar[Any] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[Any] = yaml.SafeLoader
|
||||
|
||||
name: str
|
||||
cc: Tool
|
||||
cxx: Tool
|
||||
ld: Tool
|
||||
ar: Tool
|
||||
when: list[Triple] = field(default_factory=list)
|
||||
|
||||
@property
|
||||
def native(self) -> bool:
|
||||
return not self.when or any(t.native for t in self.when)
|
||||
|
||||
@classmethod
|
||||
def to_dict(cls, data: Self) -> dict:
|
||||
return {
|
||||
"name": data.name,
|
||||
"when": [Triple.to_string(t) for t in data.when],
|
||||
"cc": data.cc,
|
||||
"cxx": data.cxx,
|
||||
"ld": data.ld,
|
||||
"ar": data.ar
|
||||
}
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.YAMLObject
|
||||
) -> yaml.Node:
|
||||
if not isinstance(data, Tooling):
|
||||
raise ValueError
|
||||
return dumper.represent_mapping(
|
||||
"tag:yaml.org,2002:map",
|
||||
cls.to_dict(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, data: dict) -> Self:
|
||||
return cls(**{
|
||||
**data,
|
||||
"when": [
|
||||
v if isinstance(v, Triple) \
|
||||
else Triple.from_string(v)
|
||||
for v in data.get("when", [])
|
||||
]
|
||||
})
|
||||
|
||||
@classmethod
|
||||
def from_list(cls, data: list[dict]) -> list[Self]:
|
||||
return [
|
||||
v if isinstance(v, Tooling) \
|
||||
else Tooling.from_dict(v)
|
||||
for v in data
|
||||
]
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_dict(
|
||||
loader.construct_mapping(node, deep=True)
|
||||
)
|
||||
|
||||
|
||||
Tooling.yaml_dumper.add_representer(
|
||||
Tooling.yaml_tag,
|
||||
Tooling.to_yaml
|
||||
)
|
||||
|
||||
Tooling.yaml_loader.add_constructor(
|
||||
Tooling.yaml_tag,
|
||||
Tooling.from_yaml
|
||||
)
|
||||
@@ -0,0 +1,196 @@
|
||||
import enum
|
||||
import platform
|
||||
import re
|
||||
import sys as system
|
||||
import yaml
|
||||
|
||||
from dataclasses import dataclass
|
||||
from functools import cache
|
||||
from typing import ClassVar
|
||||
from typing import Self
|
||||
|
||||
|
||||
class Arch(enum.StrEnum):
|
||||
X86 = "x86"
|
||||
X64 = "x86_64"
|
||||
ARM = "arm"
|
||||
AARCH64 = "aarch64"
|
||||
RISCV32 = "riscv32"
|
||||
RISCV64 = "riscv64"
|
||||
MIPS = "mips"
|
||||
MIPS64 = "mips64"
|
||||
POWERPC = "powerpc"
|
||||
POWERPC64 = "powerpc64"
|
||||
S390X = "s390x"
|
||||
WASM32 = "wasm32"
|
||||
WASM64 = "wasm64"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
class Sys(enum.StrEnum):
|
||||
LINUX = "linux"
|
||||
MACOS = "macos"
|
||||
WINDOWS = "windows"
|
||||
FREEBSD = "freebsd"
|
||||
OPENBSD = "openbsd"
|
||||
NETBSD = "netbsd"
|
||||
ANDROID = "android"
|
||||
IOS = "ios"
|
||||
WASI = "wasi"
|
||||
NONE = "none"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
class Abi(enum.StrEnum):
|
||||
GNU = "gnu"
|
||||
MSVC = "msvc"
|
||||
MINGW = "mingw"
|
||||
ANDROID = "android"
|
||||
ELF = "elf"
|
||||
NONE = "none"
|
||||
UNKNOWN = "unknown"
|
||||
|
||||
|
||||
@dataclass
|
||||
class Triple(yaml.YAMLObject):
|
||||
__arch_map = {
|
||||
"x86_64": Arch.X64,
|
||||
"amd64": Arch.X64,
|
||||
"i386": Arch.X86,
|
||||
"i686": Arch.X86,
|
||||
"armv7l": Arch.ARM,
|
||||
"armv6l": Arch.ARM,
|
||||
"aarch64": Arch.AARCH64,
|
||||
"arm64": Arch.AARCH64,
|
||||
"riscv32": Arch.RISCV32,
|
||||
"riscv64": Arch.RISCV64,
|
||||
"mips": Arch.MIPS,
|
||||
"mips64": Arch.MIPS64,
|
||||
"ppc": Arch.POWERPC,
|
||||
"ppc64": Arch.POWERPC64,
|
||||
"s390x": Arch.S390X,
|
||||
}
|
||||
|
||||
__sys_map = {
|
||||
"linux": Sys.LINUX,
|
||||
"darwin": Sys.MACOS,
|
||||
"windows": Sys.WINDOWS,
|
||||
"freebsd": Sys.FREEBSD,
|
||||
"openbsd": Sys.OPENBSD,
|
||||
"netbsd": Sys.NETBSD,
|
||||
"wasi": Sys.WASI,
|
||||
}
|
||||
|
||||
yaml_dumper: ClassVar[type[yaml.SafeDumper]] = yaml.SafeDumper
|
||||
yaml_loader: ClassVar[type[yaml.SafeLoader]] = yaml.SafeLoader
|
||||
yaml_tag: ClassVar[str] = u"!triple"
|
||||
|
||||
arch: Arch
|
||||
sys: Sys
|
||||
abi: Abi
|
||||
|
||||
@property
|
||||
def native(self) -> bool:
|
||||
host = Triple.current()
|
||||
return (
|
||||
(self.arch == Arch.UNKNOWN or self.arch == host.arch) and
|
||||
(self.sys == Sys.UNKNOWN or self.sys == host.sys) and
|
||||
(self.abi == Abi.UNKNOWN or self.abi == host.abi)
|
||||
)
|
||||
|
||||
def __str__(self) -> str:
|
||||
return f"{self.arch}-{self.sys}-{self.abi}"
|
||||
|
||||
@classmethod
|
||||
@cache
|
||||
def current(cls) -> Self:
|
||||
arch = cls._select_arch()
|
||||
sys = cls._select_sys()
|
||||
abi = cls._select_abi(arch, sys)
|
||||
return cls(arch, sys, abi)
|
||||
|
||||
@classmethod
|
||||
def to_string(cls, triple: Self) -> str:
|
||||
return str(triple)
|
||||
|
||||
@classmethod
|
||||
def to_yaml(
|
||||
cls,
|
||||
dumper: yaml.SafeDumper,
|
||||
data: yaml.Node
|
||||
) -> Self:
|
||||
if not isinstance(data, Triple):
|
||||
raise ValueError
|
||||
return dumper.represent_scalar(
|
||||
"tag:yaml.org,2002:str",
|
||||
cls.to_string(data)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_string(cls, triple: str) -> Self:
|
||||
parts = triple.strip().split('-')
|
||||
if len(parts) != 3:
|
||||
raise ValueError(f"Triple does not have 3 components: {triple}")
|
||||
|
||||
def _coerce[E: enum.StrEnum](
|
||||
ecls: type[E],
|
||||
value: str,
|
||||
fallback: E
|
||||
) -> E:
|
||||
try:
|
||||
return ecls(value)
|
||||
except ValueError:
|
||||
return fallback
|
||||
|
||||
return cls(
|
||||
arch = _coerce(Arch, parts[0].lower(), Arch.UNKNOWN),
|
||||
sys = _coerce(Sys, parts[1].lower(), Sys.UNKNOWN),
|
||||
abi = _coerce(Abi, parts[2].lower(), Abi.UNKNOWN)
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def from_yaml(
|
||||
cls,
|
||||
loader: yaml.SafeLoader,
|
||||
node: yaml.Node
|
||||
) -> Self:
|
||||
return cls.from_string(loader.construct_scalar(node))
|
||||
|
||||
@classmethod
|
||||
def _select_arch(cls) -> Arch:
|
||||
curr = platform.machine().lower()
|
||||
return cls.__arch_map.get(curr, Arch.UNKNOWN)
|
||||
|
||||
@classmethod
|
||||
def _select_sys(cls) -> Sys:
|
||||
curr = platform.system().lower()
|
||||
return cls.__sys_map.get(curr, Sys.UNKNOWN) \
|
||||
if system.platform != "android" \
|
||||
else Sys.ANDROID
|
||||
|
||||
@classmethod
|
||||
def _select_abi(cls, arch: Arch, sys: Sys) -> Abi:
|
||||
res = Abi.UNKNOWN
|
||||
if sys == Sys.WINDOWS:
|
||||
res = Abi.MSVC
|
||||
elif sys == Sys.ANDROID:
|
||||
res = Abi.ANDROID
|
||||
elif sys in (Sys.LINUX, Sys.FREEBSD, Sys.OPENBSD, Sys.NETBSD):
|
||||
libc = platform.libc_ver()[0].lower()
|
||||
res = Abi.GNU if libc == "glibc" else Abi.ELF
|
||||
return res
|
||||
|
||||
|
||||
Triple.yaml_dumper.add_representer(
|
||||
Triple.yaml_tag,
|
||||
Triple.to_yaml
|
||||
)
|
||||
|
||||
Triple.yaml_loader.add_constructor(
|
||||
Triple.yaml_tag,
|
||||
Triple.from_yaml
|
||||
)
|
||||
|
||||
Triple.yaml_loader.add_implicit_resolver(
|
||||
Triple.yaml_tag,
|
||||
re.compile(r"^[a-z0-9_]+-[a-z0-9_]+-[a-z0-9_]+$"),
|
||||
None
|
||||
)
|
||||
@@ -0,0 +1 @@
|
||||
from ._build import build
|
||||
@@ -0,0 +1,106 @@
|
||||
from pathlib import Path
|
||||
|
||||
from ..compiler import Compiler
|
||||
from ..compiler import CompilerOptions
|
||||
from ..management import latest
|
||||
from ..management import find_target_by_name
|
||||
from ..management import find_project_by_target
|
||||
from ..models import Access
|
||||
from ..models import Accessor
|
||||
from ..models import Output
|
||||
from ..models import Project
|
||||
from ..models import Result
|
||||
from ..models import Target
|
||||
from .._config import Config
|
||||
from .._logger import logger
|
||||
from ._commands import Command
|
||||
from ._commands import CompileArchive
|
||||
from ._commands import CompileDynlib
|
||||
from ._commands import CompileObject
|
||||
from ._commands import CompileProgram
|
||||
from ._except import NoTargetsFoundError
|
||||
|
||||
|
||||
def build(appcfg: Config, project: Project) -> list[Result]:
|
||||
if len(project.targets) == 0:
|
||||
raise NoTargetsFoundError(project)
|
||||
|
||||
# Put together list of commands and then execute them.
|
||||
commands: list[Command] = []
|
||||
results: list[Result] = []
|
||||
for target in project.targets:
|
||||
_prepare_build_commands(appcfg, project, target, commands)
|
||||
|
||||
executed: int = 0
|
||||
cmdcount: int = len(commands)
|
||||
for command in commands:
|
||||
result = command.execute()
|
||||
results.append(result)
|
||||
percentage = (executed / cmdcount)
|
||||
if result.success:
|
||||
logger.info(f"[{percentage:.0%}] Compiled {result.target}")
|
||||
executed += 1
|
||||
else:
|
||||
logger.error(f"[{percentage:.0%}] Failed {result.target}")
|
||||
for diagnostic in result.diagnostics:
|
||||
logger.error(f"Because {diagnostic}")
|
||||
return
|
||||
logger.info(f"[100%] Built {project.cid}")
|
||||
|
||||
|
||||
def _prepare_build_commands(
|
||||
appcfg: Config,
|
||||
project: Project,
|
||||
target: Target,
|
||||
commands: list[Command]
|
||||
) -> None:
|
||||
objects: list[Path] = []
|
||||
|
||||
# TODO: clean this up.
|
||||
compiler = Compiler(appcfg.native_tooling)
|
||||
defopts = CompilerOptions()
|
||||
|
||||
defopts.pic = target.type == Output.DLL
|
||||
defopts.incdirs.update([v.value for v in target.incs.includes])
|
||||
defopts.defines.update([v.value for v in target.defs])
|
||||
defopts.flags.update([v.value for v in target.opts])
|
||||
for file in target.srcs:
|
||||
srcfile = project.origination.src(file.value)
|
||||
dstfile = project.destination.obj(file.value)
|
||||
if latest(srcfile, dstfile):
|
||||
continue
|
||||
commands.append(CompileObject(compiler, defopts, srcfile, dstfile))
|
||||
objects.append(dstfile)
|
||||
|
||||
defopts.objfiles.update(objects)
|
||||
del objects
|
||||
|
||||
defopts.pic = False
|
||||
defopts.defines.clear()
|
||||
if len(target.libs) > 0:
|
||||
for library in target.libs:
|
||||
libtgt = find_target_by_name(library.value)
|
||||
libprj = find_project_by_target(libtgt.name)
|
||||
if libtgt.type == Output.LIB:
|
||||
defopts.libdirs.add(libprj.destination.libpath)
|
||||
defopts.archives.add(library.value)
|
||||
elif libtgt.type == Output.DLL:
|
||||
defopts.objfiles.add(libprj.destination.dll(library.value))
|
||||
|
||||
if len(defopts.objfiles) == 0:
|
||||
return
|
||||
|
||||
command: Command
|
||||
match target.type:
|
||||
case Output.EXE:
|
||||
outfile = project.destination.bin(target.name)
|
||||
command = CompileProgram(compiler, defopts, outfile)
|
||||
case Output.LIB:
|
||||
outfile = project.destination.lib(target.name)
|
||||
command = CompileArchive(compiler, defopts, outfile)
|
||||
case Output.DLL:
|
||||
outfile = project.destination.dll(target.name)
|
||||
command = CompileDynlib(compiler, defopts, outfile)
|
||||
case Output.OBJ:
|
||||
assert False, "Not Implemented"
|
||||
commands.append(command)
|
||||
@@ -0,0 +1,70 @@
|
||||
from abc import ABC
|
||||
from abc import abstractmethod
|
||||
from copy import deepcopy
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
from typing import override
|
||||
|
||||
from ..compiler import Compiler
|
||||
from ..compiler import CompilerOptions
|
||||
from ..models import Result
|
||||
|
||||
|
||||
@dataclass
|
||||
class Command(ABC):
|
||||
compiler: Compiler
|
||||
options: CompilerOptions
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
self.options = deepcopy(self.options)
|
||||
|
||||
@abstractmethod
|
||||
def execute(self) -> Result:
|
||||
...
|
||||
|
||||
|
||||
@dataclass
|
||||
class CompileProgram(Command):
|
||||
exefile: Path
|
||||
|
||||
@override
|
||||
def execute(self) -> Result:
|
||||
return self.compiler.compile_program(
|
||||
self.options,
|
||||
self.exefile
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class CompileArchive(Command):
|
||||
libfile: Path
|
||||
|
||||
@override
|
||||
def execute(self) -> Result:
|
||||
return self.compiler.compile_archive(
|
||||
self.options,
|
||||
self.libfile
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class CompileDynlib(Command):
|
||||
dynfile: Path
|
||||
|
||||
@override
|
||||
def execute(self) -> Result:
|
||||
return self.compiler.compile_library(
|
||||
self.options,
|
||||
self.dynfile
|
||||
)
|
||||
|
||||
@dataclass
|
||||
class CompileObject(Command):
|
||||
srcfile: Path
|
||||
dstfile: Path
|
||||
|
||||
@override
|
||||
def execute(self) -> Result:
|
||||
return self.compiler.compile_object(
|
||||
self.options,
|
||||
self.srcfile,
|
||||
self.dstfile
|
||||
)
|
||||
@@ -0,0 +1,11 @@
|
||||
from ..models import Project
|
||||
|
||||
|
||||
class NoTargetsFoundError(Exception):
|
||||
@property
|
||||
def project(self) -> Project:
|
||||
return self._project
|
||||
|
||||
def __init__(self, project: Project) -> None:
|
||||
super().__init__(f"No targets for project: {project.cid}")
|
||||
self._project = project
|
||||
@@ -0,0 +1 @@
|
||||
from ._execute import execute
|
||||
@@ -0,0 +1,37 @@
|
||||
from subprocess import CalledProcessError
|
||||
from subprocess import PIPE
|
||||
from subprocess import run
|
||||
|
||||
from typing import Callable
|
||||
from typing import Optional
|
||||
DataCallback = Callable[[str, str, str], None]
|
||||
ErrorCallback = Callable[[str, str, str], None]
|
||||
|
||||
|
||||
def execute(
|
||||
command: str,
|
||||
on_data: Optional[DataCallback],
|
||||
on_error: Optional[ErrorCallback]
|
||||
) -> None:
|
||||
try:
|
||||
result = run(
|
||||
command,
|
||||
shell = True,
|
||||
check = True,
|
||||
stdout = PIPE,
|
||||
stderr = PIPE
|
||||
)
|
||||
|
||||
if on_data != None:
|
||||
on_data(
|
||||
command,
|
||||
result.stdout.decode("utf-8"),
|
||||
result.stderr.decode("utf-8")
|
||||
)
|
||||
except CalledProcessError as error:
|
||||
if on_error != None:
|
||||
on_error(
|
||||
error.cmd,
|
||||
error.stdout.decode("utf-8"),
|
||||
error.stderr.decode("utf-8")
|
||||
)
|
||||
Reference in New Issue
Block a user