Skip to content

File Handling — .photons / D7

The photonscore.vault module exposes Photonscore data files through a tree of named datasets. Open a file as a :class:Vault, then read arrays through NumPy-style slicing on :attr:Vault.data.

from photonscore import Vault

v = Vault("measurement.photons")
print(v)                        # tree summary

# Coordinates of all photons
x = v.data.photons.x[:]
y = v.data.photons.y[:]
dt = v.data.photons.dt[:]

Vault

vault.Vault

Open or create a Photonscore data file.

A Vault is the entry point for .photons (D7) files: it exposes the file's datasets as a tree under :attr:data, allows reading attributes, and supports creating new datasets when opened in read/write mode.

Attributes:

Name Type Description
data Any

Tree of :class:DataGroup / :class:Dataset mirroring the on-disk layout. v.data.photons.x[:] returns the full photons/x dataset as a NumPy array.
Source code in vault.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
class Vault:
  """Open or create a Photonscore data file.

  A `Vault` is the entry point for `.photons` (D7) files: it exposes
  the file's datasets as a tree under :attr:`data`, allows reading
  attributes, and supports creating new datasets when opened in
  read/write mode.

  Attributes:
      data: Tree of :class:`DataGroup` / :class:`Dataset` mirroring the
          on-disk layout. ``v.data.photons.x[:]`` returns the full
          ``photons/x`` dataset as a NumPy array.
  """

  _impl: Any
  data: Any

  def __init__(
    self,
    path: str,
    fail_if_exists: bool = False,
    create_if_missing: bool = False,
    read_only: bool = True,
    backend: str = "",
    backend_options: str = ""
  ):
    """Open `path` as a Vault.

    Args:
        path: Path to the file. `.photons` extension auto-selects the
            D7 backend; the backend is also tried as a recovery
            fallback if the file is corrupt and opened read-only.
        fail_if_exists: When creating, raise if the file already
            exists.
        create_if_missing: Create the file if it does not exist.
            Implies write access.
        read_only: Open the file read-only. Set to ``False`` to append
            data or create new datasets.
        backend: Override the storage backend. Leave empty for the
            default selection.
        backend_options: Backend-specific options. For the D7 backend
            ``"recover full"`` attempts full recovery of a corrupted
            file.
    """
    is_dot_photons = path.lower().endswith(".photons")
    if is_dot_photons and len(backend) == 0:
      with open(path, "br") as f:
        s = f.read(1000)
        if "D7 Photons Data" in str(s):
          backend = "seven"
    self._impl = _Vault()
    self._path = path
    try:
      self._impl.open(
        path,
        fail_if_exists = fail_if_exists,
        create_if_missing = create_if_missing,
        read_only = read_only,
        backend = backend,
        backend_options = backend_options
      )
    except:
      if not is_dot_photons:
        raise
      if not read_only:
        raise
      # Retry openning
      self._impl.open(
        path,
        fail_if_exists = fail_if_exists,
        create_if_missing = create_if_missing,
        read_only = read_only,
        backend = "seven",
        backend_options = "recover full",
      )
    self.data = DataGroup()
    dss = self._impl.datasets()
    self._datasets = {}
    for path in dss:
      ds = Dataset(self, path)
      self._datasets[path] = ds
      self.data._add(path, ds)

  def __getitem__(self, key: str):
    a = self.attributes
    if key in a:
      return a[key]
    if self.data is None:
      raise Exception("Closed")
    return self.data[key]

  def __repr__(self):
    return "\n".join(
      [
        "[{0}]".format(self._path),
        "Datasets:\n" + "\n".join(
          [
            "  {0:20} {2:8} {1}".format(
              i, self._datasets[i].shape, "{0}".format(self._datasets[i].dtype)
            ) for i in self._datasets.keys()
          ]
        ),
        "Attributes:\n" + "\n".join(
          ["  {0:20} {1}".format(k, v) for k, v in self.attributes.items()]
        ),
      ]
    )

  @property
  def attributes(self):
    return self._impl.attributes()

  def set_attribute(self, name: str, value: str):
    self._impl.set_attribute(name, value)

  def create_dataset(self, name: str, dtype: dtype, shape = ()):
    if self.data is None:
      raise Exception("Closed")
    self._impl.create_dataset(name, dtype, shape)
    ds = Dataset(self, name)
    self._datasets[name] = ds
    self.data._add(name, ds)

  def close(self):
    """Close the underlying file. Subsequent reads raise."""
    if self._impl is not None:
      self._impl.close()
      self._impl = None
      self.data = None

  def read(self, dataset: str, start: int, count: int):
    """Read `count` elements from `dataset` starting at `start`.

    Prefer slicing through :attr:`data` (``v.data.photons.x[0:100]``);
    this method is the low-level primitive used by :class:`Dataset`.
    """
    return self._impl.read(dataset, start, count)

__init__(path, fail_if_exists=False, create_if_missing=False, read_only=True, backend='', backend_options='')

Open path as a Vault.

Parameters:

Name Type Description Default
path str

Path to the file. .photons extension auto-selects the D7 backend; the backend is also tried as a recovery fallback if the file is corrupt and opened read-only.

 required
fail_if_exists bool

When creating, raise if the file already exists.

 False
create_if_missing bool

Create the file if it does not exist. Implies write access.

 False
read_only bool

Open the file read-only. Set to False to append data or create new datasets.

 True
backend str

Override the storage backend. Leave empty for the default selection.

 ''
backend_options str

