parser.py

import re
import sys
from typing import Final

from pydantic import BaseModel, ConfigDict, model_validator
from pydantic_core import PydanticCustomError
from typing_extensions import Self

from module import Connection, Zone, ZoneFactory


class MapData(BaseModel):
    """Validated map content and its parsed entities."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    content: str

    zone_list: list[Zone] = []
    connection_list: list[Connection] = []
    nb_drones: int = 0

    # Regex patterns for validating the map data
    _match_nb_drones: Final[re.Pattern[str]] = re.compile(
        r"""
        nb_drones:\s(?P<nb_drones>-?\d+)$
    """,
        re.VERBOSE | re.M,
    )

    _match_start_hub: Final[re.Pattern[str]] = re.compile(
        r"""
        ^start_hub:\s(?P<name>[^\s-]+)\s(?P<x>-?\d+)\s(?P<y>-?\d+)
        (?:\s\[
            (?P<metadata>.*)
        \])?$
    """,
        re.VERBOSE | re.M,
    )

    _match_end_hub: Final[re.Pattern[str]] = re.compile(
        r"""
        ^end_hub:\s(?P<name>[^\s-]+)\s(?P<x>-?\d+)\s(?P<y>-?\d+)
        (?:\s\[
            (?P<metadata>.*)
        \])?$
    """,
        re.VERBOSE | re.M,
    )

    _match_hub: Final[re.Pattern[str]] = re.compile(
        r"""
        ^hub:\s(?P<name>[^\s-]+)\s(?P<x>-?\d+)\s(?P<y>-?\d+)
        (?:\s\[
            (?P<metadata>.*)
        \])?$
    """,
        re.VERBOSE | re.M,
    )

    _match_connection: Final[re.Pattern[str]] = re.compile(
        r"""
        ^connection:\s(?P<zone1>[^\s-]+)-(?P<zone2>[^\s-]+)
        (?:\s\[
            (?P<metadata>.*)
        \])?$
    """,
        re.VERBOSE | re.M,
    )

    @model_validator(mode="after")
    def validate_data(self) -> Self:
        """Validate and parse the raw map content.

        Returns:
            The validated model instance.

        Raises:
            PydanticCustomError: If the map content violates format rules.
            ValueError: If a line cannot be parsed.
        """
        matched: Final = re.findall(
            r"^(nb_drones|start_hub|end_hub)", self.content, re.M
        )

        if matched.count("nb_drones") != 1:
            raise PydanticCustomError(
                "nb_drones",
                "Exactly one nb_drones line is required",
            )
        elif matched.count("start_hub") != 1:
            raise PydanticCustomError(
                "start_hub",
                "Exactly one start_hub line is required",
            )
        elif matched.count("end_hub") != 1:
            raise PydanticCustomError(
                "end_hub",
                "Exactly one end_hub line is required",
            )

        unique_zone_names = []
        unique_zone_coordinates = []
        unique_connection_pairs = []
        empty_lines_count = 0
        for n_line, line in enumerate(self.content.splitlines(), 1):
            line = line.rstrip()

            try:
                if line.lstrip().startswith("#") or not line.strip():
                    empty_lines_count += 1
                    continue
                elif line.find("#") > 0:
                    line = line[: line.find("#")].rstrip()

                if (n_line == empty_lines_count + 1 and
                   not self._match_nb_drones.match(line)):
                    raise PydanticCustomError(
                        "invalid_map_data",
                        "The first non-empty line must be nb_drones"
                    )

                match_line = self._match_nb_drones.match(line)
                if match_line:
                    self.nb_drones = int(match_line.group("nb_drones"))
                    if self.nb_drones <= 0:
                        raise PydanticCustomError(
                            "invalid_nb_drones",
                            "The number of drones must be a positive integer",
                        )
                    continue

                match_line = self._match_start_hub.match(line)
                if match_line:
                    name, x, y, metadata = match_line.groups()
                    zone_coordinates = (int(x), int(y))
                    if name in unique_zone_names:
                        raise PydanticCustomError(
                            "unique name",
                            "start_hub name must be unique",
                        )
                    elif zone_coordinates in unique_zone_coordinates:
                        raise PydanticCustomError(
                            "unique coordinates",
                            "start_hub coordinates must be unique",
                        )
                    zone_start = ZoneFactory.create(
                        "start", name, zone_coordinates, metadata
                    )
                    self.zone_list.insert(0, zone_start)
                    unique_zone_names.append(name)
                    unique_zone_coordinates.append(zone_coordinates)
                    continue

                match_line = self._match_end_hub.match(line)
                if match_line:
                    name, x, y, metadata = match_line.groups()
                    zone_coordinates = (int(x), int(y))
                    if name in unique_zone_names:
                        raise PydanticCustomError(
                            "unique name",
                            "end_hub name must be unique",
                        )
                    elif zone_coordinates in unique_zone_coordinates:
                        raise PydanticCustomError(
                            "unique coordinates",
                            "end_hub coordinates must be unique",
                        )
                    zone_end = ZoneFactory.create(
                        "end", name, zone_coordinates, metadata
                    )
                    end_hub_obj = zone_end
                    self.zone_list.append(zone_end)
                    unique_zone_names.append(name)
                    unique_zone_coordinates.append(zone_coordinates)
                    continue

                match_line = self._match_hub.match(line)
                if match_line:
                    name, x, y, metadata = match_line.groups()
                    zone_coordinates = (int(x), int(y))
                    if name in unique_zone_names:
                        raise PydanticCustomError(
                            "unique name",
                            "hub name must be unique",
                        )
                    elif zone_coordinates in unique_zone_coordinates:
                        raise PydanticCustomError(
                            "unique coordinates",
                            "hub coordinates must be unique",
                        )
                    zone_hub = ZoneFactory.create(
                        "hub", name, zone_coordinates, metadata
                    )
                    self.zone_list.append(zone_hub)
                    unique_zone_names.append(name)
                    unique_zone_coordinates.append(zone_coordinates)
                    continue

                match_line = self._match_connection.match(line)
                if match_line:
                    zone1, zone2, metadata = match_line.groups()
                    if (zone1 not in unique_zone_names or
                       zone2 not in unique_zone_names):
                        raise PydanticCustomError(
                            "defined zones",
                            "Zones must be defined before connections"
                        )
                    if (tuple(sorted((zone1, zone2))) in
                       unique_connection_pairs):
                        raise PydanticCustomError(
                            "duplicates",
                            "Connection must not appear more than once",
                        )
                    zone1_obj = [zone for zone in self.zone_list
                                 if zone.name == zone1][0]
                    zone2_obj = [zone for zone in self.zone_list
                                 if zone.name == zone2][0]
                    connection = Connection(zone1_obj, zone2_obj, metadata)
                    self.connection_list.append(connection)
                    unique_connection_pairs.append(
                        tuple(
                            sorted((zone1, zone2))
                        )
                    )
                    continue

                raise ValueError("Invalid line format")
            except ValueError as e:
                raise PydanticCustomError(
                    "Parser Error",
                    f"Line {n_line}: {line}\n{e}"
                )

        self.zone_list.remove(end_hub_obj)
        self.zone_list.insert(len(self.zone_list), end_hub_obj)

        return self


class Parser:
    def __init__(self, map_file: str) -> None:
        """Create a parser for a map file.

        Args:
            map_file: Path to the input map file.
        """
        self.map_file = map_file

    def load_map_data(self) -> MapData:
        """Read the file and return validated map data.

        Returns:
            Parsed and validated map data.

        Raises:
            SystemExit: If the file cannot be read.
        """
        try:
            with open(self.map_file, "r") as file_obj:
                file_data = file_obj.read()

                return MapData(content=file_data)
        except IOError as e:
            print(e)
            sys.exit(1)

    @staticmethod
    def test_data(data: MapData) -> None:
        """Print parsed map data for debugging.

        Args:
            data: Parsed map data to display.
        """
        print("nb_drones:", data.nb_drones)

        for zone in data.zone_list:
            print(
                f"{zone.__class__.__name__}: {zone.name} "
                f"{' '.join(str(c) for c in zone.coordinates)}",
                zone.default_metadata,
            )

        for connection in data.connection_list:
            print(
                f"Connection: {connection.zone1.name} - "
                f"{connection.zone2.name} "
                f"{connection.default_metadata}"
            )