Skip to content

Readers

Lower-level routines for loading openPMD-beamphysics particle data.

Particles

load_bunch_data

load_bunch_data(h5: Group, include_time_offset: bool = True) -> dict

Load particles from the only species in this iteration of an OpenPMD BeamPhysics file into a dict of numpy arrays. Raises if more than one or no species.

Parameters:

  • h5 (Group) –

    Particle group, one iteration holding either a single species subgroup or the records themselves (legacy).

  • include_time_offset (bool, default: True ) –

    Add the "timeOffset" record to t. The position and momentum offsets are always included. Default is True.

Returns:

  • dict

    See beamphysics.readers.load_species_data.

Source code in beamphysics/readers.py
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
def load_bunch_data(h5: Group, include_time_offset: bool = True) -> dict:
    """
    Load particles from the only species in this iteration of an OpenPMD BeamPhysics file into a dict of numpy arrays.
    Raises if more than one or no species.

    Parameters
    ----------
    h5 : h5py.Group
        Particle group, one iteration holding either a single species subgroup or the records
        themselves (legacy).
    include_time_offset : bool, optional
        Add the "timeOffset" record to `t`. The position and momentum offsets
        are always included. Default is True.

    Returns
    -------
    dict
        See `beamphysics.readers.load_species_data`.
    """
    return load_species_data(
        _only_species_group(h5), include_time_offset=include_time_offset
    )

load_species_data

load_species_data(h5: Group, include_time_offset: bool = True) -> dict

Load a single species into a dict of numpy arrays.

Parameters:

  • h5 (Group) –

    Group holding the particle records.

  • include_time_offset (bool, default: True ) –

    Add the "timeOffset" record to t. The position and momentum offsets are always included. Default is True.

Returns:

  • dict

    Keys 'x', 'px', 'y', 'py', 'z', 'pz', 't', 'status', 'weight' (arrays), 'species' (str), 'total_charge' (float), and optionally 'id' (array).

Source code in beamphysics/readers.py
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
def load_species_data(h5: Group, include_time_offset: bool = True) -> dict:
    """
    Load a single species into a dict of numpy arrays.

    Parameters
    ----------
    h5 : h5py.Group
        Group holding the particle records.
    include_time_offset : bool, optional
        Add the "timeOffset" record to `t`. The position and momentum offsets
        are always included. Default is True.

    Returns
    -------
    dict
        Keys 'x', 'px', 'y', 'py', 'z', 'pz', 't', 'status', 'weight' (arrays),
        'species' (str), 'total_charge' (float), and optionally 'id' (array).
    """
    attrs = dict(h5.attrs)
    data = {}

    species_type = attrs["speciesType"]
    data["species"] = (
        species_type.decode() if isinstance(species_type, bytes) else species_type
    )

    n_particle = int(_scalar_maybe_from_array(attrs["numParticles"]))

    data["total_charge"] = attrs["totalCharge"] * attrs["chargeUnitSI"]

    for key in ["x", "px", "y", "py", "z", "pz"]:
        data[key] = particle_array(h5, key)
    data["t"] = particle_array(h5, "t", include_offset=include_time_offset)

    if "particleStatus" in h5:
        data["status"] = particle_array(h5, "particleStatus")
    else:
        data["status"] = np.full(n_particle, 1)

    # Make sure weight is populated
    if "weight" in h5:
        weight = particle_array(h5, "weight")
        if len(weight) == 1:
            weight = np.full(n_particle, weight[0])
    else:
        weight = np.full(n_particle, data["total_charge"] / n_particle)
    data["weight"] = weight

    # id should be a unique integer, no units
    # optional
    if "id" in h5:
        data["id"] = h5["id"][:]

    return data

Time offsets

load_only_time_offset

load_only_time_offset(h5: str | Path | File | Group) -> float | ndarray

Load the time offset of the only species of the only iteration.

Parameters:

  • h5 (str, pathlib.Path, h5py.File, or h5py.Group) –

    Filename of an openPMD file, or an open handle. A handle carrying the openPMD attributes is resolved to its single iteration; one that does not is taken to be the particle group itself.

Returns:

  • float or ndarray

    See load_time_offset.

Source code in beamphysics/readers.py
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
def load_only_time_offset(h5: str | pathlib.Path | File | Group) -> float | np.ndarray:
    """
    Load the time offset of the only species of the only iteration.

    Parameters
    ----------
    h5 : str, pathlib.Path, h5py.File, or h5py.Group
        Filename of an openPMD file, or an open handle. A handle carrying the
        openPMD attributes is resolved to its single iteration; one that does
        not is taken to be the particle group itself.

    Returns
    -------
    float or numpy.ndarray
        See `load_time_offset`.
    """
    with _only_iteration_only_species_group(h5) as group:
        return load_time_offset(group)

load_time_offset

load_time_offset(h5: Group) -> float | ndarray

Load the time offset of a single species.

Parameters:

  • h5 (Group) –

    Group holding the particle records.

Returns:

  • float or ndarray

    Offset in seconds: a float for a constant component, an array of length n_particle for a per-particle component, and 0.0 when the group has no "timeOffset" record.

Source code in beamphysics/readers.py
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
def load_time_offset(h5: Group) -> float | np.ndarray:
    """
    Load the time offset of a single species.

    Parameters
    ----------
    h5 : h5py.Group
        Group holding the particle records.

    Returns
    -------
    float or numpy.ndarray
        Offset in seconds: a float for a constant component, an array of length
        n_particle for a per-particle component, and 0.0 when the group has no
        "timeOffset" record.
    """
    return component_scalar_or_array_data(h5, "timeOffset", default=0.0)

particle_paths

particle_paths(h5, key='particlesPath')

Uses the basePath and particlesPath to find where openPMD particles should be

Source code in beamphysics/readers.py
174
175
176
177
178
179
180
181
182
183
184
185
186
187
def particle_paths(h5, key="particlesPath"):
    """
    Uses the basePath and particlesPath to find where openPMD particles should be

    """
    basePath = h5.attrs["basePath"].decode("utf-8")
    particlesPath = h5.attrs[key].decode("utf-8")

    if "%T" not in basePath:
        return [basePath + particlesPath]
    path1, path2 = basePath.split("%T")
    tlist = list(h5[path1])
    paths = [path1 + t + path2 + particlesPath for t in tlist]
    return paths

particle_array

particle_array(h5, component, slice=slice(None), include_offset=True)

Main routine to return particle arrays in fixed units. All units are SI except momentum, which will be in eV/c.

Example: particle_array(h5['data/00001/particles/'], 'px') Will return the momentum/x + momentumOffset/x in eV/c.

Source code in beamphysics/readers.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
def particle_array(h5, component, slice=slice(None), include_offset=True):
    """
    Main routine to return particle arrays in fixed units.
    All units are SI except momentum, which will be in eV/c.

    Example:
        particle_array(h5['data/00001/particles/'], 'px')
        Will return the momentum/x + momentumOffset/x in eV/c.


    """

    # Handle aliases
    if component in component_from_alias:
        component = component_from_alias[component]

    if component in ["momentum/x", "momentum/y", "momentum/z"]:
        unit_factor = c_light / e_charge  # convert J/(m/s) to eV/c
    else:
        unit_factor = 1.0

    # Get data
    dat = component_data(h5[component], slice=slice, unit_factor=unit_factor)

    # Look for offset component
    ocomponent = offset_component_name(component)
    if include_offset and ocomponent in h5:
        offset = component_data(h5[ocomponent], slice=slice, unit_factor=unit_factor)
        dat += offset

    return dat