raplan_rws ¶

def dynamische_combi(
    self,
    ingrepen: dict[UUID, raplan.Procedure],
    maintenance_naar_geplande_maatregel: dict[UUID, GeplandeMaatregel],
) -> Callable[[frozenset[UUID]], int | float | None]:
    """Maak een dynamische combi-functie voor het belonen/straffen van samenvallende ingrepen
    op basis van deze instellingen en het project.
    """

    def ingreep_combi(uuids: frozenset[UUID]) -> int | float | None:
        """Dynamische combi-functie voor het belonen/afstraffen van samenvallende ingrepen."""

        if uuids in self.specifieke_combis:
            return self.specifieke_combis[uuids]

        score: int | float | None = None
        volgorde = sorted(uuids)
        zelfde_complex = self.jaar_zelfde_complex
        for i, a in enumerate(volgorde):
            iga: raplan.Procedure = ingrepen[a]
            ma: raplan.Maintenance = iga.maintenance[0]
            gma: GeplandeMaatregel = maintenance_naar_geplande_maatregel[ma.uuid]
            boa: BeheerObject = gma.beheerobject
            for b in volgorde[i + 1 :]:
                igb: raplan.Procedure = ingrepen[b]
                mb: raplan.Maintenance = igb.maintenance[0]
                gmb: GeplandeMaatregel = maintenance_naar_geplande_maatregel[mb.uuid]
                bob: BeheerObject = gmb.beheerobject

                # TODO: implement netwerkanalyse hier.

                if (
                    zelfde_complex is not None
                    and boa.complex_code
                    and bob.complex_code
                    and boa.complex_code == bob.complex_code
                ):
                    score = zelfde_complex if score is None else score + zelfde_complex

        return score

    return ingreep_combi

def dynamische_uitsluiting(
    self,
    ingrepen: dict[UUID, raplan.Procedure],
    maintenance_naar_geplande_maatregel: dict[UUID, GeplandeMaatregel],
) -> Callable[[frozenset[UUID]], bool]:
    """Maak een dynamische uitsluitfunctie voor ingrepen op basis van deze instellingen en het
    project."""

    def ingreep_excl(uuids: frozenset[UUID]) -> bool:
        """Een dynamische uitsluitfunctie voor ingrepen."""

        if uuids in self.uitgesloten_combis:
            return True

        volgorde = sorted(uuids)
        verbieding = self.verbied_zelfde_complex
        for i, a in enumerate(volgorde):
            iga: raplan.Procedure = ingrepen[a]
            ma: raplan.Maintenance = iga.maintenance[0]
            gma: GeplandeMaatregel = maintenance_naar_geplande_maatregel[ma.uuid]
            boa: BeheerObject = gma.beheerobject
            for b in volgorde[i + 1 :]:
                igb: raplan.Procedure = ingrepen[b]
                mb: raplan.Maintenance = igb.maintenance[0]
                gmb: GeplandeMaatregel = maintenance_naar_geplande_maatregel[mb.uuid]
                bob: BeheerObject = gmb.beheerobject

                # TODO: implement netwerkanalyse hier.

                # Zelfde complex uitsluiten.
                if (
                    verbieding
                    and boa.complex_code
                    and bob.complex_code
                    and boa.complex_code == bob.complex_code
                ):
                    return True

        return False

    return ingreep_excl

def maak_kopie(self) -> "IngreepInstellingen":
    """Maak een kopie van deze instellingen."""
    return deepcopy(self)

def dynamische_combi(
    self,
    maintenance_naar_geplande_maatregel: dict[UUID, GeplandeMaatregel],
) -> Callable[[frozenset[UUID]], int | float | None]:
    """Maak een dynamische maatregel-combinatie beloningsfunctie op basis van deze instellingen
    en het project.
    """

    def maatregel_combi(uuids: frozenset[UUID]) -> int | float | None:
        """Een dynamische maatregel-combinatie belongingsfunctie."""

        if uuids in self.specifieke_combis:
            return self.specifieke_combis[uuids]

        beste: int | float = 0
        volgorde = sorted(uuids)
        kosten_totaal: int | float = 0.0
        for i, a in enumerate(volgorde):
            gma: GeplandeMaatregel = maintenance_naar_geplande_maatregel[a]
            kosten_totaal += gma.raming_incl
            for b in volgorde[i + 1 :]:
                gmb: GeplandeMaatregel = maintenance_naar_geplande_maatregel[b]

                # neem steeds het minimum omdat negatievere penalties beter zijn!
                for dis_a in gma.disciplines:
                    for dis_b in gmb.disciplines:
                        if dis_a == dis_b:
                            beste = min(beste, self.zelfde_discipline)
                        try:
                            combi: frozenset[Discipline] = frozenset((dis_a, dis_b))
                            score: float | int = self.discipline_combis[combi]
                            beste = min(beste, score)
                        except KeyError:
                            continue

        return None if beste == 0 else beste * kosten_totaal

    return maatregel_combi

