Script Python pour extraire les éléments communs d'entrées YAML dans un élément de base dérivable
Ce script a été développé pour optimiser les définitions YAML de kicad-footprint-generator.
Il scanne un fichier YAML d’empreinte, trouve les valeurs identiques dans la plupart des entrées, et les déplace dans un seul élément de base partagé. Chaque empreinte individuelle ne conserve alors que les paramètres qui diffèrent réellement de la base, ce qui raccourcit le fichier et réduit la duplication.
Par exemple, avec une entrée minimale comme celle-ci :
original.yaml
Vertical:
pad_width: 1.0
pad_height: 2.0
solder_mask_margin: 0.05
Horizontal:
pad_width: 1.0
pad_height: 2.0
solder_mask_margin: 0.05
special_option: truele script peut extraire les valeurs communes dans une base et réécrire le fichier à peu près comme suit :
processed.yaml
base: &base
pad_width: 1.0
pad_height: 2.0
solder_mask_margin: 0.05
Vertical:
<<: *base
Horizontal:
<<: *base
special_option: trueC’est essentiellement ce qu’il fait pour les YAML beaucoup plus grands du générateur KiCad.
yaml_extract_common_base.py
#!/usr/bin/env python3
# SPDX-FileCopyrightText: 2025 Uli Köhler <[email protected]>
# SPDX-License-Identifier: CC0-1.0
"""Extraire et consolider les valeurs communes des fichiers YAML du générateur KiCad."""
from __future__ import annotations
import argparse
import copy
import importlib
import io
import os
import sys
from dataclasses import dataclass
from pathlib import Path
from typing import List, Optional
try:
_ruamel_yaml = importlib.import_module("ruamel.yaml")
_ruamel_comments = importlib.import_module("ruamel.yaml.comments")
except ImportError as exc: # pragma: no cover - dépendance manquante à l'exécution
raise SystemExit(
"Le paquet ruamel.yaml est requis. Installez-le via 'pip install ruamel.yaml'."
) from exc
YAML = _ruamel_yaml.YAML
CommentedMap = _ruamel_comments.CommentedMap
CommentedSeq = _ruamel_comments.CommentedSeq
BASE_COMMENT_MARKER = "Configuration de base extraite par YAMLCommon"
RT_YAML = YAML(typ="rt")
RT_YAML.preserve_quotes = True
RT_YAML.allow_duplicate_keys = True
RT_YAML.width = 4096
RT_YAML.indent(mapping=2, sequence=4, offset=2)
@dataclass
class Entry:
"""Représente une seule entrée d'empreinte dans le fichier YAML personnalisé."""
leading_lines: List[str]
separator_lines: List[str]
name_line: Optional[str]
after_name_lines: List[str]
root_key: str
inner_map: CommentedMap
outer_map: CommentedMap
inherits_base: bool = False
defines_anchor: bool = False
@dataclass
class ParsedFile:
"""Contient les éléments structurels du fichier analysé."""
prefix_lines: List[str]
entries: List[Entry]
suffix_lines: List[str]
existing_base: Optional[CommentedMap] = None
def is_block_start(line: str) -> bool:
"""Retourne True si *line* marque le début d'un bloc d'empreinte."""
stripped = line.strip()
if not stripped or line.startswith(" "):
return False
if ":" not in stripped:
return False
key = stripped.split(":", 1)[0].strip()
return bool(key)
def is_name_line(line: str) -> bool:
"""Heuristique qui reconnaît la ligne de nom d'empreinte."""
stripped = line.strip()
if not stripped or ":" in stripped:
return False
if stripped.startswith("#") or stripped.startswith("-"):
return False
return line.startswith(" ")
def parse_file(text: str, anchor_name: str) -> ParsedFile:
"""Analyser le fichier YAML non standard en entrées structurées."""
lines = text.splitlines(keepends=True)
idx = 0
prefix_lines: List[str] = []
suffix_lines: List[str] = []
entries: List[Entry] = []
first_block = True
while True:
interim: List[str] = []
while idx < len(lines) and not is_block_start(lines[idx]):
interim.append(lines[idx])
idx += 1
if idx >= len(lines):
if first_block:
prefix_lines = interim
else:
suffix_lines = interim
break
if first_block:
prefix_lines = interim
leading_lines: List[str] = []
first_block = False
else:
leading_lines = interim
block_lines: List[str] = []
while idx < len(lines) and lines[idx].strip() != "":
block_lines.append(lines[idx])
idx += 1
defines_anchor = any(f"&{anchor_name}" in line for line in block_lines)
block_lines, inherits_base = strip_merge_marker(block_lines, anchor_name)
block_text = "".join(block_lines)
if not block_text.strip():
raise ValueError("Bloc vide rencontré ; le format de fichier n'est peut-être pas pris en charge.")
separator_lines: List[str] = []
while idx < len(lines) and lines[idx].strip() == "":
separator_lines.append(lines[idx])
idx += 1
while idx < len(lines) and lines[idx].lstrip().startswith("#"):
separator_lines.append(lines[idx])
idx += 1
name_line: Optional[str] = None
if idx < len(lines) and is_name_line(lines[idx]):
name_line = lines[idx]
idx += 1
after_name_lines: List[str] = []
while idx < len(lines) and not is_block_start(lines[idx]):
after_name_lines.append(lines[idx])
idx += 1
data = RT_YAML.load(block_text)
if not isinstance(data, CommentedMap) or len(data) != 1:
raise ValueError("Chaque bloc doit être un mapping avec une seule clé racine (par ex. Vertical).")
root_key = next(iter(data))
inner_map = data[root_key]
if not isinstance(inner_map, CommentedMap):
raise ValueError("Les entrées de premier niveau doivent correspondre à un dictionnaire de paramètres.")
entries.append(
Entry(
leading_lines=leading_lines,
separator_lines=separator_lines,
name_line=name_line,
after_name_lines=after_name_lines,
root_key=root_key,
inner_map=inner_map,
outer_map=data,
inherits_base=inherits_base,
defines_anchor=defines_anchor,
)
)
return ParsedFile(prefix_lines=prefix_lines, entries=entries, suffix_lines=suffix_lines)
def strip_merge_marker(block_lines: List[str], anchor_name: str) -> tuple[List[str], bool]:
"""Supprimer les références de fusion vers l'ancre partagée d'un bloc YAML."""
filtered: List[str] = []
removed = False
target = f"<<: *{anchor_name}"
for line in block_lines:
if line.strip() == target:
removed = True
continue
filtered.append(line)
return filtered, removed
def strip_generated_base(parsed: ParsedFile, anchor_name: str) -> ParsedFile:
"""Supprimer tout bloc de base précédemment généré et le commentaire associé."""
clean_prefix = [line for line in parsed.prefix_lines if BASE_COMMENT_MARKER not in line]
filtered_entries: List[Entry] = []
base_map: Optional[CommentedMap] = parsed.existing_base
for entry in parsed.entries:
if entry.defines_anchor:
base_map = entry.inner_map
continue
filtered_entries.append(entry)
return ParsedFile(
prefix_lines=clean_prefix,
entries=filtered_entries,
suffix_lines=parsed.suffix_lines,
existing_base=base_map,
)
def ensure_base_inheritance(parsed: ParsedFile) -> None:
"""Marquer les entrées comme héritant de la base si elles manquent de clés de base."""
if not parsed.existing_base:
return
for entry in parsed.entries:
if entry.inherits_base:
continue
if not entry_supports_base(entry.inner_map, parsed.existing_base):
entry.inherits_base = True
def normalize(value):
"""Convertir les nœuds ruamel en objets Python standards pour les tests d'égalité."""
if isinstance(value, CommentedMap):
return {k: normalize(v) for k, v in value.items()}
if isinstance(value, CommentedSeq):
return [normalize(v) for v in value]
return value
def merged_with_base(base_map: CommentedMap, overrides: CommentedMap) -> CommentedMap:
"""Retourner une copie profonde de *base_map* mise à jour avec *overrides*."""
result = copy.deepcopy(base_map)
for key, value in overrides.items():
if (
key in result
and isinstance(result[key], CommentedMap)
and isinstance(value, CommentedMap)
):
result[key] = merged_with_base(result[key], value)
else:
result[key] = copy.deepcopy(value)
return result
def values_equal(lhs, rhs) -> bool:
"""Égalité profonde qui ignore les métadonnées spécifiques à ruamel."""
return normalize(lhs) == normalize(rhs)
def _round_number(value, digits: Optional[int]):
"""Arrondir la valeur numérique *value* à *digits* décimales si demandé.
Si *digits* est None, la valeur est retournée inchangée. Les valeurs non
numériques sont également retournées inchangées.
"""
if digits is None:
return value
try:
# Traiter les entiers/flottants tout en laissant les autres types inchangés
if isinstance(value, (int, float)):
return round(value, digits)
except TypeError:
pass
return value
def _round_node(node, digits: Optional[int]):
"""Arrondir récursivement tous les scalaires numériques dans *node* sur place."""
if digits is None:
return node
if isinstance(node, CommentedMap):
for key, val in list(node.items()):
node[key] = _round_node(val, digits)
return node
if isinstance(node, CommentedSeq):
for idx, val in enumerate(list(node)):
node[idx] = _round_node(val, digits)
return node
return _round_number(node, digits)
def collect_common(
entries: List[Entry],
threshold: float,
fallback_base: Optional[CommentedMap] = None,
) -> CommentedMap:
"""Retourner un mapping avec les clés communes à toutes les entrées."""
total = len(entries)
result = CommentedMap()
if total == 0:
return result
occurrences: dict[str, List[object]] = {}
for entry in entries:
items = entry.inner_map.items()
if entry.inherits_base and fallback_base is not None:
combined = merged_with_base(fallback_base, entry.inner_map)
items = combined.items()
for key, value in items:
occurrences.setdefault(key, []).append(value)
for key, values in occurrences.items():
ratio = len(values) / total
if ratio < threshold:
continue
first = values[0]
if all(values_equal(first, other) for other in values[1:]):
result[key] = copy.deepcopy(first)
return result
def entry_supports_base(entry_map: CommentedMap, base_map: CommentedMap) -> bool:
"""Vérifier si *entry_map* contient toutes les clés trouvées dans *base_map*."""
for key, base_value in base_map.items():
if key not in entry_map:
return False
entry_value = entry_map[key]
if isinstance(base_value, CommentedMap) and isinstance(entry_value, CommentedMap):
if not entry_supports_base(entry_value, base_value):
return False
elif isinstance(base_value, CommentedSeq) and isinstance(entry_value, CommentedSeq):
if not values_equal(entry_value, base_value):
return False
else:
if not values_equal(entry_value, base_value):
return False
return True
def build_overrides(entry_map: CommentedMap, base_map: CommentedMap) -> CommentedMap:
"""Retourner le sous-ensemble de *entry_map* qui diffère de *base_map*."""
overrides = CommentedMap()
for key, entry_value in entry_map.items():
if key not in base_map:
overrides[key] = copy.deepcopy(entry_value)
continue
base_value = base_map[key]
if isinstance(entry_value, CommentedMap) and isinstance(base_value, CommentedMap):
child = build_overrides(entry_value, base_value)
if child:
overrides[key] = child
continue
if isinstance(entry_value, CommentedSeq) and isinstance(base_value, CommentedSeq):
if not values_equal(entry_value, base_value):
overrides[key] = copy.deepcopy(entry_value)
continue
if not values_equal(entry_value, base_value):
overrides[key] = copy.deepcopy(entry_value)
return overrides
def dump_map(node: CommentedMap) -> str:
"""Convertir *node* en texte YAML."""
buffer = io.StringIO()
RT_YAML.dump(node, buffer)
return buffer.getvalue()
def _supports_color(stream) -> bool:
"""Retourner True si *stream* prend probablement en charge les codes de couleur ANSI."""
return bool(getattr(stream, "isatty", lambda: False)()) and os.environ.get("TERM") not in {None, "", "dumb"}
def _colorize(text: str, color_code: str = "31") -> str:
"""Envelopper *text* dans des codes de couleur ANSI si la stderr actuelle le prend en charge."""
if _supports_color(sys.stderr):
return f"\033[{color_code}m{text}\033[0m"
return text
def inject_merge_line(rendered: str, anchor_name: str, indent: str = " ") -> str:
"""Insérer une référence de fusion YAML juste après la clé racine."""
lines = rendered.splitlines(keepends=True)
if not lines:
return f"{indent}<<: *{anchor_name}\n"
if not lines[0].endswith("\n"):
lines[0] += "\n"
merge_line = f"{indent}<<: *{anchor_name}\n"
lines.insert(1, merge_line)
return "".join(lines)
def render_with_merge(root_key: str, overrides: CommentedMap, anchor_name: str) -> str:
"""Convertir un mapping d'entrée en YAML précédé d'une référence de fusion."""
if overrides:
container = CommentedMap()
container[root_key] = overrides
return inject_merge_line(dump_map(container), anchor_name)
return f"{root_key}:\n <<: *{anchor_name}\n"
def rewrite(
parsed: ParsedFile,
base_map: Optional[CommentedMap],
threshold: float,
anchor_name: str,
round_digits: Optional[int],
target_path: Path,
) -> None:
"""Réécrire le fichier YAML avec un bloc de base suivi d'overrides par entrée."""
fragments: List[str] = []
fragments.extend(parsed.prefix_lines)
base_inner = None
if base_map and len(base_map):
base_inner = copy.deepcopy(base_map)
# Appliquer l'arrondi à la base partagée avant l'ancrage, si demandé.
if round_digits is not None:
_round_node(base_inner, round_digits)
base_inner.yaml_set_anchor(anchor_name, always_dump=True)
# L'élément de base doit toujours être nommé "base" dans le
# YAML généré, indépendamment de la clé racine d'origine.
base_container = CommentedMap()
base_container["base"] = base_inner
base_container.yaml_set_start_comment(
f"Configuration de base extraite par YAMLCommon (threshold={threshold:.2f})",
indent=0,
)
fragments.append(dump_map(base_container))
fragments.append("\n")
for entry in parsed.entries:
fragments.extend(entry.leading_lines)
# Si nous avons une base générée et que l'entrée la prend explicitement
# en charge ou a été précédemment marquée comme héritant de la base
# (par ex. parce qu'il lui manquait des clés de base), toujours convertir
# l'entrée en utilisant une référence de fusion. Cela garantit que chaque
# entrée qui a supprimé des valeurs par rapport à la base est toujours
# déclarée comme en dérivant.
entry_map_for_write = copy.deepcopy(entry.outer_map)
if base_inner and (entry_supports_base(entry.inner_map, base_inner) or entry.inherits_base):
# Construire les overrides par rapport à la base (déjà arrondie).
inner_copy = copy.deepcopy(entry.inner_map)
if round_digits is not None:
_round_node(inner_copy, round_digits)
overrides = build_overrides(inner_copy, base_inner)
# S'il n'y a aucun override (i.e. l'entrée est identique à la
# base), émettre tout de même une fusion pour que l'entrée reste présente.
fragments.append(render_with_merge(entry.root_key, overrides, anchor_name))
else:
# Pas de base ou non prise en charge par la base : arrondir tout le mapping d'entrée.
if round_digits is not None:
_round_node(entry_map_for_write, round_digits)
fragments.append(dump_map(entry_map_for_write))
fragments.extend(entry.separator_lines)
if entry.name_line:
fragments.append(entry.name_line if entry.name_line.endswith("\n") else entry.name_line + "\n")
fragments.extend(entry.after_name_lines)
fragments.extend(parsed.suffix_lines)
target_path.write_text("".join(fragments), encoding="utf-8")
def print_base(base_map: Optional[CommentedMap], threshold: float) -> None:
"""Afficher l'élément de base extrait sur stdout."""
if not base_map:
print(f"Aucun attribut de base n'a atteint le seuil ({threshold:.2f}).")
return
container = CommentedMap()
container["base"] = base_map
print(f"Élément de base extrait (threshold={threshold:.2f}) :")
sys.stdout.write(dump_map(container))
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("yaml_file", type=Path, help="Chemin vers le fichier YAML d'empreinte")
parser.add_argument(
"-t",
"--threshold",
type=float,
default=0.9,
help="Fraction d'entrées dans lesquelles une clé doit apparaître (par défaut : 0.9)",
)
parser.add_argument(
"-r",
"--round",
type=int,
default=None,
help=(
"Arrondir toutes les valeurs numériques au nombre donné de décimales "
"avant d'écrire la sortie."
),
)
parser.add_argument(
"-o",
"--output",
type=Path,
help=(
"Chemin de sortie pour le fichier YAML réécrit (par défaut : écraser le fichier d'entrée, "
"requiert --overwrite)."
),
)
parser.add_argument(
"--overwrite",
action="store_true",
help=(
"Autoriser l'écrasement du fichier YAML original ; effectif uniquement lorsque "
"--output est défini sur le chemin du fichier original."
),
)
parser.add_argument(
"--anchor-name",
default="base",
help="Nom d'ancre utilisé pour le mapping de base partagé (par défaut : base)",
)
return parser.parse_args()
def main() -> None:
args = parse_args()
if not 0 < args.threshold <= 1:
raise SystemExit("Le seuil doit être compris dans (0, 1].")
text = args.yaml_file.read_text(encoding="utf-8")
parsed = parse_file(text, args.anchor_name)
parsed = strip_generated_base(parsed, args.anchor_name)
ensure_base_inheritance(parsed)
if not parsed.entries:
raise SystemExit("Aucune entrée n'a été trouvée dans le fichier fourni.")
common_map = collect_common(parsed.entries, args.threshold, parsed.existing_base)
if parsed.existing_base:
base_for_write = copy.deepcopy(parsed.existing_base)
if common_map:
additional = build_overrides(common_map, base_for_write)
if additional:
base_for_write = merged_with_base(base_for_write, additional)
else:
base_for_write = copy.deepcopy(common_map) if common_map else None
print_base(base_for_write, args.threshold)
target = args.output or args.yaml_file
write_requires_overwrite = target.resolve() == args.yaml_file.resolve()
if write_requires_overwrite and not args.overwrite:
warning = (
"--overwrite n'a pas été fourni ; le fichier d'entrée a été laissé inchangé et aucune sortie n'a été écrite."
)
print(_colorize(warning), file=sys.stderr)
return
rewrite(
parsed,
base_for_write if base_for_write else None,
args.threshold,
args.anchor_name,
args.round,
target,
)
if __name__ == "__main__":
main()Check out similar posts by category:
Python
If this post helped you, please consider buying me a coffee or donating via PayPal to support research & publishing of new posts on TechOverflow