diff --git a/game/theater/theaterloader.py b/game/theater/theaterloader.py index c86d48ec..c5584aa4 100644 --- a/game/theater/theaterloader.py +++ b/game/theater/theaterloader.py @@ -1,6 +1,7 @@ from __future__ import annotations import datetime +from collections.abc import Iterator from dataclasses import dataclass from pathlib import Path from typing import Any @@ -57,9 +58,16 @@ class SeasonData: class TheaterLoader: + THEATER_RESOURCE_DIR = Path("resources/theaters") + def __init__(self, name: str) -> None: self.name = name - self.descriptor_path = Path("resources/theaters") / self.name / "info.yaml" + self.descriptor_path = self.THEATER_RESOURCE_DIR / self.name / "info.yaml" + + @classmethod + def each(cls) -> Iterator[ConflictTheater]: + for theater_dir in cls.THEATER_RESOURCE_DIR.iterdir(): + yield TheaterLoader(theater_dir.name).load() def load(self) -> ConflictTheater: with self.descriptor_path.open() as descriptor_file: diff --git a/resources/tools/datalint.py b/resources/tools/datalint.py index e69de29b..fb8c4c92 100644 --- a/resources/tools/datalint.py +++ b/resources/tools/datalint.py @@ -0,0 +1,219 @@ +from __future__ import annotations + +import argparse +import contextlib +import logging +import subprocess +import sys +import textwrap +from abc import ABC, abstractmethod +from collections.abc import Iterable, Iterator +from datetime import date +from pathlib import Path +from typing import TextIO, TypeAlias + +from game import VERSION +from game.airfields import AirfieldData +from game.theater import ConflictTheater +from game.theater.theaterloader import TheaterLoader + + +class ReportElement(ABC): + @abstractmethod + def __str__(self) -> str: + ... + + +class Heading(ReportElement): + def __init__(self, level: int, text: str) -> None: + self.level = level + self.text = text + + def __str__(self) -> str: + return f"{'#' * self.level} {self.text}" + + +class H1(Heading): + def __init__(self, text: str) -> None: + super().__init__(1, text) + + +class H2(Heading): + def __init__(self, text: str) -> None: + super().__init__(2, text) + + +class H3(Heading): + def __init__(self, text: str) -> None: + super().__init__(3, text) + + +class Paragraph(ReportElement): + def __init__(self, text: str) -> None: + self.text = text + + def __str__(self) -> str: + return textwrap.fill(self.text, width=80) + + +class UnorderedList(ReportElement): + def __init__(self, items: Iterable[str]) -> None: + self.items = list(items) + if not self.items: + raise ValueError("List has no data") + + def __str__(self) -> str: + return "\n".join(f"* {item}" for item in self.items) + + +class Reporter: + def __init__(self, output: TextIO) -> None: + self.output = output + + def write(self, element: ReportElement) -> None: + print(f"{element}\n", file=self.output) + + +ReportStream: TypeAlias = Iterator[ReportElement] + + +class LinterBase(ABC): + def stream_reports(self) -> ReportStream: + ... + + +class AirfieldLinter(LinterBase): + def stream_reports(self) -> ReportStream: + yield H2("Airfield data") + yield Paragraph( + "This section lists airfields with missing data. Not all data is able to " + "be exported from DCS directly, and the data reported in this section must " + "be maintained by hand. For any airfield listed in this section, " + "Liberation will not be able to (not necessarily a complete list):" + ) + yield UnorderedList( + [ + "Include ATC, TACAN, or ILS information in the kneeboard", + "Configure preset radio channels for capable aircraft", + ] + ) + for theater in TheaterLoader.each(): + yield from self.document_airfields_with_missing_data(theater) + self.iter_airfields_with_no_data(theater) + + def document_airfields_with_missing_data( + self, theater: ConflictTheater + ) -> ReportStream: + airfields = list(self.iter_airfields_with_no_data(theater)) + + yield H3(theater.terrain.name) + if not airfields: + yield Paragraph("No missing data.") + return + + yield Paragraph("The following airfields are missing data:") + yield UnorderedList(airfields) + + @staticmethod + def iter_airfields_with_no_data(theater: ConflictTheater) -> Iterator[str]: + for airfield in theater.terrain.airport_list(): + try: + AirfieldData.for_airport(theater, airfield) + except KeyError: + yield airfield.name + + +class UncheckedDataLinter(LinterBase): + def stream_reports(self) -> ReportStream: + yield H2("Unchecked data") + yield Paragraph("Any types of data not mentioned above are **not checked**.") + + +class Linter(LinterBase): + def __init__(self, output: TextIO) -> None: + self.reporter = Reporter(output) + + def run(self) -> None: + for report in self.stream_reports(): + self.reporter.write(report) + + def stream_reports(self) -> ReportStream: + yield H1("Liberation data report") + yield self.describe_version() + yield Paragraph( + "This report documents missing supplemental data in Liberation. This is " + "only able to report data that is missing as compared to pydcs. If pydcs " + "is missing DCS data, that cannot be reported." + ) + yield Paragraph( + "**Accuracy of data cannot be verified by this report.** If data not " + "mentioned in this report is present but **wrong**, file a bug." + ) + yield from AirfieldLinter().stream_reports() + yield from UncheckedDataLinter().stream_reports() + + def describe_version(self) -> ReportElement: + sha = subprocess.run( + ["git", "rev-parse", "--short", "HEAD"], + check=True, + capture_output=True, + text=True, + ).stdout.strip() + return Paragraph( + f"This report was generated for DCS Liberation {VERSION} ({sha}) on " + f"{date.today()} with pydcs {self.describe_pydcs()}." + ) + + def describe_pydcs(self) -> str: + result = subprocess.run( + ["pip", "freeze"], check=True, capture_output=True, text=True + ) + pydcs_lines = [l for l in result.stdout.splitlines() if "pydcs" in l] + if len(pydcs_lines) != 1: + raise RuntimeError( + "Could not find unique pydcs package in `pip freeze` output:\n" + f"{result.stdout}" + ) + version = pydcs_lines[0] + if version.startswith("-e git+"): + return self.format_pip_git(version) + return version + + @staticmethod + def format_pip_git(version: str) -> str: + _, _, version = version.partition("+") + if version.endswith("#egg=pydcs"): + version, _, _ = version.rpartition("#") + _, _, sha = version.partition("@") + return sha + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser() + + parser.add_argument( + "output_path", + type=Path, + nargs="?", + help=( + "Write the report to the given path. If omitted, the report will be " + "written to stdout." + ), + ) + + return parser.parse_args() + + +def main() -> None: + logging.basicConfig(level=logging.DEBUG) + args = parse_args() + if args.output_path is None: + context = contextlib.nullcontext(sys.stdout) + else: + context = args.output_path.open("w") + with context as output: + Linter(output).run() + + +if __name__ == "__main__": + main()