# Licensed under a 3-clause BSD style license - see LICENSE.rstimporthtmlimportjsonimportloggingfromcollectionsimportdefaultdictfromcollections.abcimportMappingfromenumimportEnumfrompathlibimportPathfromtypingimportList,OptionalfromastropyimportunitsasuimportyamlfrompydanticimportBaseModel,ConfigDictfromgammapy.makersimportMapDatasetMakerfromgammapy.utils.scriptsimportmake_path,read_yamlfromgammapy.utils.typesimportAngleType,EnergyType,PathType,TimeType__all__=["AnalysisConfig"]CONFIG_PATH=Path(__file__).resolve().parent/"config"DOCS_FILE=CONFIG_PATH/"docs.yaml"log=logging.getLogger(__name__)defdeep_update(d,u):"""Recursively update a nested dictionary. Taken from: https://stackoverflow.com/a/3233356/19802442 """fork,vinu.items():ifisinstance(v,Mapping):d[k]=deep_update(d.get(k,{}),v)else:d[k]=vreturndclassReductionTypeEnum(str,Enum):spectrum="1d"cube="3d"classFrameEnum(str,Enum):icrs="icrs"galactic="galactic"classRequiredHDUEnum(str,Enum):events="events"gti="gti"aeff="aeff"bkg="bkg"edisp="edisp"psf="psf"rad_max="rad_max"classBackgroundMethodEnum(str,Enum):reflected="reflected"fov="fov_background"ring="ring"classSafeMaskMethodsEnum(str,Enum):aeff_default="aeff-default"aeff_max="aeff-max"edisp_bias="edisp-bias"offset_max="offset-max"bkg_peak="bkg-peak"classMapSelectionEnum(str,Enum):counts="counts"exposure="exposure"background="background"psf="psf"edisp="edisp"classGammapyBaseConfig(BaseModel):model_config=ConfigDict(arbitrary_types_allowed=True,validate_assignment=True,extra="forbid",validate_default=True,use_enum_values=True,json_encoders={u.Quantity:lambdav:f"{v.value}{v.unit}"},)def_repr_html_(self):try:returnself.to_html()exceptAttributeError:returnf"<pre>{html.escape(str(self))}</pre>"classSkyCoordConfig(GammapyBaseConfig):frame:Optional[FrameEnum]=Nonelon:Optional[AngleType]=Nonelat:Optional[AngleType]=NoneclassEnergyAxisConfig(GammapyBaseConfig):min:Optional[EnergyType]=Nonemax:Optional[EnergyType]=Nonenbins:Optional[int]=NoneclassSpatialCircleConfig(GammapyBaseConfig):frame:Optional[FrameEnum]=Nonelon:Optional[AngleType]=Nonelat:Optional[AngleType]=Noneradius:Optional[AngleType]=NoneclassEnergyRangeConfig(GammapyBaseConfig):min:Optional[EnergyType]=Nonemax:Optional[EnergyType]=NoneclassTimeRangeConfig(GammapyBaseConfig):start:Optional[TimeType]=Nonestop:Optional[TimeType]=NoneclassFluxPointsConfig(GammapyBaseConfig):energy:EnergyAxisConfig=EnergyAxisConfig()source:str="source"parameters:dict={"selection_optional":"all"}classLightCurveConfig(GammapyBaseConfig):time_intervals:TimeRangeConfig=TimeRangeConfig()energy_edges:EnergyAxisConfig=EnergyAxisConfig()source:str="source"parameters:dict={"selection_optional":"all"}classFitConfig(GammapyBaseConfig):fit_range:EnergyRangeConfig=EnergyRangeConfig()classExcessMapConfig(GammapyBaseConfig):correlation_radius:AngleType="0.1 deg"parameters:dict={}energy_edges:EnergyAxisConfig=EnergyAxisConfig()classBackgroundConfig(GammapyBaseConfig):method:Optional[BackgroundMethodEnum]=Noneexclusion:Optional[PathType]=Noneparameters:dict={}classSafeMaskConfig(GammapyBaseConfig):methods:List[SafeMaskMethodsEnum]=[SafeMaskMethodsEnum.aeff_default]parameters:dict={}classEnergyAxesConfig(GammapyBaseConfig):energy:EnergyAxisConfig=EnergyAxisConfig(min="1 TeV",max="10 TeV",nbins=5)energy_true:EnergyAxisConfig=EnergyAxisConfig(min="0.5 TeV",max="20 TeV",nbins=16)classSelectionConfig(GammapyBaseConfig):offset_max:AngleType="2.5 deg"classWidthConfig(GammapyBaseConfig):width:AngleType="5 deg"height:AngleType="5 deg"classWcsConfig(GammapyBaseConfig):skydir:SkyCoordConfig=SkyCoordConfig()binsize:AngleType="0.02 deg"width:WidthConfig=WidthConfig()binsize_irf:AngleType="0.2 deg"classGeomConfig(GammapyBaseConfig):wcs:WcsConfig=WcsConfig()selection:SelectionConfig=SelectionConfig()axes:EnergyAxesConfig=EnergyAxesConfig()classDatasetsConfig(GammapyBaseConfig):type:ReductionTypeEnum=ReductionTypeEnum.spectrumstack:bool=Truegeom:GeomConfig=GeomConfig()map_selection:List[MapSelectionEnum]=MapDatasetMaker.available_selectionbackground:BackgroundConfig=BackgroundConfig()safe_mask:SafeMaskConfig=SafeMaskConfig()on_region:SpatialCircleConfig=SpatialCircleConfig()containment_correction:bool=TrueclassObservationsConfig(GammapyBaseConfig):datastore:PathType=Path("$GAMMAPY_DATA/hess-dl3-dr1/")obs_ids:List[int]=[]obs_file:Optional[PathType]=Noneobs_cone:SpatialCircleConfig=SpatialCircleConfig()obs_time:TimeRangeConfig=TimeRangeConfig()required_irf:List[RequiredHDUEnum]=["aeff","edisp","psf","bkg"]classLogConfig(GammapyBaseConfig):level:str="info"filename:Optional[PathType]=Nonefilemode:Optional[str]=Noneformat:Optional[str]=Nonedatefmt:Optional[str]=NoneclassGeneralConfig(GammapyBaseConfig):log:LogConfig=LogConfig()outdir:str="."n_jobs:int=1datasets_file:Optional[PathType]=Nonemodels_file:Optional[PathType]=None
[docs]classAnalysisConfig(GammapyBaseConfig):"""Gammapy analysis configuration."""general:GeneralConfig=GeneralConfig()observations:ObservationsConfig=ObservationsConfig()datasets:DatasetsConfig=DatasetsConfig()fit:FitConfig=FitConfig()flux_points:FluxPointsConfig=FluxPointsConfig()excess_map:ExcessMapConfig=ExcessMapConfig()light_curve:LightCurveConfig=LightCurveConfig()def__str__(self):"""Display settings in pretty YAML format."""info=self.__class__.__name__+"\n\n\t"data=self.to_yaml()data=data.replace("\n","\n\t")info+=datareturninfo.expandtabs(tabsize=4)
[docs]@classmethoddefread(cls,path):"""Read from YAML file."""config=read_yaml(path)returnAnalysisConfig(**config)
[docs]@classmethoddeffrom_yaml(cls,config_str):"""Create from YAML string."""settings=yaml.safe_load(config_str)returnAnalysisConfig(**settings)
[docs]defwrite(self,path,overwrite=False):"""Write to YAML file."""path=make_path(path)ifpath.exists()andnotoverwrite:raiseIOError(f"File exists already: {path}")path.write_text(self.to_yaml())
[docs]defto_yaml(self):"""Convert to YAML string."""data=json.loads(self.model_dump_json())returnyaml.dump(data,sort_keys=False,indent=4,width=80,default_flow_style=None)
[docs]defset_logging(self):"""Set logging config. Calls ``logging.basicConfig``, i.e. adjusts global logging state. """self.general.log.level=self.general.log.level.upper()logging.basicConfig(**self.general.log.model_dump())log.info("Setting logging config: {!r}".format(self.general.log.model_dump()))
[docs]defupdate(self,config=None):"""Update config with provided settings. Parameters ---------- config : str or `AnalysisConfig` object, optional Configuration settings provided in dict() syntax. Default is None. """ifisinstance(config,str):other=AnalysisConfig.from_yaml(config)elifisinstance(config,AnalysisConfig):other=configelse:raiseTypeError(f"Invalid type: {config}")config_new=deep_update(self.model_dump(exclude_defaults=True),other.model_dump(exclude_defaults=True),)returnAnalysisConfig(**config_new)
@staticmethoddef_get_doc_sections():"""Return dictionary with commented docs from docs file."""doc=defaultdict(str)withopen(DOCS_FILE)asf:forlineinfilter(lambdaline:notline.startswith("---"),f):line=line.strip("\n")ifline.startswith("# Section: "):keyword=line.replace("# Section: ","")doc[keyword]+=line+"\n"returndoc