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

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 

23 

24from allocate.domain_layer.entities.allocated_science_target import AllocatedScienceTarget 

25from allocate.domain_layer.entities.allocation_version import AllocationVersion 

26 

27SourceConflictCheck = Callable[[AllocationVersion], None] 

28 

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""" 

33 

34 

35def gbt_algorithm(av: AllocationVersion): 

36 logging.info("Running the GBT source conflict algorithm") 

37 return [] 

38 

39 

40def vla_algorithm(av: AllocationVersion): 

41 logging.info("Running the VLA source conflict algorithm") 

42 return [] 

43 

44 

45def vlba_algorithm(av: AllocationVersion): 

46 logging.info("Running the VLBA source conflict algorithm") 

47 return [] 

48 

49 

50def unknown_facility_algorithm(av: AllocationVersion): 

51 logging.info("Running the source conflict algorithm for an unknown facility") 

52 return [] 

53 

54 

55FACILITY_ALGORITHMS: defaultdict[str, SourceConflictCheck] = defaultdict( 

56 lambda: unknown_facility_algorithm, GBT=gbt_algorithm, VLA=vla_algorithm, VLBA=vlba_algorithm 

57) 

58 

59 

60class SourceConflictCheckMode(Enum): 

61 ALGORITHM = 0 # Use defined algorithm 

62 # Only one type at this point since it does nothing 

63 

64 def source_conflict_check(self, av: AllocationVersion): 

65 match self: 

66 case self.ALGORITHM: 

67 return FACILITY_ALGORITHMS[av.facility.facility_name](av) 

68 

69 

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