139 lines
3.9 KiB
Python
139 lines
3.9 KiB
Python
from __future__ import annotations
|
|
|
|
from collections import Counter, defaultdict
|
|
from pathlib import Path
|
|
|
|
from openpyxl import load_workbook
|
|
|
|
|
|
ENGINE_FIELD_INDICES = {
|
|
"fuel_type": 1,
|
|
"cycle": 2,
|
|
"work_env": 3,
|
|
"tvc": 11,
|
|
"size": 12,
|
|
"price_kd": 13,
|
|
"entry_cost": 14,
|
|
"classification": 16,
|
|
"note": 17,
|
|
"part_name": 18,
|
|
"config_name": 19,
|
|
}
|
|
|
|
|
|
def summarize_workbook(workbook_path: str | Path) -> dict[str, object]:
|
|
path = Path(workbook_path)
|
|
workbook = load_workbook(path, data_only=False)
|
|
|
|
sheet_row_counts = {
|
|
sheet_name: len(_non_empty_rows(workbook[sheet_name]))
|
|
for sheet_name in workbook.sheetnames
|
|
}
|
|
|
|
engine_sheet = workbook["Engine Database"]
|
|
engine_rows = _non_empty_rows(engine_sheet)
|
|
engine_headers = [
|
|
cell.value
|
|
for cell in next(engine_sheet.iter_rows(min_row=1, max_row=1))
|
|
if cell.value is not None
|
|
]
|
|
|
|
duplicate_engine_config = _sample_duplicates(
|
|
Counter((row[0], row[19]) for row in engine_rows if _present(row[0]))
|
|
)
|
|
duplicate_engine_config_with_fuel = _sample_duplicates(
|
|
Counter((row[0], row[19], row[1]) for row in engine_rows if _present(row[0]))
|
|
)
|
|
|
|
work_env_values = sorted(
|
|
{row[3] for row in engine_rows if _present(row[3])}, key=str
|
|
)
|
|
|
|
return {
|
|
"workbook": path.name,
|
|
"sheet_names": workbook.sheetnames,
|
|
"sheet_row_counts": sheet_row_counts,
|
|
"engine_headers": engine_headers,
|
|
"engine_work_env_values": work_env_values,
|
|
"candidate_key_conflicts": {
|
|
"engine_plus_config_name": {
|
|
"count": len(duplicate_engine_config),
|
|
"samples": duplicate_engine_config[:12],
|
|
},
|
|
"engine_plus_config_name_plus_fuel_type": {
|
|
"count": len(duplicate_engine_config_with_fuel),
|
|
"samples": duplicate_engine_config_with_fuel[:12],
|
|
},
|
|
},
|
|
"engine_field_variance": _field_variance(engine_rows),
|
|
"recommended_family_fields": [
|
|
"engine_name",
|
|
"part_name",
|
|
"cycle",
|
|
"size_m",
|
|
"entry_cost",
|
|
],
|
|
"recommended_variant_fields": [
|
|
"fuel_type",
|
|
"work_env",
|
|
"sl_isp",
|
|
"vac_isp",
|
|
"min_thrust_kn",
|
|
"max_thrust_kn",
|
|
"mass_t",
|
|
"twr",
|
|
"throttle_ratio",
|
|
"tvc_deg",
|
|
"price_kd",
|
|
"ignitions",
|
|
"has_unlimited_ignitions",
|
|
"mod_source",
|
|
"engine_note",
|
|
"config_name",
|
|
"tech_required",
|
|
"config_note",
|
|
],
|
|
}
|
|
|
|
|
|
def _field_variance(engine_rows: list[tuple[object, ...]]) -> dict[str, int]:
|
|
grouped_rows: dict[object, list[tuple[object, ...]]] = defaultdict(list)
|
|
for row in engine_rows:
|
|
grouped_rows[row[0]].append(row)
|
|
|
|
variance: dict[str, int] = {}
|
|
for field_name, index in ENGINE_FIELD_INDICES.items():
|
|
changing = 0
|
|
for rows in grouped_rows.values():
|
|
values = {row[index] for row in rows if _present(row[index])}
|
|
if len(values) > 1:
|
|
changing += 1
|
|
variance[field_name] = changing
|
|
return variance
|
|
|
|
|
|
def _sample_duplicates(counter: Counter[tuple[object, ...]]) -> list[dict[str, object]]:
|
|
samples: list[dict[str, object]] = []
|
|
for key, count in counter.items():
|
|
if count <= 1:
|
|
continue
|
|
|
|
record = {"count": count}
|
|
for index, value in enumerate(key):
|
|
record[f"key_{index + 1}"] = value
|
|
samples.append(record)
|
|
|
|
return samples
|
|
|
|
|
|
def _non_empty_rows(worksheet: object) -> list[tuple[object, ...]]:
|
|
return [
|
|
row
|
|
for row in worksheet.iter_rows(min_row=2, values_only=True)
|
|
if any(_present(value) for value in row)
|
|
]
|
|
|
|
|
|
def _present(value: object) -> bool:
|
|
return value not in (None, "")
|