Developer Tutorial

TIP: add the following to the top of your notebook or ipython script to have it automatically reload the imports as you make changes. That makes it much nicer to develop at the same time as working in the notebook:

%load_ext autoreload
%autoreload 2

Using the EventSource interactively.

Note that you can find detailed tutorials about working with events and EventSources in general in the ctapipe documentation.

The purpose of the HESSEventSource is too make HESS data work with the ctapipe-process command-line tool. Howwever, during development instead of running the tool, it’s often nice to make a small test script to help explore the functionality of the HESSEventSource interactively, either in a python REPL like ipython, or in a Jupyter Notebook.

A basic test is as follows:

%matplotlib inline

from ctapipe.coordinates import EngineeringCameraFrame
from ctapipe.io import EventSource
from ctapipe.utils import get_dataset_path
from ctapipe.visualization import CameraDisplay

# load the test file. Note you can substitute your own local filename here instead
# , e.g. filename="my_local_dst_file.root"
FILENAME = get_dataset_path("example_hess_dst.root")

# now just load the first event in the file:
with EventSource(FILENAME, max_events=1) as source:
    for event in source:
        print(event.index)

/opt/hostedtoolcache/Python/3.13.5/x64/lib/python3.13/site-packages/ctapipe/instrument/camera/geometry.py:616: FromNameWarning: .from_name uses pre-defined data that is likely different from the data being analyzed. Access instrument information via the SubarrayDescription instead.
  warn_from_name()

{'event_id': np.int64(13572096655360), 'obs_id': np.int32(170720)}

Now event is the first loaded event, and source will be an instance of HESSIOEventSource, if everything worked correctly

source

HESSEventSource

EventSource for HESS DSTs.

allowed_tels	None	list of allowed tel_ids, others will be ignored. If None, all telescopes in the input stream will be included (default: None)
input_url	/home/runner/.cache/ctapipe/minio-cta.zeuthen.desy.de/dpps-testdata-public/data/ctapipe-test-data/v1.1.0/example_hess_dst.root	Path to the input file containing events. (default: traitlets.Undefined)
max_events	1	Maximum number of events that will be read from the file (default: None)

Look at attributes of the event source and see if they make sense:

source.obs_ids

[np.uint64(170720)]

for obs_id, ob in source.observation_blocks.items():
    print(ob)
ob

{'actual_duration': <Quantity 1872. s>,
 'actual_start_time': <Time object: scale='tai' format='mjd' value=0.0>,
 'obs_id': np.uint64(170720),
 'producer_id': 'HESS',
 'sb_id': np.uint64(170720),
 'scheduled_duration': <Quantity nan min>,
 'scheduled_start_time': <Time object: scale='tai' format='mjd' value=0.0>,
 'state': <ObservationBlockState.UNKNOWN: -1>,
 'subarray_pointing_frame': <CoordinateFrameType.UNKNOWN: -1>,
 'subarray_pointing_lat': <Quantity nan deg>,
 'subarray_pointing_lon': <Quantity nan deg>}

ctapipe.containers.ObservationBlockContainer:
                        obs_id: Observation Block ID with default
                                18446744073709551615
                         sb_id: ID of the parent SchedulingBlock with default
                                18446744073709551615 with type <class
                                'numpy.uint64'>
                   producer_id: Origin of the obs_id, i.e. name of the telescope
                                site or 'simulation' with default unknown with
                                type <class 'str'>
                         state: State of this OB with default
                                ObservationBlockState.UNKNOWN with type <enum
                                'ObservationBlockState'>
         subarray_pointing_lat: latitude of the nominal center coordinate of
                                this observation with default nan deg [deg]
         subarray_pointing_lon: longitude of the nominal center coordinate of
                                this observation with default nan deg [deg]
       subarray_pointing_frame: Frame in which the subarray_target is non-
                                moving. If the frame is ALTAZ, the meaning of
                                (lon,lat) is (azimuth, altitude) while for ICRS
                                it is (right-ascension, declination) with
                                default CoordinateFrameType.UNKNOWN with type
                                <enum 'CoordinateFrameType'>
            scheduled_duration: expected duration from scheduler with default
                                nan min [min]
          scheduled_start_time: expected start time from scheduler with default
                                0.0
             actual_start_time: true start time with default 0.0
               actual_duration: true duration with default nan min [min]

source.atmosphere_density_profile

source.metadata

{'is_simulation': False}

source.subarray

SubarrayDescription(name='HESS', n_tels=4)

source.subarray.info()

Subarray : HESS
Num Tels : 4
Footprint: 0.01 km2
Height   : 1800.00 m
Lon/Lat  : 16.5051989 deg, -23.2771843 deg

    Type     Count Tel IDs
------------ ----- -------
MST_HESS1_1U     4 1-4

source.subarray.peek()

<ctapipe.visualization.mpl_array.ArrayDisplay at 0x7f36ca8d2e40>

Look at the images

first let’s see what telescopes have data in this event:

event.dl1.tel.keys()

dict_keys([1, 2, 3, 4])

Ok, then we can look at tel_id=1 (which is “CT1”)

The repr of a Container class gives you its contents:

event.dl1.tel[1]

ctapipe.containers.DL1CameraContainer:
                         image: Numpy array of camera image, after waveform
                                extraction.Shape: (n_pixel) if n_channels is 1
                                or data is gain selectedelse: (n_channels,
                                n_pixel) with default None
                     peak_time: Numpy array containing position of the peak of
                                the pulse as determined by the extractor.Shape:
                                (n_pixel) if n_channels is 1 or data is gain
                                selectedelse: (n_channels, n_pixel) with default
                                None
                    image_mask: Boolean numpy array where True means the pixel
                                has passed cleaning. Shape: (n_pixel, ) with
                                default None as a 1-D array with dtype bool
                      is_valid: True if image extraction succeeded, False if
                                failed or in the case of TwoPass methods, that
                                the first pass only was returned. with default
                                False
                    parameters: Image parameters with default None with type
                                <class
                                'ctapipe.containers.ImageParametersContainer'>

So, then you can look at elements:

event.dl1.tel[1].image.sum()

np.float32(0.0)

Display an image to see if things look ok:

tel_id = 1
image = event.dl1.tel[tel_id].image

# best to get the geometry and transform
# it to the EngineeringCameraFrame, which is what you would see if
# looking at the front of the camera
geometry = source.subarray.tel[1].camera.geometry.transform_to(EngineeringCameraFrame())

disp = CameraDisplay(geometry, image=image)