Coverage for core / src / sensorkit / std / collect.py: 85%

1# SPDX-License-Identifier: Apache-2.0 

2from __future__ import annotations 

3 

4from datetime import datetime 

5from typing import Literal, Optional 

6 

7from pydantic import BaseModel, Field 

8 

9import sensorkit.api as sk 

10from sensorkit.astro.common import ReferenceFrame 

11from sensorkit.astro.target import ( 

12 AltAzTarget, 

13 EphemerisTarget, 

14 FrameTarget, 

15 ICRSTarget, 

16 RateTarget, 

17 StateVectorTarget, 

18 Target, 

19 TLETarget, 

20) 

21from sensorkit.std.instrument import FrameType 

22 

23 

24class CameraParameterSet(BaseModel): 

25 """Camera Parameter Set""" 

26 

27 frame_type: FrameType = Field(default=FrameType.LIGHT) 

28 integration_time_seconds: float = Field(...) 

29 frame_count: int = Field(...) 

30 filter_name: Optional[str] = Field(default=None) 

31 readout_mode: Optional[int] = Field(default=None) 

32 gain: Optional[float] = Field(default=None) 

33 binning_x: Optional[int] = Field(default=None) 

34 binning_y: Optional[int] = Field(default=None) 

35 

36 

37@sk.declare_keyword 

38class Collect(BaseModel): 

39 """Collect context information snapshot. 

40 

41 This keyword represents a snapshot of the complete collect state at a specific point 

42 in time during collection—for example, before an individual frame is captured. It 

43 carries the target being observed, camera parameters, and the current position in 

44 the frame sequence. It is primarily used with the standard collect task and other 

45 compatible observation tasks. 

46 

47 Attributes: 

48 target: The astronomical target to observe (can be various target types like TLE, 

49 catalog object, ephemeris, etc.) 

50 target_id: Optional name or identifier of the object being observed. 

51 params: Camera configuration parameters including integration time, frame count, 

52 binning, gain, frame type, and filter 

53 frame_number: Current frame number in the collection sequence (default: 0) 

54 """ 

55 

56 target: Target 

57 target_id: str | None = None 

58 params: CameraParameterSet 

59 frame_number: int = 0 

60 

61 @property 

62 def track_mode(self) -> Literal["fixed", "sidereal", "rate", "unknown"]: 

63 match self.target: 

64 case TLETarget() | RateTarget() | EphemerisTarget() | StateVectorTarget(): 

65 return "rate" 

66 case ICRSTarget() | FrameTarget(frame=ReferenceFrame.ICRF): 

67 return "sidereal" 

68 case AltAzTarget() | FrameTarget(frame=ReferenceFrame.ALTAZ): 

69 return "fixed" 

70 case FrameTarget(): 

71 return "rate" 

72 

73 return "unknown" 

74 

75 def get_fits_cards(self): 

76 yield "TARGET", (self.target_id, "Target ID") 

77 yield "TRKMODE", (self.track_mode, "Tracking mode") 

78 yield "FRAMENUM", (self.frame_number, "Frame number") 

79 

80 

81class StandardCollectTask(sk.CollectTask): 

82 """Task to collect one or more camera frames while tracking a target. 

83 

84 This task executes a standard observation by configuring a camera to capture a sequence 

85 of frames while a mount tracks a specified target. It handles the complete observation 

86 workflow including target tracking, camera configuration, auxiliary device sequencing, 

87 and frame acquisition. 

88 

89 Implementations must build and propagate context data to any operations that require it. 

90 The base context must be taken from the `TaskExecution`. (FIXME: do this internally) 

91 

92 Mandatory context keywords: 

93 - Collect: Snapshot of the current collect state, prior to each operation 

94 - AltAzPointing 

95 - RADecPointing 

96 - AxisRates 

97 - SitePosition 

98 

99 Attributes: 

100 task_type: Always "standard_collect" for this task type 

101 target: The astronomical target to observe (e.g., TLE, catalog object, ephemeris, 

102 ICRS coordinates, alt-az position, etc.) 

103 camera_params: Camera configuration including integration time, frame count, 

104 binning, gain, frame type, and filter selection 

105 end_time: *Advisory* deadline for task completion. Deprecated and slated for removal 

106 sidereal_frames: List of frame numbers that should use sidereal tracking mode. 

107 Defaults to empty list 

108 target_id: Optional name or identifier of the object being observed. If not supplied, 

109 the handler may infer an identifier (e.g., NORAD ID for TLE targets, catalog name 

110 for catalog targets) on a per-target basis 

111 """ 

112 

113 task_type: Literal["standard_collect"] = "standard_collect" 

114 target: Target 

115 camera_params: CameraParameterSet 

116 end_time: datetime | None = None 

117 sidereal_frames: list[int] = Field(default_factory=list) 

118 target_id: str | None = None