refactor(scripts): make hardware utility scripts version-agnostic with argparse

All 4 scripts now accept CLI arguments via argparse instead of hardcoded
paths, making them reusable across hardware versions and ready for MCP
server integration:

- extract_easyeda_db.py: --db-path, --output-dir, --verbose
- extract_bom.py: --extract-root or --epro-dir, --output-dir, --versions
- generate_jlcpcb_bom.py: --bom-dir, --version, --output-file
- convert_to_kicad.py: --bom-dir, --kicad-lib-dir, --versions

Core logic in each script is refactored into callable functions that
return results, enabling programmatic use from MCP tools.
This commit is contained in:
Fábio Cunha 2026-05-25 18:05:29 +01:00
parent 0a39905348
commit 30b85f346a
4 changed files with 328 additions and 70 deletions

View file

@ -3,26 +3,39 @@
Reads the extracted BOM knowledge base and converts each unique LCSC
component into KiCad symbol + footprint + 3D model files.
Usage:
python convert_to_kicad.py --bom-dir ./bom_output --kicad-lib-dir ./kicad/libs
python convert_to_kicad.py --help
"""
import argparse
import json
import subprocess
import sys
import time
from pathlib import Path
BOM_DIR = Path("/tmp/epro_bom_output")
KICAD_LIB_DIR = Path("/home/fcunha/savearth/flow-meter-pcb/kicad/libs")
VERSIONS = ["2_3", "2_4", "2_5", "2_6", "2_7", "2_8"]
DEFAULT_VERSIONS = ["2_3", "2_4", "2_5", "2_6", "2_7", "2_8"]
def main():
KICAD_LIB_DIR.mkdir(parents=True, exist_ok=True)
def convert_to_kicad(bom_dir: Path, kicad_lib_dir: Path, versions: list):
"""Convert LCSC components to KiCad library files.
Args:
bom_dir: Directory containing bom_v{version}.json files.
kicad_lib_dir: Output directory for KiCad library files.
versions: List of version suffixes to process.
Returns:
tuple: (success_count, failed_list)
"""
kicad_lib_dir.mkdir(parents=True, exist_ok=True)
# Collect all unique LCSC parts across all versions
lcsc_parts = {} # lcsc_id -> component info (latest version wins)
for version in VERSIONS:
bom_file = BOM_DIR / f"bom_v{version}.json"
for version in versions:
bom_file = bom_dir / f"bom_v{version}.json"
if not bom_file.exists():
continue
with open(bom_file) as f:
@ -33,12 +46,12 @@ def main():
lcsc_parts[lcsc] = comp
print(f"Found {len(lcsc_parts)} unique LCSC parts to convert")
print(f"Output: {KICAD_LIB_DIR}")
print(f"Output: {kicad_lib_dir}")
print()
output_sym = KICAD_LIB_DIR / "savearth.kicad_sym"
output_fp = KICAD_LIB_DIR / "savearth.pretty"
output_3d = KICAD_LIB_DIR / "savearth.3dshapes"
output_sym = kicad_lib_dir / "savearth.kicad_sym"
output_fp = kicad_lib_dir / "savearth.pretty"
output_3d = kicad_lib_dir / "savearth.3dshapes"
output_fp.mkdir(parents=True, exist_ok=True)
output_3d.mkdir(parents=True, exist_ok=True)
@ -103,6 +116,45 @@ def main():
for p in sorted(output_3d.glob("*")):
print(f" 3D model: {p.name}")
return success, failed
def build_parser():
"""Build and return the argument parser."""
parser = argparse.ArgumentParser(
description="Convert LCSC components from BOM to KiCad library files using easyeda2kicad.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
Examples:
%(prog)s --bom-dir ./bom_output --kicad-lib-dir ./kicad/libs
%(prog)s --bom-dir ./bom_output --kicad-lib-dir ./libs --versions 2_7 2_8
"""
)
parser.add_argument(
"--bom-dir", required=True,
help="Directory containing bom_v{version}.json files (produced by extract_bom.py)."
)
parser.add_argument(
"--kicad-lib-dir", required=True,
help="Output directory for KiCad library files (symbols, footprints, 3D models)."
)
parser.add_argument(
"--versions", nargs="+", default=DEFAULT_VERSIONS,
help=f"Version suffixes to process (default: {' '.join(DEFAULT_VERSIONS)})."
)
return parser
def main():
parser = build_parser()
args = parser.parse_args()
convert_to_kicad(
bom_dir=Path(args.bom_dir),
kicad_lib_dir=Path(args.kicad_lib_dir),
versions=args.versions,
)
if __name__ == "__main__":
main()

