Coverage for middle_layer/allocate/domain_layer/services/source_conflict_service.py: 59.38%
32 statements
« prev ^ index » next coverage.py v7.10.5, created at 2026-04-13 06:13 +0000
« prev ^ index » next coverage.py v7.10.5, created at 2026-04-13 06:13 +0000
1# Copyright 2024 Associated Universities, Inc.
2#
3# This file is part of Telescope Time Allocation Tools (TTAT).
4#
5# TTAT is free software: you can redistribute it and/or modify
6# it under the terms of the GNU General Public License as published by
7# the Free Software Foundation, either version 3 of the License, or
8# any later version.
9#
10# TTAT is distributed in the hope that it will be useful,
11# but WITHOUT ANY WARRANTY; without even the implied warranty of
12# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
13# GNU General Public License for more details.
14#
15# You should have received a copy of the GNU General Public License
16# along with TTAT. If not, see <https://www.gnu.org/licenses/>.
17import itertools
18import logging
19from collections import defaultdict
20from enum import Enum
21from random import choice
22from typing import Callable, Iterator
24from allocate.domain_layer.entities.allocated_science_target import AllocatedScienceTarget
25from allocate.domain_layer.entities.allocation_version import AllocationVersion
27SourceConflictCheck = Callable[[AllocationVersion], None]
29"""
30Source Conflict Checkers are functions that work on an AllocationVersions and its associated list of
31AllocationDispositions and each AD's list of AllocatedScienceTargets
32"""
35def gbt_algorithm(av: AllocationVersion):
36 logging.info("Running the GBT source conflict algorithm")
37 return []
40def vla_algorithm(av: AllocationVersion):
41 logging.info("Running the VLA source conflict algorithm")
42 return []
45def vlba_algorithm(av: AllocationVersion):
46 logging.info("Running the VLBA source conflict algorithm")
47 return []
50def unknown_facility_algorithm(av: AllocationVersion):
51 logging.info("Running the source conflict algorithm for an unknown facility")
52 return []
55FACILITY_ALGORITHMS: defaultdict[str, SourceConflictCheck] = defaultdict(
56 lambda: unknown_facility_algorithm, GBT=gbt_algorithm, VLA=vla_algorithm, VLBA=vlba_algorithm
57)
60class SourceConflictCheckMode(Enum):
61 ALGORITHM = 0 # Use defined algorithm
62 # Only one type at this point since it does nothing
64 def source_conflict_check(self, av: AllocationVersion):
65 match self:
66 case self.ALGORITHM:
67 return FACILITY_ALGORITHMS[av.facility.facility_name](av)
70def check_for_source_conflicts(
71 av: AllocationVersion, mode: SourceConflictCheckMode = SourceConflictCheckMode.ALGORITHM
72) -> list[AllocatedScienceTarget]:
73 """Check for source conflicts for the associated AllocatedScienceTargets found in the AllocationDispositions of the
74 provided Allocation Versions
75 :param av: AllocationVersion containing allocated science targets to check
76 :param mode: Source Conflict Check mode to use
77 :return: List of AllocatedScienceTargets that are conflicted
78 """
79 conflicted_asts = mode.source_conflict_check(av)
80 return conflicted_asts