Backend-specific options. For the D7 backend "recover full" attempts full recovery of a corrupted file.

 ''
Source code in vault.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def __init__(
  self,
  path: str,
  fail_if_exists: bool = False,
  create_if_missing: bool = False,
  read_only: bool = True,
  backend: str = "",
  backend_options: str = ""
):
  """Open `path` as a Vault.

  Args:
      path: Path to the file. `.photons` extension auto-selects the
          D7 backend; the backend is also tried as a recovery
          fallback if the file is corrupt and opened read-only.
      fail_if_exists: When creating, raise if the file already
          exists.
      create_if_missing: Create the file if it does not exist.
          Implies write access.
      read_only: Open the file read-only. Set to ``False`` to append
          data or create new datasets.
      backend: Override the storage backend. Leave empty for the
          default selection.
      backend_options: Backend-specific options. For the D7 backend
          ``"recover full"`` attempts full recovery of a corrupted
          file.
  """
  is_dot_photons = path.lower().endswith(".photons")
  if is_dot_photons and len(backend) == 0:
    with open(path, "br") as f:
      s = f.read(1000)
      if "D7 Photons Data" in str(s):
        backend = "seven"
  self._impl = _Vault()
  self._path = path
  try:
    self._impl.open(
      path,
      fail_if_exists = fail_if_exists,
      create_if_missing = create_if_missing,
      read_only = read_only,
      backend = backend,
      backend_options = backend_options
    )
  except:
    if not is_dot_photons:
      raise
    if not read_only:
      raise
    # Retry openning
    self._impl.open(
      path,
      fail_if_exists = fail_if_exists,
      create_if_missing = create_if_missing,
      read_only = read_only,
      backend = "seven",
      backend_options = "recover full",
    )
  self.data = DataGroup()
  dss = self._impl.datasets()
  self._datasets = {}
  for path in dss:
    ds = Dataset(self, path)
    self._datasets[path] = ds
    self.data._add(path, ds)

close()

Close the underlying file. Subsequent reads raise.

Source code in vault.py 
249
250
251
252
253
254
def close(self):
  """Close the underlying file. Subsequent reads raise."""
  if self._impl is not None:
    self._impl.close()
    self._impl = None
    self.data = None

read(dataset, start, count)

Read count elements from dataset starting at start.

Prefer slicing through :attr:data (v.data.photons.x[0:100]); this method is the low-level primitive used by :class:Dataset.

Source code in vault.py 
256
257
258
259
260
261
262
def read(self, dataset: str, start: int, count: int):
  """Read `count` elements from `dataset` starting at `start`.

  Prefer slicing through :attr:`data` (``v.data.photons.x[0:100]``);
  this method is the low-level primitive used by :class:`Dataset`.
  """
  return self._impl.read(dataset, start, count)

Dataset

vault.Dataset

A single dataset inside a :class:Vault.

Supports NumPy-style slicing — ds[0:100], ds[::2], ds[(0, 1, 2)] — and an :meth:append operation for files opened in read/write mode.

Source code in vault.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
class Dataset:
  """A single dataset inside a :class:`Vault`.

  Supports NumPy-style slicing — ``ds[0:100]``, ``ds[::2]``,
  ``ds[(0, 1, 2)]`` — and an :meth:`append` operation for files opened
  in read/write mode.
  """

  def __init__(self, vault, name):
    self._vault = vault
    self._name = name

  def __getitem__(self, key):
    if isinstance(key, slice):
      return self._read(key.start, key.stop, key.step)
    elif isinstance(key, int):
      return self[key:key + 1]
    elif isinstance(key, tuple):
      return concatenate([self[i] for i in key])
    else:
      raise Exception("Unsupported slicing")

  @property
  def name(self):
    return self._name

  @property
  def dtype(self):
    return self._vault._impl.datasets(self._name)["dtype"]

  @property
  def shape(self):
    return self._vault._impl.datasets(self._name)["shape"]

  def _read(self, start, stop, step):
    shape = self.shape
    if start is None:
      start = 0
    if start < 0:
      start = shape[0] + start
    start = min(shape[0], start)
    if stop is None:
      stop = shape[0] - start
    if stop < 0:
      stop = shape[0] + stop
    count = abs(stop - start)
    data = self._vault.read(self._name, start, count)
    if step is None:
      step = 1
    return data[::step]

  def append(self, data):
    self._vault._impl.append(self._name, data)

DataGroup

vault.DataGroup

A named group of datasets and sub-groups.

Groups are built automatically by :class:Vault so that paths such as "photons/x" become attribute access like v.data.photons.x.

Source code in vault.py 
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
class DataGroup:
  """A named group of datasets and sub-groups.

  Groups are built automatically by :class:`Vault` so that paths such
  as ``"photons/x"`` become attribute access like ``v.data.photons.x``.
  """

  def __init__(self):
    self._items = {}
    self._groups = {}

  def _add(self, path, item):
    pp = _parse_path(path)
    group = self
    item_path = pp[:-1]
    item_name = pp[-1]
    for group_name in item_path:
      if group_name in group._items:
        raise Exception("Item already exists")
      if group_name in group._groups:
        group = group._groups[group_name]
      else:
        sub_group = DataGroup()
        group.__setattr__(group_name, sub_group)
        group._groups[group_name] = sub_group
        group = sub_group
    group.__setattr__(item_name, item)
    group._items[item_name] = item

  def __getitem__(self, name: str):
    is_item = name in self._items
    if is_item:
      return self._items[name]
    return self._groups[name]