update
This commit is contained in:
@@ -0,0 +1 @@
|
||||
"""Application service helpers."""
|
||||
@@ -0,0 +1,80 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from decimal import Decimal, InvalidOperation, ROUND_HALF_UP
|
||||
|
||||
|
||||
FUEL_FACTORS: list[tuple[str, Decimal]] = [
|
||||
("HTPB", Decimal("0.00177")),
|
||||
("Solid", Decimal("0.00178")),
|
||||
("PBAN", Decimal("0.001772")),
|
||||
("Kerolox", Decimal("0.0010290673")),
|
||||
("Hydrolox", Decimal("0.00036235886")),
|
||||
("Methalox", Decimal("0.000845043157")),
|
||||
("Hydrazine", Decimal("0.001004")),
|
||||
("MMH/NTO", Decimal("0.00116557")),
|
||||
]
|
||||
|
||||
FUEL_DENSITY_T_PER_LITER = dict(FUEL_FACTORS)
|
||||
VALUE_STEP = Decimal("0.000001")
|
||||
|
||||
|
||||
def list_fuel_factors() -> list[dict[str, float | str]]:
|
||||
return [
|
||||
{"fuel": fuel, "tonnes_per_liter": _to_float(factor)}
|
||||
for fuel, factor in FUEL_FACTORS
|
||||
]
|
||||
|
||||
|
||||
def convert_value(mode: str, raw_value: object) -> dict[str, object]:
|
||||
value = _parse_positive_decimal(raw_value)
|
||||
normalized_mode = mode.strip().lower()
|
||||
|
||||
if normalized_mode == "volume":
|
||||
results = [
|
||||
{
|
||||
"fuel": fuel,
|
||||
"value": _to_float(value * factor),
|
||||
"unit": "t",
|
||||
}
|
||||
for fuel, factor in FUEL_FACTORS
|
||||
]
|
||||
return {
|
||||
"mode": normalized_mode,
|
||||
"input_value": _to_float(value),
|
||||
"input_unit": "L",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
if normalized_mode == "mass":
|
||||
results = [
|
||||
{
|
||||
"fuel": fuel,
|
||||
"value": _to_float(value / factor),
|
||||
"unit": "L",
|
||||
}
|
||||
for fuel, factor in FUEL_FACTORS
|
||||
]
|
||||
return {
|
||||
"mode": normalized_mode,
|
||||
"input_value": _to_float(value),
|
||||
"input_unit": "t",
|
||||
"results": results,
|
||||
}
|
||||
|
||||
raise ValueError("mode must be either 'volume' or 'mass'")
|
||||
|
||||
|
||||
def _parse_positive_decimal(raw_value: object) -> Decimal:
|
||||
try:
|
||||
value = Decimal(str(raw_value))
|
||||
except (InvalidOperation, ValueError, TypeError) as exc:
|
||||
raise ValueError("value must be numeric") from exc
|
||||
|
||||
if value <= 0:
|
||||
raise ValueError("value must be greater than zero")
|
||||
|
||||
return value
|
||||
|
||||
|
||||
def _to_float(value: Decimal) -> float:
|
||||
return float(value.quantize(VALUE_STEP, rounding=ROUND_HALF_UP))
|
||||
@@ -0,0 +1,278 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from dataclasses import dataclass, field
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from sqlalchemy import delete, select
|
||||
|
||||
from app.extensions import db
|
||||
from app.models import Asset, AssetLogEntry
|
||||
|
||||
|
||||
EVENT_LINE_PATTERN = re.compile(r"^(\d{4}-\d{2}-\d{2})\s+(.+)$")
|
||||
LOG_BOOK_IGNORED_SHEETS = {"Overview", "Model"}
|
||||
|
||||
BODY_LOCATION_KEYWORDS = [
|
||||
("earth-sun l2", "Earth-Sun L2"),
|
||||
("earth-sun lagrange point 2", "Earth-Sun L2"),
|
||||
("star port", "Star Port"),
|
||||
("iapetus", "Iapetus"),
|
||||
("callisto", "Callisto"),
|
||||
("ganymede", "Ganymede"),
|
||||
("europa", "Europa"),
|
||||
("jupiter", "Jupiter"),
|
||||
("saturn", "Saturn"),
|
||||
("neptune", "Neptune"),
|
||||
("mars", "Mars"),
|
||||
("venus", "Venus"),
|
||||
("mercury", "Mercury"),
|
||||
("lunar orbit", "Lunar Orbit"),
|
||||
("leo", "LEO"),
|
||||
("earth orbit", "Earth"),
|
||||
("earth", "Earth"),
|
||||
("sun", "Sun"),
|
||||
]
|
||||
|
||||
LOG_BOOK_ASSET_SPECS = {
|
||||
"ST-01": {
|
||||
"name": "万星源号 ST-01",
|
||||
"asset_type": "Exploration Mothership",
|
||||
"program": "Stellaria",
|
||||
"home_region": "LEO",
|
||||
"note": "First Class of Stellaria",
|
||||
},
|
||||
"XH-01": {
|
||||
"name": "羲和号 XH-01",
|
||||
"asset_type": "Exploration Mothership",
|
||||
"program": "羲和计划",
|
||||
"home_region": "LEO",
|
||||
"note": "羲和计划首舰",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class LogBookSheetSummary:
|
||||
sheet_name: str
|
||||
asset_name: str
|
||||
state_entries: int = 0
|
||||
event_entries: int = 0
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"sheet_name": self.sheet_name,
|
||||
"asset_name": self.asset_name,
|
||||
"state_entries": self.state_entries,
|
||||
"event_entries": self.event_entries,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class LogBookImportSummary:
|
||||
workbook: str
|
||||
imported_assets: list[LogBookSheetSummary] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"workbook": self.workbook,
|
||||
"imported_assets": [item.to_dict() for item in self.imported_assets],
|
||||
}
|
||||
|
||||
|
||||
def import_log_book_data(workbook_path: str | Path) -> LogBookImportSummary:
|
||||
path = Path(workbook_path)
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Workbook not found: {path}")
|
||||
|
||||
workbook = load_workbook(path, data_only=True, read_only=True)
|
||||
summary = LogBookImportSummary(workbook=path.name)
|
||||
|
||||
try:
|
||||
for sheet_name in _iter_importable_sheet_names(workbook.sheetnames):
|
||||
asset_spec = _resolve_asset_spec(sheet_name)
|
||||
sheet_summary = _import_sheet(workbook[sheet_name], sheet_name, asset_spec)
|
||||
summary.imported_assets.append(sheet_summary)
|
||||
db.session.commit()
|
||||
except Exception:
|
||||
db.session.rollback()
|
||||
raise
|
||||
|
||||
return summary
|
||||
|
||||
|
||||
def _iter_importable_sheet_names(sheet_names: list[str]) -> list[str]:
|
||||
return [sheet_name for sheet_name in sheet_names if sheet_name not in LOG_BOOK_IGNORED_SHEETS]
|
||||
|
||||
|
||||
def _resolve_asset_spec(sheet_name: str) -> dict[str, str | None]:
|
||||
asset_spec: dict[str, str | None] = {
|
||||
"name": sheet_name,
|
||||
"asset_type": "Vehicle",
|
||||
"program": None,
|
||||
"home_region": None,
|
||||
"note": f"Imported from log_book.xlsx sheet {sheet_name}.",
|
||||
}
|
||||
asset_spec.update(LOG_BOOK_ASSET_SPECS.get(sheet_name, {}))
|
||||
return asset_spec
|
||||
|
||||
|
||||
def _import_sheet(worksheet: object, sheet_name: str, asset_spec: dict[str, str]) -> LogBookSheetSummary:
|
||||
asset = db.session.execute(select(Asset).where(Asset.name == asset_spec["name"])).scalar_one_or_none()
|
||||
if asset is None:
|
||||
asset = Asset(
|
||||
name=asset_spec["name"],
|
||||
asset_type=asset_spec["asset_type"],
|
||||
program=asset_spec["program"],
|
||||
home_region=asset_spec["home_region"],
|
||||
note=asset_spec["note"],
|
||||
)
|
||||
db.session.add(asset)
|
||||
db.session.flush()
|
||||
|
||||
asset.asset_type = asset_spec["asset_type"]
|
||||
asset.program = asset_spec["program"]
|
||||
asset.home_region = asset_spec["home_region"]
|
||||
asset.note = asset_spec["note"]
|
||||
|
||||
db.session.execute(delete(AssetLogEntry).where(AssetLogEntry.asset_id == asset.id))
|
||||
db.session.flush()
|
||||
|
||||
summary = LogBookSheetSummary(sheet_name=sheet_name, asset_name=asset.name)
|
||||
for row in worksheet.iter_rows(min_row=2, values_only=True):
|
||||
if not any(value not in (None, "") for value in row):
|
||||
continue
|
||||
|
||||
log_name = _clean_text(row[0])
|
||||
start_at = _coerce_datetime(row[1])
|
||||
end_at = _coerce_datetime(row[2])
|
||||
mission_detail = _clean_text(row[3])
|
||||
sub_log = _clean_text(row[4])
|
||||
explicit_location = _clean_text(row[7])
|
||||
inferred_location = _infer_location(log_name, mission_detail, sub_log, explicit_location)
|
||||
|
||||
if log_name is None or start_at is None:
|
||||
continue
|
||||
|
||||
db.session.add(
|
||||
AssetLogEntry(
|
||||
asset_id=asset.id,
|
||||
entry_kind="state",
|
||||
title=log_name,
|
||||
state_label=log_name,
|
||||
mission_label=log_name,
|
||||
location=inferred_location,
|
||||
start_at=start_at,
|
||||
end_at=end_at,
|
||||
summary=mission_detail,
|
||||
note=sub_log,
|
||||
)
|
||||
)
|
||||
summary.state_entries += 1
|
||||
|
||||
return summary
|
||||
|
||||
|
||||
def _clean_text(value: object) -> str | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
cleaned = str(value).strip()
|
||||
return cleaned or None
|
||||
|
||||
|
||||
def _coerce_datetime(value: object) -> datetime | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
if isinstance(value, datetime):
|
||||
normalized = value
|
||||
else:
|
||||
normalized = datetime.fromisoformat(str(value))
|
||||
if normalized.tzinfo is None:
|
||||
return normalized.replace(tzinfo=timezone.utc)
|
||||
return normalized.astimezone(timezone.utc)
|
||||
|
||||
|
||||
def _infer_location(
|
||||
log_name: str | None,
|
||||
mission_detail: str | None,
|
||||
sub_log: str | None,
|
||||
explicit_location: str | None,
|
||||
) -> str | None:
|
||||
if explicit_location:
|
||||
return explicit_location
|
||||
|
||||
haystacks = [
|
||||
(log_name or "").casefold(),
|
||||
(mission_detail or "").casefold(),
|
||||
(sub_log or "").casefold(),
|
||||
]
|
||||
for keyword, location in BODY_LOCATION_KEYWORDS:
|
||||
if any(keyword in haystack for haystack in haystacks):
|
||||
return location
|
||||
|
||||
if mission_detail and "transfer" in mission_detail.casefold():
|
||||
return "Sun"
|
||||
if log_name and "maintenance" in log_name.casefold():
|
||||
return "LEO"
|
||||
if log_name and "construction" in log_name.casefold():
|
||||
return "LEO"
|
||||
return None
|
||||
|
||||
|
||||
def _infer_event_location(description: str, default_location: str | None) -> str | None:
|
||||
lowered = description.casefold()
|
||||
for keyword, location in BODY_LOCATION_KEYWORDS:
|
||||
if keyword in lowered:
|
||||
return location
|
||||
return default_location
|
||||
|
||||
|
||||
def _parse_sub_log_events(
|
||||
asset_id: object,
|
||||
parent_title: str,
|
||||
parent_start_at: datetime,
|
||||
parent_end_at: datetime | None,
|
||||
default_location: str | None,
|
||||
sub_log: str | None,
|
||||
) -> list[AssetLogEntry]:
|
||||
if sub_log is None:
|
||||
return []
|
||||
|
||||
events = []
|
||||
for raw_line in sub_log.splitlines():
|
||||
line = raw_line.strip()
|
||||
if not line:
|
||||
continue
|
||||
match = EVENT_LINE_PATTERN.match(line)
|
||||
if match is None:
|
||||
continue
|
||||
|
||||
event_at = _coerce_datetime(match.group(1))
|
||||
if event_at is None:
|
||||
continue
|
||||
|
||||
# Skip obvious workbook typos that fall far outside the parent interval.
|
||||
if event_at < parent_start_at.replace(year=parent_start_at.year - 1):
|
||||
continue
|
||||
if parent_end_at is not None and event_at > parent_end_at.replace(year=parent_end_at.year + 1):
|
||||
continue
|
||||
|
||||
description = match.group(2).strip()
|
||||
events.append(
|
||||
AssetLogEntry(
|
||||
asset_id=asset_id,
|
||||
entry_kind="event",
|
||||
title=description,
|
||||
state_label=None,
|
||||
mission_label=parent_title,
|
||||
location=_infer_event_location(description, default_location),
|
||||
start_at=event_at,
|
||||
end_at=None,
|
||||
summary=parent_title,
|
||||
note=line,
|
||||
)
|
||||
)
|
||||
|
||||
return events
|
||||
@@ -0,0 +1,316 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from decimal import Decimal, InvalidOperation
|
||||
from pathlib import Path
|
||||
|
||||
from openpyxl import load_workbook
|
||||
from sqlalchemy import delete, inspect
|
||||
|
||||
from app.extensions import db
|
||||
from app.models import CommunicationPart, EngineFamily, EngineVariant, TankSpec, VehicleCost
|
||||
|
||||
|
||||
@dataclass
|
||||
class WorkbookImportSummary:
|
||||
workbook: str
|
||||
replaced_existing: bool
|
||||
imported_rows: dict[str, int] = field(default_factory=dict)
|
||||
warnings: list[str] = field(default_factory=list)
|
||||
|
||||
def to_dict(self) -> dict[str, object]:
|
||||
return {
|
||||
"workbook": self.workbook,
|
||||
"replaced_existing": self.replaced_existing,
|
||||
"imported_rows": self.imported_rows,
|
||||
"warnings": self.warnings,
|
||||
}
|
||||
|
||||
|
||||
def import_workbook_data(
|
||||
workbook_path: str | Path, replace_existing: bool = False
|
||||
) -> WorkbookImportSummary:
|
||||
path = Path(workbook_path)
|
||||
if not path.exists():
|
||||
raise FileNotFoundError(f"Workbook not found: {path}")
|
||||
|
||||
_assert_schema_ready()
|
||||
|
||||
summary = WorkbookImportSummary(workbook=path.name, replaced_existing=replace_existing)
|
||||
workbook = load_workbook(path, data_only=True, read_only=True)
|
||||
|
||||
try:
|
||||
if replace_existing:
|
||||
_clear_import_tables()
|
||||
elif _database_has_data():
|
||||
raise ValueError("数据库已有数据。若要重导，请使用 --replace。")
|
||||
|
||||
family_count, variant_count = _import_engines(workbook["Engine Database"], summary)
|
||||
communication_count = _import_communication_parts(workbook["Communication"])
|
||||
tank_count = _import_tank_specs(workbook["Tank Chart"])
|
||||
vehicle_cost_count = _import_vehicle_costs(workbook["KSP Vehicle Cost"])
|
||||
|
||||
db.session.commit()
|
||||
except Exception:
|
||||
db.session.rollback()
|
||||
raise
|
||||
|
||||
summary.imported_rows = {
|
||||
"engine_families": family_count,
|
||||
"engine_variants": variant_count,
|
||||
"communication_parts": communication_count,
|
||||
"tank_specs": tank_count,
|
||||
"vehicle_costs": vehicle_cost_count,
|
||||
}
|
||||
return summary
|
||||
|
||||
|
||||
def _assert_schema_ready() -> None:
|
||||
existing_tables = set(inspect(db.engine).get_table_names())
|
||||
required_tables = {
|
||||
"engine_families",
|
||||
"engine_variants",
|
||||
"communication_parts",
|
||||
"tank_specs",
|
||||
"vehicle_costs",
|
||||
}
|
||||
missing_tables = sorted(required_tables - existing_tables)
|
||||
if missing_tables:
|
||||
missing = ", ".join(missing_tables)
|
||||
raise ValueError(f"数据库缺少表：{missing}。请先执行数据库迁移。")
|
||||
|
||||
|
||||
def _database_has_data() -> bool:
|
||||
for model in (EngineFamily, EngineVariant, CommunicationPart, TankSpec, VehicleCost):
|
||||
if db.session.query(model).first() is not None:
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
def _clear_import_tables() -> None:
|
||||
for model in (EngineVariant, EngineFamily, CommunicationPart, TankSpec, VehicleCost):
|
||||
db.session.execute(delete(model))
|
||||
db.session.flush()
|
||||
|
||||
|
||||
def _import_engines(worksheet: object, summary: WorkbookImportSummary) -> tuple[int, int]:
|
||||
families_by_name: dict[str, EngineFamily] = {}
|
||||
variant_count = 0
|
||||
|
||||
for row_number, row in enumerate(worksheet.iter_rows(min_row=2, values_only=True), start=2):
|
||||
if not _row_has_values(row):
|
||||
continue
|
||||
|
||||
engine_name = _text(row[0])
|
||||
if engine_name is None:
|
||||
summary.warnings.append(f"Engine Database 第 {row_number} 行缺少 Engine，已跳过。")
|
||||
continue
|
||||
|
||||
family = families_by_name.get(engine_name)
|
||||
if family is None:
|
||||
family = EngineFamily(
|
||||
engine_name=engine_name,
|
||||
part_name=_text(row[18]),
|
||||
cycle=_text(row[2]),
|
||||
size_m=_decimal(row[12]),
|
||||
entry_cost=_integer(row[14]),
|
||||
)
|
||||
families_by_name[engine_name] = family
|
||||
db.session.add(family)
|
||||
else:
|
||||
family.part_name = family.part_name or _text(row[18])
|
||||
family.cycle = family.cycle or _text(row[2])
|
||||
family.size_m = family.size_m or _decimal(row[12])
|
||||
family.entry_cost = family.entry_cost or _integer(row[14])
|
||||
|
||||
fuel_type = _text(row[1])
|
||||
if fuel_type is None:
|
||||
summary.warnings.append(f"Engine Database 第 {row_number} 行缺少 Fuel Type，已跳过。")
|
||||
continue
|
||||
|
||||
min_thrust = _decimal(row[6])
|
||||
max_thrust = _decimal(row[7])
|
||||
mass = _decimal(row[8])
|
||||
|
||||
variant = EngineVariant(
|
||||
family=family,
|
||||
fuel_type=fuel_type,
|
||||
work_env=_text(row[3]),
|
||||
sl_isp=_decimal(row[4]),
|
||||
vac_isp=_decimal(row[5]),
|
||||
min_thrust_kn=min_thrust,
|
||||
max_thrust_kn=max_thrust,
|
||||
mass_t=mass,
|
||||
twr=_decimal(row[9]) or _calculate_twr(max_thrust, mass),
|
||||
throttle_ratio=_decimal(row[10]) or _calculate_throttle_ratio(min_thrust, max_thrust),
|
||||
tvc_deg=_decimal(row[11]),
|
||||
price_kd=_integer(row[13]),
|
||||
ignitions=_integer(row[15]),
|
||||
has_unlimited_ignitions=_integer(row[15]) is None,
|
||||
mod_source=_text(row[16]),
|
||||
engine_note=_text(row[17]),
|
||||
config_name=_text(row[19]),
|
||||
tech_required=_text(row[20]),
|
||||
config_note=_text(row[21]),
|
||||
source_sheet_row=row_number,
|
||||
)
|
||||
db.session.add(variant)
|
||||
variant_count += 1
|
||||
|
||||
return len(families_by_name), variant_count
|
||||
|
||||
|
||||
def _import_communication_parts(worksheet: object) -> int:
|
||||
count = 0
|
||||
for row in worksheet.iter_rows(min_row=2, values_only=True):
|
||||
if not _row_has_values(row):
|
||||
continue
|
||||
|
||||
part_name = _text(row[0])
|
||||
if part_name is None:
|
||||
continue
|
||||
|
||||
db.session.add(
|
||||
CommunicationPart(
|
||||
part_name=part_name,
|
||||
display_name=_text(row[1]),
|
||||
mass_t=_decimal(row[2]),
|
||||
is_active=_boolean(row[3]),
|
||||
is_deployable=_boolean(row[4]),
|
||||
antenna_type=_text(row[5]),
|
||||
deployed_diameter_m=_decimal(row[6]),
|
||||
range_raw=_decimal(row[7]),
|
||||
range_km=_decimal(row[8]),
|
||||
range_au=_decimal(row[9]),
|
||||
range_light_year=_decimal(row[10]),
|
||||
angle_deg=_decimal(row[11]),
|
||||
speed=_decimal(row[12]),
|
||||
idle_power_watt=_decimal(row[13]),
|
||||
idle_power_text=_text(row[14]),
|
||||
transmitting_power_watt=_decimal(row[15]),
|
||||
transmitting_power_text=_text(row[16]),
|
||||
source=_text(row[17]),
|
||||
entry_cost=_integer(row[18]),
|
||||
cost=_integer(row[19]),
|
||||
description=_text(row[20]),
|
||||
rescale_factor=_decimal(row[21]),
|
||||
tweakscale=_text(row[22]),
|
||||
is_feeder=_boolean(row[23]),
|
||||
tech_required=_text(row[24]),
|
||||
note=_text(row[25]),
|
||||
)
|
||||
)
|
||||
count += 1
|
||||
|
||||
return count
|
||||
|
||||
|
||||
def _import_tank_specs(worksheet: object) -> int:
|
||||
count = 0
|
||||
for row in worksheet.iter_rows(min_row=2, values_only=True):
|
||||
if not _row_has_values(row):
|
||||
continue
|
||||
|
||||
tank_name = _text(row[0])
|
||||
if tank_name is None:
|
||||
continue
|
||||
|
||||
db.session.add(
|
||||
TankSpec(
|
||||
tank_name=tank_name,
|
||||
fuel_type=_text(row[1]),
|
||||
dry_mass_t=_decimal(row[2]),
|
||||
fuel_mass_t=_decimal(row[3]),
|
||||
wet_mass_t=_decimal(row[4]),
|
||||
tank_volume_l=_decimal(row[5]),
|
||||
mass_ratio=_decimal(row[6]),
|
||||
kiloliters_per_ton=_decimal(row[7]),
|
||||
vehicle_name=_text(row[8]),
|
||||
source=_text(row[9]),
|
||||
note=_text(row[10]),
|
||||
)
|
||||
)
|
||||
count += 1
|
||||
|
||||
return count
|
||||
|
||||
|
||||
def _import_vehicle_costs(worksheet: object) -> int:
|
||||
count = 0
|
||||
for row in worksheet.iter_rows(min_row=2, values_only=True):
|
||||
if not _row_has_values(row):
|
||||
continue
|
||||
|
||||
vehicle_name = _text(row[0])
|
||||
if vehicle_name is None:
|
||||
continue
|
||||
|
||||
db.session.add(
|
||||
VehicleCost(
|
||||
vehicle_name=vehicle_name,
|
||||
launch_price=_integer(row[1]),
|
||||
source=_text(row[2]),
|
||||
)
|
||||
)
|
||||
count += 1
|
||||
|
||||
return count
|
||||
|
||||
|
||||
def _row_has_values(row: tuple[object, ...]) -> bool:
|
||||
return any(value not in (None, "") for value in row)
|
||||
|
||||
|
||||
def _text(value: object) -> str | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
cleaned = str(value).strip()
|
||||
return cleaned or None
|
||||
|
||||
|
||||
def _decimal(value: object) -> Decimal | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
if isinstance(value, Decimal):
|
||||
return value
|
||||
try:
|
||||
return Decimal(str(value))
|
||||
except (InvalidOperation, ValueError, TypeError):
|
||||
return None
|
||||
|
||||
|
||||
def _integer(value: object) -> int | None:
|
||||
decimal_value = _decimal(value)
|
||||
if decimal_value is None:
|
||||
return None
|
||||
return int(decimal_value.to_integral_value())
|
||||
|
||||
|
||||
def _boolean(value: object) -> bool | None:
|
||||
if value in (None, ""):
|
||||
return None
|
||||
if isinstance(value, bool):
|
||||
return value
|
||||
if isinstance(value, (int, float)):
|
||||
return bool(value)
|
||||
|
||||
text = str(value).strip().lower()
|
||||
if text in {"true", "1", "yes", "y"}:
|
||||
return True
|
||||
if text in {"false", "0", "no", "n"}:
|
||||
return False
|
||||
return None
|
||||
|
||||
|
||||
def _calculate_twr(max_thrust: Decimal | None, mass: Decimal | None) -> Decimal | None:
|
||||
if max_thrust is None or mass in (None, Decimal("0")):
|
||||
return None
|
||||
return max_thrust / Decimal("9.80665") / mass
|
||||
|
||||
|
||||
def _calculate_throttle_ratio(
|
||||
min_thrust: Decimal | None, max_thrust: Decimal | None
|
||||
) -> Decimal | None:
|
||||
if min_thrust is None or max_thrust in (None, Decimal("0")):
|
||||
return None
|
||||
return min_thrust / max_thrust
|
||||
@@ -0,0 +1,138 @@
|
||||
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, "")
|
||||
Reference in New Issue
Block a user