Coverage for event_normalizer/schema.py: 100%

1"""Mappings between GraceDB source paths and normalized flat field names.""" 

2 

3from __future__ import annotations 

4 

5from collections.abc import Iterable 

6 

7from .models import EventLabels 

8 

9SUPPORTED_IFOS = ( 

10 "H1", 

11 "L1", 

12 "V1", 

13 "K1", 

14) 

15 

16 

17def get_normalized_field_names( 

18 source_paths: Iterable[str], 

19) -> list[str]: 

20 """ 

21 Return normalized flat field names for GraceDB source paths. 

22 

23 Parameters 

24 ---------- 

25 source_paths: 

26 Dot-separated source paths from a GraceDB payload. 

27 

28 Returns 

29 ------- 

30 list[str] 

31 Unique normalized flat field names, preserving source-path order. 

32 

33 Examples 

34 -------- 

35 >>> get_normalized_field_names( 

36 ... [ 

37 ... "graceid", 

38 ... "extra_attributes.CoincInspiral.mass", 

39 ... "extra_attributes.SingleInspiral.snr", 

40 ... ] 

41 ... ) 

42 [ 

43 "uid", 

44 "coinc_mass", 

45 "single_H1_single_snr", 

46 "single_L1_single_snr", 

47 "single_V1_single_snr", 

48 "single_K1_single_snr", 

49 ] 

50 """ 

51 field_names: list[str] = [] 

52 

53 for source_path in source_paths: 

54 for field_name in get_normalized_names_for_path(source_path): 

55 if field_name not in field_names: 

56 field_names.append(field_name) 

57 

58 return field_names 

59 

60 

61def get_normalized_names_for_path( 

62 source_path: str, 

63) -> list[str]: 

64 """ 

65 Return normalized flat field names for one GraceDB source path. 

66 

67 A source path may correspond to multiple normalized fields. For example, 

68 SingleInspiral detector-specific values expand once per supported IFO. 

69 """ 

70 # Labels are handled differently by different consumers. The normalizer 

71 # parses them into EventLabels, but does not prescribe an ingestion policy. 

72 if source_path == "labels": 

73 return list(EventLabels.model_fields) 

74 

75 # GraceDB event identifier. 

76 if source_path == "graceid": 

77 return ["uid"] 

78 

79 # The source instruments representation becomes scalar detector flags. 

80 if source_path == "instruments": 

81 return [f"instruments_{ifo}" for ifo in SUPPORTED_IFOS] 

82 

83 coinc_prefix = "extra_attributes.CoincInspiral." 

84 if source_path.startswith(coinc_prefix): 

85 field = source_path.removeprefix(coinc_prefix) 

86 return [f"coinc_{field}"] 

87 

88 burst_prefix = "extra_attributes.MultiBurst." 

89 if source_path.startswith(burst_prefix): 

90 field = source_path.removeprefix(burst_prefix) 

91 

92 if field == "hoft": 

93 return _channel_fields() 

94 

95 return [f"burst_{field}"] 

96 

97 mly_scores_prefix = "extra_attributes.MLyBurst.scores." 

98 if source_path.startswith(mly_scores_prefix): 

99 score = source_path.removeprefix(mly_scores_prefix) 

100 return [f"mly_scores_{score}"] 

101 

102 mly_prefix = "extra_attributes.MLyBurst." 

103 if source_path.startswith(mly_prefix): 

104 field = source_path.removeprefix(mly_prefix) 

105 

106 if field == "channels": 

107 return _channel_fields() 

108 

109 return [f"mly_{field}"] 

110 

111 single_prefix = "extra_attributes.SingleInspiral." 

112 if source_path.startswith(single_prefix): 

113 field = source_path.removeprefix(single_prefix) 

114 

115 if field == "channel": 

116 return _channel_fields() 

117 

118 # The detector identity is represented by single_H1, single_L1, etc. 

119 if field == "ifo": 

120 return [] 

121 

122 return [f"single_{ifo}_single_{field}" for ifo in SUPPORTED_IFOS] 

123 

124 # Root-level normalized fields normally retain their source name. 

125 return [source_path] 

126 

127 

128def _channel_fields() -> list[str]: 

129 """Return normalized root-level channel field names.""" 

130 return [f"{ifo}_channel" for ifo in SUPPORTED_IFOS]