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
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
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
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
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
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
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
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
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
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
201
202
203
204
205
206
207
208
209
210
211
212
213
214
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
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
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