Compare commits
37 commits
5776c49059
...
076f654a2d
Author | SHA1 | Date | |
---|---|---|---|
076f654a2d | |||
63d30013a3 | |||
e306971534 | |||
2dfdabed26 | |||
5361fbd9a0 | |||
519613eeeb | |||
282e2ac7a1 | |||
3ce03d17c1 | |||
4414a9a712 | |||
959107c245 | |||
73dc9598b7 | |||
0a035f2cd6 | |||
7d0ffb8a65 | |||
2648c2e6ba | |||
d23bbdd7c9 | |||
df2537dd6d | |||
a5ca3a6044 | |||
88434db29a | |||
adbdf47ef4 | |||
9349775193 | |||
aad5c140dd | |||
c702283696 | |||
b44b6bd4d4 | |||
fc3e191dc1 | |||
e84417c077 | |||
517da87ffb | |||
bf5a8a65d7 | |||
67b8d4c981 | |||
470d7aa1e9 | |||
592023d214 | |||
c7db231dc2 | |||
a4f49d6ac0 | |||
fdbd850d23 | |||
b197a15689 | |||
5666177582 | |||
4ab0a575b7 | |||
b969aa886e |
16 changed files with 161 additions and 104 deletions
3
.gitignore
vendored
3
.gitignore
vendored
|
@ -1,4 +1,5 @@
|
|||
__pycache__
|
||||
/tmp
|
||||
/.idea
|
||||
/.venv
|
||||
/.vscode
|
||||
/dist
|
||||
|
|
|
@ -1,32 +1,39 @@
|
|||
[build-system]
|
||||
requires = ['hatchling']
|
||||
build-backend = 'hatchling.build'
|
||||
requires = ["hatchling"]
|
||||
build-backend = "hatchling.build"
|
||||
|
||||
[project]
|
||||
authors = [
|
||||
{ name = 'Marc Beninca', email = 'git@marc.beninca.link' },
|
||||
{ name = "Marc Beninca", email = "git@marc.beninca.link" },
|
||||
]
|
||||
maintainers = [
|
||||
{ name = 'Marc Beninca', email = 'git@marc.beninca.link' },
|
||||
{ name = "Marc Beninca", email = "git@marc.beninca.link" },
|
||||
]
|
||||
classifiers = [
|
||||
'Programming Language :: Python :: 3',
|
||||
'License :: OSI Approved :: GNU Affero General Public License v3',
|
||||
'Operating System :: OS Independent',
|
||||
"Programming Language :: Python :: 3",
|
||||
"License :: OSI Approved :: GNU Affero General Public License v3",
|
||||
"Operating System :: OS Independent",
|
||||
]
|
||||
dependencies = []
|
||||
description = 'Read Write eXecute'
|
||||
dynamic = ['version']
|
||||
description = "Read Write eXecute"
|
||||
dynamic = ["version"]
|
||||
keywords = []
|
||||
license-files = { paths = ['license.md'] }
|
||||
name = 'rwx'
|
||||
readme = 'readme.md'
|
||||
requires-python = '>= 3.10'
|
||||
license-files = { paths = ["license.md"] }
|
||||
name = "rwx"
|
||||
readme = "readme.md"
|
||||
requires-python = ">= 3.11"
|
||||
|
||||
[project.scripts]
|
||||
# command = 'package.module:function'
|
||||
# command = "package.module:function"
|
||||
|
||||
[project.urls]
|
||||
|
||||
[tool.hatch.version]
|
||||
path = 'rwx/__init__.py'
|
||||
path = "rwx/__init__.py"
|
||||
|
||||
[tool.ruff]
|
||||
line-length = 80
|
||||
|
||||
[tool.ruff.lint]
|
||||
ignore = ["COM812", "D203", "D213", "ISC001"]
|
||||
select = ["ALL"]
|
||||
|
|
|
@ -1 +1,3 @@
|
|||
__version__ = '0.0.1'
|
||||
"""Read Write eXecute."""
|
||||
|
||||
__version__ = "0.0.1"
|
||||
|
|
|
@ -1,18 +1,19 @@
|
|||
#! /usr/bin/env python3
|
||||
|
||||
import os
|
||||
"""Entry point."""
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import fs
|
||||
|
||||
|
||||
if __name__ == '__main__':
|
||||
file_path = os.path.realpath(__file__)
|
||||
root_path = os.path.dirname(file_path)
|
||||
directory_path = os.path.join(root_path, 'tmp')
|
||||
file_path = os.path.join(directory_path, 'file')
|
||||
if __name__ == "__main__":
|
||||
file_path = Path(__file__).resolve()
|
||||
root_path = file_path.parent
|
||||
directory_path = root_path / "tmp"
|
||||
file_path = directory_path / "file"
|
||||
|
||||
fs.wipe(directory_path)
|
||||
fs.make_directory(directory_path)
|
||||
fs.write(file_path, 'Martine écrit beaucoup.')
|
||||
fs.write(file_path, "Martine écrit beaucoup.")
|
||||
fs.empty_file(file_path)
|
||||
fs.write(file_path, 'Martine écrit moins.')
|
||||
fs.write(file_path, "Martine écrit moins.")
|
||||
|
|
|
@ -1,6 +1,9 @@
|
|||
"""Handle system arguments."""
|
||||
|
||||
import sys
|
||||
|
||||
|
||||
def split() -> tuple[str, list[str]]:
|
||||
"""Split command & actual arguments."""
|
||||
command, *arguments = sys.argv
|
||||
return command, arguments
|
||||
|
|
|
@ -1,15 +1,17 @@
|
|||
"""Handle system commands & packages."""
|
||||
|
||||
commands: list[str] = []
|
||||
packages: list[str] = []
|
||||
|
||||
|
||||
def need(command: str) -> None:
|
||||
"""Assert package dependency for a command."""
|
||||
match command:
|
||||
case 'debootstrap':
|
||||
package = 'debootstrap'
|
||||
case 'mksquashfs' | 'unsquashfs':
|
||||
package = 'squashfs-tools'
|
||||
case "debootstrap":
|
||||
package = "debootstrap"
|
||||
case "mksquashfs" | "unsquashfs":
|
||||
package = "squashfs-tools"
|
||||
case _:
|
||||
package = None
|
||||
if package:
|
||||
if package not in packages:
|
||||
if package and package not in packages:
|
||||
packages.append(package)
|
||||
|
|
|
@ -1,14 +1,19 @@
|
|||
import ps
|
||||
|
||||
import rwx.cmd
|
||||
|
||||
rwx.cmd.need('mksquashfs')
|
||||
rwx.cmd.need("mksquashfs")
|
||||
|
||||
|
||||
def mksquashfs(input_root: str, output_file: str):
|
||||
ps.run([
|
||||
'mksquashfs',
|
||||
ps.run(
|
||||
[
|
||||
"mksquashfs",
|
||||
input_root,
|
||||
output_file,
|
||||
'-comp', 'zstd',
|
||||
'-Xcompression-level', str(18),
|
||||
])
|
||||
"-comp",
|
||||
"zstd",
|
||||
"-Xcompression-level",
|
||||
str(18),
|
||||
]
|
||||
)
|
||||
|
|
|
@ -1,20 +1,20 @@
|
|||
import cmd
|
||||
|
||||
import ps
|
||||
|
||||
cmd.need('debootstrap')
|
||||
cmd.need("debootstrap")
|
||||
|
||||
BOOTSTRAP_ARCHITECTURE = 'amd64'
|
||||
BOOTSTRAP_VARIANT = 'minbase'
|
||||
BOOTSTRAP_ARCHITECTURE = "amd64"
|
||||
BOOTSTRAP_VARIANT = "minbase"
|
||||
|
||||
|
||||
def bootstrap(root_path: str, suite: str, mirror_location: str):
|
||||
command = [
|
||||
('debootstrap',),
|
||||
('--arch', BOOTSTRAP_ARCHITECTURE),
|
||||
('--variant', BOOTSTRAP_VARIANT),
|
||||
("debootstrap",),
|
||||
("--arch", BOOTSTRAP_ARCHITECTURE),
|
||||
("--variant", BOOTSTRAP_VARIANT),
|
||||
(suite,),
|
||||
(root_path,),
|
||||
(mirror_location,),
|
||||
]
|
||||
completed_process = ps.run(command)
|
||||
return completed_process
|
||||
return ps.run(command)
|
||||
|
|
2
rwx/err/__init__.py
Normal file
2
rwx/err/__init__.py
Normal file
|
@ -0,0 +1,2 @@
|
|||
class Exception(Exception):
|
||||
pass
|
|
@ -1,36 +1,41 @@
|
|||
"""Operations involving FileSystems."""
|
||||
|
||||
import os
|
||||
import shutil
|
||||
|
||||
from rwx import ps
|
||||
|
||||
CHARSET = 'UTF-8'
|
||||
CHARSET = "UTF-8"
|
||||
|
||||
|
||||
def create_image(file_path: str, size_bytes: int):
|
||||
def create_image(file_path: str, size_bytes: int) -> None:
|
||||
"""Create a virtual device image file."""
|
||||
ps.run(
|
||||
('qemu-img', 'create'),
|
||||
('-f', 'qcow2'),
|
||||
("qemu-img", "create"),
|
||||
("-f", "qcow2"),
|
||||
(file_path, size_bytes),
|
||||
)
|
||||
|
||||
|
||||
def empty_file(path: str):
|
||||
write(path, str())
|
||||
def empty_file(path: str) -> None:
|
||||
"""Empty the file at provided path."""
|
||||
write(path, "")
|
||||
|
||||
|
||||
def get_mount_uuid(path: str):
|
||||
def get_mount_uuid(path: str) -> str:
|
||||
"""Return the UUID of provided mountpoint path."""
|
||||
return ps.run_line(
|
||||
('findmnt',),
|
||||
('--noheadings',),
|
||||
('--output', 'UUID'),
|
||||
("findmnt",),
|
||||
("--noheadings",),
|
||||
("--output", "UUID"),
|
||||
(path,),
|
||||
)
|
||||
|
||||
|
||||
def get_path_mount(path: str):
|
||||
return ps.run_line(
|
||||
('stat',),
|
||||
('--format', '%m'),
|
||||
("stat",),
|
||||
("--format", "%m"),
|
||||
(path,),
|
||||
)
|
||||
|
||||
|
@ -44,7 +49,7 @@ def make_directory(path: str):
|
|||
|
||||
|
||||
def read_file(file_path: str):
|
||||
with open(file_path, 'br') as file_object:
|
||||
with open(file_path, "br") as file_object:
|
||||
return file_object.read()
|
||||
|
||||
|
||||
|
@ -66,5 +71,5 @@ def wipe(path: str):
|
|||
|
||||
|
||||
def write(file_path: str, text: str, charset=CHARSET):
|
||||
with open(file_path, 'bw') as file_object:
|
||||
with open(file_path, "bw") as file_object:
|
||||
file_object.write(text.encode(charset))
|
||||
|
|
|
@ -1,33 +1,39 @@
|
|||
import cmd
|
||||
|
||||
import ps
|
||||
|
||||
cmd.need('grub-mkimage')
|
||||
cmd.need("grub-mkimage")
|
||||
|
||||
COMPRESSION = 'xz'
|
||||
COMPRESSION = "xz"
|
||||
ENV_BYTES = 1024
|
||||
ENV_COMMENT = '#'
|
||||
ENV_HEADER = f'''{ENV_COMMENT} GRUB Environment Block
|
||||
'''
|
||||
ENV_COMMENT = "#"
|
||||
ENV_HEADER = f"""{ENV_COMMENT} GRUB Environment Block
|
||||
"""
|
||||
MODULES = {
|
||||
'i386-pc': [
|
||||
('biosdisk',),
|
||||
('ntldr',),
|
||||
]
|
||||
"i386-pc": [
|
||||
("biosdisk",),
|
||||
("ntldr",),
|
||||
],
|
||||
}
|
||||
|
||||
|
||||
def make_image(image_format: str, image_path: str, modules: list[str],
|
||||
memdisk_path: str, pubkey_path: str = None) -> None:
|
||||
def make_image(
|
||||
image_format: str,
|
||||
image_path: str,
|
||||
modules: list[str],
|
||||
memdisk_path: str,
|
||||
pubkey_path: str | None = None,
|
||||
) -> None:
|
||||
args = [
|
||||
('grub-mkimage',),
|
||||
('--compress', COMPRESSION),
|
||||
('--format', image_format),
|
||||
('--output', image_path),
|
||||
('--memdisk', memdisk_path),
|
||||
("grub-mkimage",),
|
||||
("--compress", COMPRESSION),
|
||||
("--format", image_format),
|
||||
("--output", image_path),
|
||||
("--memdisk", memdisk_path),
|
||||
]
|
||||
if pubkey_path:
|
||||
args.append(('--pubkey', pubkey_path))
|
||||
args.append(("--pubkey", pubkey_path))
|
||||
args.extend(modules)
|
||||
if modules := MODULES.get(image_format, None):
|
||||
if modules := MODULES.get(image_format):
|
||||
args.extend(modules)
|
||||
ps.run(*args)
|
||||
|
|
|
@ -2,21 +2,31 @@ import logging
|
|||
import sys
|
||||
|
||||
|
||||
def get_logger(name: str) -> logging.Logger:
|
||||
def get_file_logger(name: str) -> logging.Logger:
|
||||
formatter = logging.Formatter(
|
||||
"%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d >>> %(message)s"
|
||||
"%(name)s: %(asctime)s | %(levelname)s | %(filename)s:%(lineno)s | %(process)d >>> %(message)s",
|
||||
)
|
||||
|
||||
# file_handler = logging.FileHandler('log.txt')
|
||||
# file_handler.setFormatter(formatter)
|
||||
# file_handler.setLevel(logging.INFO)
|
||||
|
||||
#
|
||||
out_handler = logging.StreamHandler(stream=sys.stdout)
|
||||
out_handler.setFormatter(formatter)
|
||||
out_handler.setLevel(logging.INFO)
|
||||
|
||||
#
|
||||
logger = logging.getLogger(name)
|
||||
# logger.addHandler(file_handler)
|
||||
logger.addHandler(out_handler)
|
||||
logger.setLevel(logging.INFO)
|
||||
#
|
||||
return logger
|
||||
|
||||
|
||||
def get_stream_logger(level: int) -> logging.Logger:
|
||||
out_handler = logging.StreamHandler(stream=sys.stdout)
|
||||
out_handler.setLevel(level)
|
||||
#
|
||||
logger = logging.getLogger()
|
||||
logger.addHandler(out_handler)
|
||||
logger.setLevel(level)
|
||||
#
|
||||
return logger
|
||||
|
||||
|
||||
stream = get_stream_logger(logging.INFO)
|
||||
|
|
|
@ -2,7 +2,7 @@ from os import path
|
|||
|
||||
|
||||
class Project:
|
||||
def __init__(self, file_path: str):
|
||||
def __init__(self, file_path: str) -> None:
|
||||
self.file: str = path.realpath(file_path)
|
||||
self.root: str = path.dirname(self.file)
|
||||
self.name: str = path.basename(self.root)
|
||||
|
|
|
@ -1,4 +1,7 @@
|
|||
"""Project consisting only of a Sphinx documentation."""
|
||||
|
||||
from os import path
|
||||
|
||||
from sphinx.cmd.build import build_main
|
||||
|
||||
from rwx.fs import wipe
|
||||
|
@ -6,22 +9,28 @@ from rwx.prj import Project
|
|||
|
||||
|
||||
class SphinxProject(Project):
|
||||
def __init__(self, file_path: str):
|
||||
def __init__(self, file_path: str) -> None:
|
||||
super().__init__(file_path)
|
||||
|
||||
def build(self):
|
||||
output_root: str = path.join(self.root, 'out')
|
||||
def build(self) -> None:
|
||||
output_root: str = path.join(self.root, "out")
|
||||
wipe(output_root)
|
||||
arguments: list[str] = [
|
||||
"-E",
|
||||
"-j", "2",
|
||||
"-b", "html",
|
||||
"-D", f"project={self.name}",
|
||||
"-D", "master_doc={}".format("index"),
|
||||
"-D", "html_theme={}".format("sphinx_rtd_theme"),
|
||||
"-c", self.root,
|
||||
"-j",
|
||||
"2",
|
||||
"-b",
|
||||
"html",
|
||||
"-D",
|
||||
f"project={self.name}",
|
||||
"-D",
|
||||
"master_doc={}".format("index"),
|
||||
"-D",
|
||||
"html_theme={}".format("sphinx_rtd_theme"),
|
||||
"-c",
|
||||
self.root,
|
||||
# "-C",
|
||||
path.join(self.root, self.name),
|
||||
path.join(output_root, self.name),
|
||||
path.join(output_root, "web"),
|
||||
]
|
||||
build_main(arguments)
|
||||
|
|
|
@ -14,7 +14,9 @@ def get_tuples_args(tuples) -> list[str]:
|
|||
|
||||
|
||||
def run(*tuples) -> subprocess.CompletedProcess:
|
||||
return subprocess.run(get_tuples_args(tuples), capture_output=False)
|
||||
return subprocess.run(
|
||||
get_tuples_args(tuples), capture_output=False, check=True
|
||||
)
|
||||
|
||||
|
||||
def run_line(*tuples, charset: str = txt.CHARSET) -> str:
|
||||
|
@ -23,6 +25,8 @@ def run_line(*tuples, charset: str = txt.CHARSET) -> str:
|
|||
|
||||
|
||||
def run_lines(*tuples, charset: str = txt.CHARSET) -> list[str]:
|
||||
process = subprocess.run(get_tuples_args(tuples), capture_output=True)
|
||||
process = subprocess.run(
|
||||
get_tuples_args(tuples), capture_output=True, check=True
|
||||
)
|
||||
string = process.stdout.decode(charset)
|
||||
return string.rstrip().splitlines()
|
||||
|
|
|
@ -1 +1 @@
|
|||
CHARSET = 'UTF-8'
|
||||
CHARSET = "UTF-8"
|
||||
|
|
Loading…
Reference in a new issue