def dynamische_uitsluiting(
    self,
    maintenance_naar_geplande_maatregel: dict[UUID, GeplandeMaatregel],
) -> Callable[[frozenset[UUID]], bool]:
    """Maak een dynamische uitsluitfunctie op basis van deze instellingen en het project."""

    def maatregel_excl(uuids: frozenset[UUID]) -> bool:
        """Een dynamische uitsluitfunctie voor combinaties van maatregelen."""

        if uuids in self.uitgesloten_combis:
            return True

        if not self.verbied_duplicatie:
            return False

        taken: set[tuple[BeheerObject, ElementBouwdeel, str]] = set()
        for uuid in uuids:
            gm: GeplandeMaatregel = maintenance_naar_geplande_maatregel[uuid]
            taak = (gm.beheerobject, gm.element_bouwdeel, gm.maatregel)

            if taak in taken:
                return True
            else:
                taken.add(taak)

        return False

    return maatregel_excl

def maak_kopie(self) -> "MaatregelSynchronisatie":
    """Maak een kopie van deze instellingen."""
    return deepcopy(self)

def dynamische_functie(
    self,
    maintenance_naar_geplande_maatregel: dict[UUID, GeplandeMaatregel],
) -> Callable[[UUID, float], int | float | None]:
    """Maak een dynamische verschuivingsfunctie op basis van de instellingen en het project."""

    def verschuiving(maintenance_id: UUID, delta: float) -> int | float | None:
        naar_voren = delta < 0
        naar_achter = int(not naar_voren)  # 1 voor naar achter, anders 0

        if maintenance_id in self.specifiek:
            spec = self.specifiek[maintenance_id]
            if isinstance(spec, (int, float)):
                return spec * abs(delta)
            else:
                return abs(delta) * spec[naar_achter]

        if naar_voren and self.gratis_naar_voren:
            return 0

        afstand = abs(delta)

        if isinstance(self.vaste_factor, (int, float)):
            kosten = afstand * self.vaste_factor
        else:
            factor = self.vaste_factor[naar_achter]
            kosten = afstand * factor

        if self.kosten_per_jaar_factor != 0:
            if isinstance(self.kosten_per_jaar_factor, (int, float)):
                factor = self.kosten_per_jaar_factor
            else:
                factor = self.kosten_per_jaar_factor[naar_achter]
            maatregel = maintenance_naar_geplande_maatregel[maintenance_id]
            kosten += afstand * factor * maatregel.frequentie() * maatregel.raming_incl

        if self.frequentie_factor != 0:
            if isinstance(self.frequentie_factor, (int, float)):
                factor = self.frequentie_factor
            else:
                factor = self.frequentie_factor[naar_achter]
            kosten += afstand * factor * maatregel.frequentie()

        return kosten

    return verschuiving

def maak_kopie(self) -> "MaatregelVerschuiving":
    """Maak een kopie van deze instellingen."""
    return deepcopy(self)

def maak_kopie(self) -> "ModelInstellingen":
    """Maak een kopie van deze instellingen."""
    return deepcopy(self)

def maak_kopie(self) -> "Omvaarroute":
    """Maak een kopie van dit object."""
    return deepcopy(self)

def maak_kopie(self) -> "Onbereikbaar":
    """Maak een kopie van dit object."""
    return deepcopy(self)

def maak_kopie(self) -> "OhsBepData":
    """Maak een kopie van dit object."""
    return deepcopy(self)

@classmethod
def van_excel_bestand(
    cls, bestandspad: str | Path, instellingen: OhsInstellingen
) -> "OhsBepData":
    """Importeer de BEP data uit een OHS Excel-bestand."""
    wb: openpyxl.Workbook = openpyxl.open(bestandspad)
    ws = wb["BEP"]
    assert isinstance(ws, Worksheet)
    return cls.van_excel_worksheet(ws, instellingen=instellingen)

@classmethod
def van_excel_worksheet(cls, ws: Worksheet, instellingen: OhsInstellingen) -> "OhsBepData":
    """Importeer de BEP data uit het BEP sheet uit een OHS Excel-bestand."""
    data = OhsBepData()

    class_fields = ["beheer_object", "element_bouwdeel", "discipline"]

    header_row: int | None = None
    for candidate, cell in enumerate(ws["A:A"]):
        if str(cell.value).lower().replace(" ", "_") == "object":
            header_row = candidate
            break

    assert header_row is not None, (
        "OHS BEP sheet moet altijd een 'object' kolom hebben ergens in kolom A."
    )

    header: dict[str, int] = {
        str(col[header_row].value).lower().replace(" ", "_"): index
        for index, col in enumerate(ws.iter_cols())
    }
    rename = dict(
        beheer_object="object", element_bouwdeel="element_/_bouwdeel", discipline="type"
    )
    for want, old in rename.items():
        header[want] = header.pop(old)

    field_to_index: dict[str, int] = dict()
    for f in class_fields:
        if f in header:
            field_to_index[f] = header[f]

    # Excel rows are 1-based and we are one past the header row.
    for row in ws.iter_rows(min_row=header_row + 2):
        params = {f: row[index].value for f, index in field_to_index.items()}
        if not params["beheer_object"] or not params["element_bouwdeel"]:
            continue
        if (
            "kosten" in params["element_bouwdeel"].lower()
            or "kosten" in params["beheer_object"].lower()
        ):
            continue
        try:
            new = OhsBepRij.van_kolomwaarden(**params, instellingen=instellingen)
            data.rijen.append(new)
        except SerdeError as e:
            logger.warning(f"Kan '{params}' niet inladen als OHS-BEP rij: {e}")
        except ValueError as e:
            logger.warning(f"Kan '{params}' niet inladen als OHS-BEP rij: {e}")

    return data