View file

@ -5,17 +5,23 @@ Parses the project.json inside each .epro archive (ZIP) to extract
devices (components) with their LCSC part numbers, manufacturer info,
footprints, values, and categories. Outputs per-version BOMs and a
cross-version evolution analysis.
Usage:
python extract_bom.py --extract-root /tmp/epro_extract --output-dir ./bom_output
python extract_bom.py --epro-dir hardware/v2.8/easyeda --output-dir ./bom_output
python extract_bom.py --help
"""
import argparse
import json
import os
import sys
import tempfile
import zipfile
from collections import defaultdict
from pathlib import Path
EXTRACT_ROOT = Path("/tmp/epro_extract")
OUTPUT_DIR = Path("/tmp/epro_bom_output")
DEFAULT_VERSIONS = ["2_3", "2_4", "2_5", "2_6", "2_7", "2_8"]
VERSIONS = ["2_3", "2_4", "2_5", "2_6", "2_7", "2_8"]
DOC_TYPES = {
"SCH ESP32": "esp32_schematic",
"SCH Power": "power_schematic",
@ -79,30 +85,54 @@ def parse_project_json(proj_path: Path) -> dict:
return components
def identify_doc_type(dir_name: str) -> tuple:
def identify_doc_type(dir_name: str, versions: list) -> tuple:
"""Identify version and document type from directory name."""
for pattern, doc_type in DOC_TYPES.items():
if pattern in dir_name:
# Extract version
for v in VERSIONS:
for v in versions:
if f"v{v}" in dir_name or f"v{v.replace('_', '.')}" in dir_name:
return v, doc_type
return None, None
def main():
OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
def extract_epro_files(epro_dir: Path, target_dir: Path):
"""Extract all .epro ZIP archives from a directory."""
target_dir.mkdir(parents=True, exist_ok=True)
for epro_file in sorted(epro_dir.glob("*.epro")):
extract_path = target_dir / epro_file.stem
if extract_path.exists():
continue
try:
with zipfile.ZipFile(epro_file, 'r') as zf:
zf.extractall(extract_path)
except zipfile.BadZipFile:
print(f"Warning: {epro_file.name} is not a valid ZIP archive, skipping.")
def run_extraction(extract_root: Path, output_dir: Path, versions: list):
"""Core extraction logic. Returns the knowledge dict.
Args:
extract_root: Root directory containing extracted .epro project folders.
output_dir: Directory for output BOM files.
versions: List of version suffixes to process (e.g. ["2_3", "2_8"]).
Returns:
dict: Knowledge base with component data per version and evolution analysis.
"""
output_dir.mkdir(parents=True, exist_ok=True)
# Collect components per version (merging ESP32 + Power schematics + PCB)
version_components = defaultdict(dict) # version -> {dev_id: component}
version_sources = defaultdict(list) # version -> [source_files]
for dirname in sorted(os.listdir(EXTRACT_ROOT)):
proj_json = EXTRACT_ROOT / dirname / "project.json"
for dirname in sorted(os.listdir(extract_root)):
proj_json = extract_root / dirname / "project.json"
if not proj_json.exists():
continue
version, doc_type = identify_doc_type(dirname)
version, doc_type = identify_doc_type(dirname, versions)
if not version:
continue
@ -112,7 +142,7 @@ def main():
# Generate per-version BOM files
all_versions_data = {}
for version in VERSIONS:
for version in versions:
comps = version_components.get(version, {})
if not comps:
continue
@ -126,7 +156,7 @@ def main():
all_versions_data[version] = sorted_comps
# Write per-version JSON
out_file = OUTPUT_DIR / f"bom_v{version}.json"
out_file = output_dir / f"bom_v{version}.json"
with open(out_file, "w") as f:
json.dump(sorted_comps, f, indent=2, ensure_ascii=False)
@ -148,7 +178,7 @@ def main():
mpn_versions = defaultdict(set) # mpn -> set of versions
mpn_info = {} # mpn -> latest component info
for version in VERSIONS:
for version in versions:
for comp in all_versions_data.get(version, []):
mpn = comp["mpn"] or comp["title"]
if mpn:
@ -156,9 +186,9 @@ def main():
mpn_info[mpn] = comp
# Components added/removed between versions
for i in range(1, len(VERSIONS)):
prev_v = VERSIONS[i - 1]
curr_v = VERSIONS[i]
for i in range(1, len(versions)):
prev_v = versions[i - 1]
curr_v = versions[i]
prev_mpns = {(c["mpn"] or c["title"]) for c in all_versions_data.get(prev_v, [])}
curr_mpns = {(c["mpn"] or c["title"]) for c in all_versions_data.get(curr_v, [])}
@ -175,8 +205,9 @@ def main():
print(f" - {mpn} ({info.get('lcsc_part', '')})")
# Components present in ALL versions (stable BOM)
stable = {mpn for mpn, vs in mpn_versions.items() if len(vs) == len(VERSIONS)}
print(f"\n\n=== STABLE COMPONENTS (all {len(VERSIONS)} versions) ===")
available_versions = [v for v in versions if v in all_versions_data]
stable = {mpn for mpn, vs in mpn_versions.items() if len(vs) == len(available_versions)} if available_versions else set()
print(f"\n\n=== STABLE COMPONENTS (all {len(available_versions)} versions) ===")
print(f"Total: {len(stable)} components")
for mpn in sorted(stable):
info = mpn_info[mpn]
@ -190,16 +221,16 @@ def main():
"stable_components": list(stable),
}
for version in VERSIONS:
for version in versions:
knowledge["versions"][f"v{version.replace('_', '.')}"] = {
"component_count": len(all_versions_data.get(version, [])),
"components": all_versions_data.get(version, []),
}
# Evolution entries
for i in range(1, len(VERSIONS)):
prev_v = VERSIONS[i - 1]
curr_v = VERSIONS[i]
for i in range(1, len(versions)):
prev_v = versions[i - 1]
curr_v = versions[i]
prev_mpns = {(c["mpn"] or c["title"]) for c in all_versions_data.get(prev_v, [])}
curr_mpns = {(c["mpn"] or c["title"]) for c in all_versions_data.get(curr_v, [])}
knowledge["evolution"].append({
@ -209,12 +240,70 @@ def main():
"removed": sorted(prev_mpns - curr_mpns),
})
out_kb = OUTPUT_DIR / "savearth_hw_knowledge.json"
out_kb = output_dir / "savearth_hw_knowledge.json"
with open(out_kb, "w") as f:
json.dump(knowledge, f, indent=2, ensure_ascii=False)
print(f"\n\nKnowledge base written to: {out_kb}")
print(f"Per-version BOMs written to: {OUTPUT_DIR}/bom_v*.json")
print(f"Per-version BOMs written to: {output_dir}/bom_v*.json")
return knowledge
def build_parser():
"""Build and return the argument parser."""
parser = argparse.ArgumentParser(
description="Extract BOM & component data from EasyEDA Pro .epro exports.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
Examples:
%(prog)s --extract-root /tmp/epro_extract --output-dir ./bom_output
%(prog)s --epro-dir hardware/v2.8/easyeda --output-dir ./bom_output
%(prog)s --extract-root /tmp/epro_extract --output-dir ./out --versions 2_7 2_8
"""
)
group = parser.add_mutually_exclusive_group(required=True)
group.add_argument(
"--extract-root",
help="Root directory of pre-extracted .epro projects (each subfolder has project.json)."
)
group.add_argument(
"--epro-dir",
help="Directory containing .epro ZIP archives to extract and parse."
)
parser.add_argument(
"--output-dir", required=True,
help="Output directory for generated BOM JSON files and knowledge base."
)
parser.add_argument(
"--versions", nargs="+", default=DEFAULT_VERSIONS,
help=f"Version suffixes to process (default: {' '.join(DEFAULT_VERSIONS)})."
)
return parser
def main():
parser = build_parser()
args = parser.parse_args()
output_dir = Path(args.output_dir)
if args.epro_dir:
epro_dir = Path(args.epro_dir)
if not epro_dir.exists():
print(f"Error: --epro-dir '{epro_dir}' does not exist.")
sys.exit(1)
# Extract .epro archives to a temporary directory
extract_root = Path(tempfile.mkdtemp(prefix="epro_extract_"))
print(f"Extracting .epro archives from {epro_dir} to {extract_root}")
extract_epro_files(epro_dir, extract_root)
else:
extract_root = Path(args.extract_root)
if not extract_root.exists():
print(f"Error: --extract-root '{extract_root}' does not exist.")
sys.exit(1)
run_extraction(extract_root, output_dir, args.versions)
if __name__ == "__main__":