def maak_kopie(self) -> "OhsBepRij":
    """Maak een kopie van dit object."""
    return deepcopy(self)

def maak_kopie(self) -> "OhsData":
    """Maak een kopie van dit object."""
    return deepcopy(self)

@classmethod
def van_excel_bestand(cls, bestandspad: str | Path, instellingen: OhsInstellingen) -> "OhsData":
    """Importeer de relevante data uit een OHS Excel-bestand."""
    wb: openpyxl.Workbook = openpyxl.open(bestandspad)
    return cls.van_excel_workbook(wb, instellingen=instellingen)

@classmethod
def van_excel_workbook(cls, wb: Workbook, instellingen: OhsInstellingen) -> "OhsData":
    """Importeer de relevante data uit een ingeladen OHS Excel-workbook."""
    bep_sheet = wb["BEP"]
    assert isinstance(bep_sheet, Worksheet)
    bep = OhsBepData.van_excel_worksheet(bep_sheet, instellingen=instellingen)

    element_functie_sheet = wb["Element-Functie"]
    assert isinstance(element_functie_sheet, Worksheet)
    element_functie = OhsElementFunctieData.van_excel_worksheet(
        element_functie_sheet, instellingen=instellingen
    )

    return OhsData(bep=bep, element_functie=element_functie)

def verwerk(self) -> "VerwerkteOhsData":
    """Verwerk deze rauwe data tot een verwerkt data object met beheerobjecten en structuren."""
    result = VerwerkteOhsData()
    result.verwerk_data(self)
    return result

def maak_kopie(self) -> "OhsElementFunctieData":
    """Maak een kopie van dit object."""
    return deepcopy(self)

@classmethod
def van_excel_bestand(
    cls, bestandspad: str | Path, instellingen: OhsInstellingen
) -> "OhsElementFunctieData":
    """Importeer de Element-Functie data uit een OHS Excel-bestand."""
    wb: openpyxl.Workbook = openpyxl.open(bestandspad)
    ws = wb["Element-Functie"]
    assert isinstance(ws, Worksheet)
    return cls.van_excel_worksheet(ws, instellingen=instellingen)

@classmethod
def van_excel_worksheet(
    cls, ws: Worksheet, instellingen: OhsInstellingen
) -> "OhsElementFunctieData":
    """Importeer de Element-Functie data uit het Element-Functie sheet uit een OHS
    Excel-bestand.
    """
    data = OhsElementFunctieData()
    class_fields = [
        "beheer_object",
        "element_bouwdeel",
        "schutten",
        "spuien",
        "kruisen",
        "keren",
        "vgm",
        "secundair",
    ]

    header_row: int | None = None
    for candidate, cell in enumerate(ws["A:A"]):
        if str(cell.value).lower().replace(" ", "_") == "beheer_object":
            header_row = candidate
            break

    assert header_row is not None

    header: dict[str, int] = {
        str(col[header_row].value).lower().replace(" ", "_"): index
        for index, col in enumerate(ws.iter_cols())
    }
    rename = {"element_bouwdeel": "element_-_bouwdeel_naam"}
    for want, old in rename.items():
        header[want] = header.pop(old)

    # Excel rows are 1-based and we are one past the header row.
    beheer_object: str | None = None
    for row in ws.iter_rows(min_row=header_row + 2):
        params = {f: row[header[f]].value for f in class_fields}
        # Object doesn't have an entry in each row, only when it changes.
        nieuw_object = params.pop("beheer_object")
        if nieuw_object:
            candi = str(nieuw_object).strip()
            if candi:
                beheer_object = candi

        if isinstance(beheer_object, str) and "kosten" in beheer_object.lower():
            continue

        if params["element_bouwdeel"] is None or "kosten" in params["element_bouwdeel"].lower():
            continue

        for k, v in params.items():
            if k in {"schutten", "spuien", "kruisen", "keren", "vgm", "secundair"}:
                params[k] = str(v).strip().lower() == "x"
        try:
            new = OhsElementFunctieRij.van_kolomwaarden(
                **params,
                beheer_object=beheer_object,  # type: ignore
                instellingen=instellingen,
            )
            data.rijen.append(new)
        except SerdeError as e:
            logger.warning(f"Kan {params} niet inladen als Element-Functie rij: {e}")
        except ValueError as e:
            logger.warning(f"Kan {params} niet inladen als Element-Functie rij: {e}")

    return data