View file

@ -1,4 +1,15 @@
#!/usr/bin/env python3
"""Extract schematic data from EasyEDA Pro project databases (.eprj).
Parses the SQLite-backed .eprj file, decompresses schematic documents,
and outputs raw JSON schematics, a basic netlist graph, and BOM data
for each hardware version found in the project.
Usage:
python extract_easyeda_db.py --db-path /path/to/project.eprj --output-dir /path/to/output
python extract_easyeda_db.py --help
"""
import argparse
import sqlite3
import base64
import zlib
@ -6,13 +17,14 @@ import json
import os
import sys
def decompress_dataStr(b64_str):
if not b64_str.startswith("base64"):
return b64_str
b64_data = b64_str[6:]
compressed = base64.b64decode(b64_data)
decompressed = None
for wbits in [31, 15, -15]:
try:
@ -20,12 +32,13 @@ def decompress_dataStr(b64_str):
break
except zlib.error:
continue
if decompressed is None:
raise ValueError("Could not decompress data")
return decompressed.decode('utf-8', errors='ignore')
def parse_json_lines(decompressed_str):
lines = decompressed_str.strip().split('\n')
parsed = []
@ -37,7 +50,18 @@ def parse_json_lines(decompressed_str):
pass
return parsed
def extract_db(db_path, output_dir_base):
def extract_db(db_path, output_dir_base, verbose=False):
"""Extract schematic data from an EasyEDA Pro project database.
Args:
db_path: Path to the .eprj SQLite database.
output_dir_base: Base directory for output files.
verbose: If True, print extra details during extraction.
Returns:
dict: Mapping of version names to their output directories.
"""
conn = sqlite3.connect(db_path)
c = conn.cursor()
@ -46,10 +70,10 @@ def extract_db(db_path, output_dir_base):
row = c.fetchone()
if not row:
print("No project found.")
return
return {}
boards = json.loads(row[0])
# 2. Get components from components table
c.execute("SELECT uuid, title, dataStr FROM components")
components_db = {}
@ -62,39 +86,40 @@ def extract_db(db_path, output_dir_base):
"data": comp_data
}
except Exception as e:
print(f"Failed to parse component {cuuid}: {e}")
if verbose:
print(f"Failed to parse component {cuuid}: {e}")
results = {}
for board in boards:
version_name = board.get("name")
sch_uuid = board.get("sch")
if not version_name or not sch_uuid:
continue
print(f"Processing version: {version_name}")
# Determine the target directory format (e.g., Savearth_v2_8 -> v2.8)
# For simplicity, we just use the name directly if it doesn't match a strict format,
# but let's try to parse "v2_8" to "v2.8"
v_folder = version_name.lower().replace("savearth_", "").replace("savearth ", "").replace("_", ".")
if not v_folder.startswith("v"):
v_folder = "v" + v_folder
out_dir = os.path.join(output_dir_base, v_folder, "parsed_data")
os.makedirs(out_dir, exist_ok=True)
results[version_name] = out_dir
# 3. Get all schematic documents for this sch_uuid
c.execute("SELECT title, docType, dataStr FROM documents WHERE schematic_uuid=?", (sch_uuid,))
docs = c.fetchall()
schematic_data = {}
netlist_graph = {"nets": {}, "components": []}
for doc_title, doc_type, dataStr in docs:
try:
decomp = decompress_dataStr(dataStr)
parsed = parse_json_lines(decomp)
schematic_data[doc_title] = parsed
# Very basic Netlist graph extraction
for item in parsed:
if not isinstance(item, list) or len(item) == 0:
@ -106,18 +131,19 @@ def extract_db(db_path, output_dir_base):
# item[1] is usually the UUID, subsequent elements are coords/nets
net_id = item[1]
netlist_graph["nets"][net_id] = item
except Exception as e:
print(f"Failed to parse document {doc_title}: {e}")
if verbose:
print(f"Failed to parse document {doc_title}: {e}")
# 4. Write Schematic Raw JSON
with open(os.path.join(out_dir, "schematic_raw.json"), "w") as f:
json.dump(schematic_data, f, indent=2)
# 5. Write basic Netlist JSON
with open(os.path.join(out_dir, "netlist.json"), "w") as f:
json.dump(netlist_graph, f, indent=2)
# 6. Build BOM from the components used in the schematic
bom_extended = []
for doc_title, parsed in schematic_data.items():
@ -136,11 +162,42 @@ def extract_db(db_path, output_dir_base):
with open(os.path.join(out_dir, "bom_extended.json"), "w") as f:
json.dump(bom_extended, f, indent=2)
if verbose:
print(f"{out_dir}: {len(schematic_data)} docs, {len(bom_extended)} components")
conn.close()
print("Extraction complete.")
return results
def build_parser():
"""Build and return the argument parser."""
parser = argparse.ArgumentParser(
description="Extract schematic data from EasyEDA Pro project databases (.eprj).",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
Examples:
%(prog)s --db-path hardware/v2.8/easyeda/project.eprj --output-dir hardware/
%(prog)s --db-path project.eprj --output-dir ./output --verbose
"""
)
parser.add_argument(
"--db-path", required=True,
help="Path to the EasyEDA Pro project database (.eprj SQLite file)."
)
parser.add_argument(
"--output-dir", required=True,
help="Base output directory for extracted data (version subdirs will be created)."
)
parser.add_argument(
"--verbose", action="store_true",
help="Print extra details during extraction."
)
return parser
if __name__ == "__main__":
db_path = "/home/fcunha/savearth/savearth-hw-project/hardware/v2.8/easyeda/Savearth GBT.eprj"
output_dir = "/home/fcunha/savearth/savearth-hw-project/hardware/"
extract_db(db_path, output_dir)
parser = build_parser()
args = parser.parse_args()
extract_db(args.db_path, args.output_dir, verbose=args.verbose)

View file

@ -1,16 +1,37 @@
#!/usr/bin/env python3
"""Generate JLCPCB-compatible CSV BOM from extracted JSON BOM."""
"""Generate JLCPCB-compatible CSV BOM from extracted JSON BOM.
Reads a per-version BOM JSON file (produced by extract_bom.py) and
writes a JLCPCB-compatible CSV with columns:
Comment, Designator, Footprint, LCSC Part #, Manufacturer, MPN, Value, Category
Usage:
python generate_jlcpcb_bom.py --bom-dir ./bom_output --version 2_8
python generate_jlcpcb_bom.py --help
"""
import argparse
import csv
import json
import sys
from pathlib import Path
BOM_DIR = Path("/home/fcunha/savearth/flow-meter-pcb/hardware/bom")
def generate_jlcpcb_csv(bom_dir: Path, version: str, output_file: Path = None):
"""Generate a JLCPCB-compatible CSV BOM from a JSON BOM.
def main():
version = sys.argv[1] if len(sys.argv) > 1 else "2_8"
bom_file = BOM_DIR / f"bom_v{version}.json"
Args:
bom_dir: Directory containing bom_v{version}.json files.
version: Hardware version suffix (e.g. "2_8").
output_file: Optional explicit output path. Defaults to bom_dir/bom_v{version}_jlcpcb.csv.
Returns:
Path: Path to the generated CSV file.
"""
bom_file = bom_dir / f"bom_v{version}.json"
if not bom_file.exists():
print(f"Error: BOM file not found: {bom_file}")
sys.exit(1)
with open(bom_file) as f:
components = json.load(f)
@ -18,8 +39,10 @@ def main():
# Filter to real components only (has LCSC part)
real = [c for c in components if c.get("lcsc_part")]
csv_file = BOM_DIR / f"bom_v{version}_jlcpcb.csv"
with open(csv_file, "w", newline="") as f:
if output_file is None:
output_file = bom_dir / f"bom_v{version}_jlcpcb.csv"
with open(output_file, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow(["Comment", "Designator", "Footprint", "LCSC Part #", "Manufacturer", "MPN", "Value", "Category"])
for c in real:
@ -34,7 +57,44 @@ def main():
c.get("category", ""),
])
print(f"Written {len(real)} components to {csv_file}")
print(f"Written {len(real)} components to {output_file}")
return output_file
def build_parser():
"""Build and return the argument parser."""
parser = argparse.ArgumentParser(
description="Generate JLCPCB-compatible CSV BOM from extracted JSON BOM.",
formatter_class=argparse.RawDescriptionHelpFormatter,
epilog="""\
Examples:
%(prog)s --bom-dir ./bom_output --version 2_8
%(prog)s --bom-dir ./bom_output --version 2_7 --output-file custom_bom.csv
"""
)
parser.add_argument(
"--bom-dir", required=True,
help="Directory containing the JSON BOM files (bom_v{version}.json)."
)
parser.add_argument(
"--version", default="2_8",
help="Hardware version suffix (default: 2_8). Example: 2_7, 2_8."
)
parser.add_argument(
"--output-file",
help="Optional explicit path for the output JLCPCB CSV file."
)
return parser
def main():
parser = build_parser()
args = parser.parse_args()
bom_dir = Path(args.bom_dir)
output_file = Path(args.output_file) if args.output_file else None
generate_jlcpcb_csv(bom_dir, args.version, output_file)
if __name__ == "__main__":