Spool

Spools are containers/managers of patches. The spool interface is designed to manage a variety of data sources, including a group of patches loaded into memory, archives of local files, and a variety of remote resources.

For file-backed spools, DASCore first scans metadata and coordinate summaries, then loads patch data only when you access a patch from the spool. This means methods like get_contents() work from patch summary metadata, while spool[0] or iteration materializes loaded patches.

Data Sources

The simplest way to get the appropriate spool for a specified input is to use the spool function, which knows about many different input types and returns a Spool instance.

Patches (in-memory)

import dascore as dc

patch_list = [dc.get_example_patch()]

spool1 = dc.spool(patch_list)

A Single file

import dascore as dc
from upath import UPath
# Import fetch to read DASCore example files 
from dascore.utils.downloader import fetch

path_to_das_file = fetch("terra15_das_1_trimmed.hdf5")
# To read DAS data stored locally on your machine, simply replace the above line with:
# path_to_das_file = "/path/to/data/directory/data.EXT"

spool2 = dc.spool(path_to_das_file)

# UPath-backed resources also work for file-backed spools.
remote_file = UPath("memory://dascore/tutorial/spool_file.h5")
dc.write(dc.get_example_patch(), remote_file, "DASDAE")
spool2_remote = dc.spool(remote_file)

A directory of DAS files

import dascore as dc
# Import fetch to read DASCore example files 
from dascore.utils.downloader import fetch

# Fetch a sample file path from DASCore (just to get a usable path for the rest of the cell)
directory_path = fetch('terra15_das_1_trimmed.hdf5').parent
# To read a directory of DAS data stored locally on your machine, 
# simply replace the above line with:
# directory_path = "/path/to/data/directory/"

# Update will create an index of the contents for fast querying/access
spool3 = dc.spool(directory_path).update()
/home/runner/work/dascore/dascore/dascore/io/core.py:1638: UserWarning: 
/home/runner/work/dascore/dascore/.test_data_cache/0.0.0/das_vader_1.jld2 is a legacy DASVader JLD2 file with 
anonymous object references in 'htime'. This file class is not supported by the current HDF5 stack (h5py 3.16.0 / 
HDF5 2.0.0). Install a compatibility stack such as h5py<3.16 with HDF5 1.14.x, then retry.
  for result, source_info in iterator:

/home/runner/work/dascore/dascore/dascore/io/core.py:1489: UserWarning: Failed to scan 
<dascore.utils.hdf5._ManagedH5pyFile object at 0x7f8c8fc658b0>
  warnings.warn(f"Failed to scan {resource}", UserWarning)

/home/runner/work/dascore/dascore/dascore/io/core.py:1489: UserWarning: Failed to scan 
<dascore.utils.hdf5._ManagedH5pyFile object at 0x7f8c8f8cb650>
  warnings.warn(f"Failed to scan {resource}", UserWarning)

/home/runner/work/dascore/dascore/dascore/io/core.py:1489: UserWarning: Failed to scan 
<dascore.utils.hdf5._ManagedH5pyFile object at 0x7f8c8fa7e0f0>
  warnings.warn(f"Failed to scan {resource}", UserWarning)

/home/runner/work/dascore/dascore/dascore/io/core.py:1489: UserWarning: Failed to scan 
<dascore.utils.hdf5._ManagedH5pyFile object at 0x7f8c8cf43710>
  warnings.warn(f"Failed to scan {resource}", UserWarning)

/home/runner/work/dascore/dascore/dascore/io/core.py:1489: UserWarning: Failed to scan 
<dascore.utils.hdf5._ManagedH5pyFile object at 0x7f8c8cf438f0>
  warnings.warn(f"Failed to scan {resource}", UserWarning)







If you want the index file to exist somewhere else, for example if you can’t write to the data directory, you can specify an index path.




import tempfile
from pathlib import Path

index_path = Path(tempfile.mkdtemp()) / "index.h5"

# Update will create an index of the contents for fast querying/access.
spool = dc.spool(directory_path, index_path=index_path).update()




A new spool created for the same directory will find this index file automatically. The exception is when a valid index file already exists inside the data directory itself, which takes precedence.












Warning







If you remove files from a directory that has already been indexed, you should delete the index file and then call update again on the spool like this:


spool.indexer.index_path.unlink()
spool.update()


It is best not to delete files once added to a directory managed by DASCore.








Path attributes (hive-style directories)


Directory spools parse key=value pairs out of the paths inside the spool, in the style of Hive partitioning. Each directory segment can hold one pair (acquisition_key=XX.R2D1..RAW/), and any segment — including the file name, whose extension is ignored — can hold several separated by __ (the same separator DASCore’s default patch names use). The parsed values become string attributes: they show up in get_contents(), work with select, and are set on loaded patches. When a path attribute and an attribute stored inside the file share a name, the path wins — renaming a directory (or file) is how you attach or correct metadata without rewriting data.




import tempfile
from pathlib import Path

import dascore as dc

# Build a small hive-partitioned directory of DAS files.
data_path = Path(tempfile.mkdtemp()) / "experiment=demo" / "cable=north"
data_path.mkdir(parents=True)
dc.get_example_patch().io.write(data_path / "tag=raw.h5", "DASDAE")

spool = dc.spool(data_path.parent.parent).update()
df = spool.get_contents()
assert df["experiment"].iloc[0] == "demo"
assert len(spool.select(cable="north")) == 1

# The loaded patch carries the path attributes too.
patch = spool[0]
assert patch.attrs["experiment"] == "demo"
assert patch.attrs.tag == "raw"




Renaming a partition directory and calling update() refreshes the affected attributes by rewriting index rows — file contents are not re-read, so this stays fast even for very large directories.




root = data_path.parent.parent
(root / "experiment=demo").rename(root / "experiment=final")
spool = spool.update()
assert spool.get_contents()["experiment"].iloc[0] == "final"




A few details worth knowing:


    

  • Values are always strings; select supports equality, collections, unix-style globs, and regular expressions on them.
    • 

  • Removing a key=value pair from a path triggers a rescan of the affected files (the file’s own value has to be recovered), while adding or changing pairs does not.
    • 

  • A path key that matches a coordinate name (e.g. distance=...) resolves attrs-first in bare select kwargs; use the _coords argument to target the coordinate explicitly.
    • 



However they were constructed, all spools are the same class and share the same behavior and methods.












Note







Remote file resources work with dc.spool(...), dc.read(...), dc.scan(...), and dc.scan_payloads(...).
















Note







For remote UPath resources, DASCore treats metadata operations differently from full reads. See the Working with Remote Patches tutorial for the remote-cache policy, spool implications, and examples.














Accessing patches


Patches are extracted from the spool via simple iteration or indexing. New spools are returned via slicing.




import dascore as dc

spool = dc.get_example_spool()

# Extract first patch in the spool.
patch = spool[0]

# Iterate patches in spool.
for patch in spool:
    ...

# Slice spool to create new spool which excludes first patch.
new_spool = spool[1:]




For file-backed spools, indexing or iteration is the point where DASCore loads the underlying patch data. Metadata-only operations such as get_contents() and select(...) can usually run without loading full patch arrays.


An array can also be used (just like numpy) to select/re-arrange spool contents. For example, a boolean array can be used to de-select patches:




import dascore as dc
import numpy as np

spool = dc.get_example_spool()

# Get bool array, true values indicate patch is kept, false is discarded.
bool_array = np.ones(len(spool), dtype=np.bool_)
bool_array[1] = False

# Remove patch at position 1 from spool.
new = spool[bool_array]




and an integer array can be used to deselect/rearrange patches




import dascore as dc
import numpy as np

spool = dc.get_example_spool()

# Get an array of integers which indicate the index of included patches
index_array = np.array([2, 0])

# create a new spool with patch 2 and patch 0.
new = spool[index_array]








get_contents


The get_contents method returns a dataframe listing the spool contents. It realizes the whole relation, so it can be expensive over a large remote archive.




import dascore as dc
spool = dc.get_example_spool()

# Return dataframe with contents of spool (each row has metadata of a patch)
contents = spool.get_contents()

















source_path
source_format
source_version
source_patch_id
dims
time_min
time_max
time_step
distance_min
distance_max
distance_step
distance_units
time_units
tag
category






0
memorypatch://f4d4576ef0e044cea2fb962cce2fc182
memory


distance,time
2020-01-03 00:00:00
2020-01-03 00:00:07.996
0 days 00:00:00.004000
0.0
299.0
1.0
m
s
random
DAS




1
memorypatch://84c5f4d3579c42bc91a05e76275f71fd
memory


distance,time
2020-01-03 00:00:08
2020-01-03 00:00:15.996
0 days 00:00:00.004000
0.0
299.0
1.0
m
s
random
DAS




2
memorypatch://d151be1c56a84939a9c02ab916e5a70e
memory


distance,time
2020-01-03 00:00:16
2020-01-03 00:00:23.996
0 days 00:00:00.004000
0.0
299.0
1.0
m
s
random
DAS













The columns returned by get_contents() come from the same patch summary metadata exposed by Patch.summary, so fields such as time_min, time_max, and distance_step are available without loading the underlying patch data. Source metadata such as source_path, source_format, and source_patch_id are also available for file-backed spools.






select


The select method selects a subset of a spool and returns a new spool. get_contents will now reflect a subset of the original data requested by the select operation.




import dascore as dc
spool = dc.get_example_spool()

# Select a spool with data after Jan 3rd, 2020.
subspool = spool.select(time=('2020-01-03T00:00:09', ...))




In addition to trimming the data along a specified dimension (as shown above), select can be used to filter patches that meet a specified criteria.




import dascore as dc
# Load a spool which has many diverse patches.
spool = dc.get_example_spool('diverse_das')

# Only include patches from two named data sources.
subspool = spool.select(acquisition_key={'DAS2.R2D1..RAW', 'DAS3.R2D1..RAW'})

# Only include spools which match some unix-style query on their tags.
subspool = spool.select(tag='some*')








unselect


unselect is the complement of select: it returns the patches the same selection would have removed. Each keyword means what it means in select, so this is how to name what a spool does not want without spelling out the rest of the archive.




import dascore as dc
spool = dc.get_example_spool('diverse_das')

# Everything except the patches tagged 'some_tag'.
subspool = spool.unselect(tag='some_tag')
assert len(subspool) + len(spool.select(tag='some_tag')) == len(spool)




The patches’ own coordinates are not accepted. Selecting on one trims each patch to the range as well as dropping the patches which miss it entirely, so the complement of time=(t1, t2) is the part of every patch before t1 together with the part after t2 — one patch becoming two. That is subdivision rather than filtering; select the ranges to keep instead, or use Patch.unselect on each patch.


The coordinates an attached inventory defines along the fiber are different, and are accepted: removing one of those chooses which channels a patch holds. A patch may then be cut into the pieces the query did not match, so len can grow here as well.






chunk


The chunk method controls how data are grouped together in patches within the spool. It can be used to merge contiguous patches together, specify the size of patches for processing, specify overlap with previous patches, etc.




import dascore as dc
spool = dc.get_example_spool()

# Chunk spool for 3 s increments with 1 s overlaps
# and keep any segments at the end that don't have the full 3 s.
subspool = spool.chunk(time=3, overlap=1, keep_partial=True)

# Merge all contiguous segments along time dimension.
merged_spool = spool.chunk(time=None)




The chunk length can also be a quantity, either in the coordinate’s own units or as a data size. A data size chunks so that each patch’s data array is about that large, which is the usual way to keep patches inside a memory budget without working out the sample arithmetic by hand.




from dascore.units import s, megabytes

# The same 3 second chunk, with the units stated explicitly.
unit_chunked = spool.chunk(time=3 * s)

# Chunk so each patch's data array is at most ~1 MB.
size_chunked = spool.chunk(time=1 * megabytes)




A data size measures the data array only; coordinates, attrs, and any copies made later during processing are extra, so the patch as a whole is somewhat larger. The sample count is rounded down, so the patch’s data never exceeds the requested size, and MB is 106 bytes while MiB is 220. overlap accepts the same forms.






concatenate


Similar to chunk, Spool.concatenate is used to combine patches together. However, concatenate doesn’t account for coordinate values along the concatenation axis, and can even be used to create new patch dimensions. Like chunk, it produces a lazy, plan-backed result.


import dascore as dc

patch = dc.get_example_patch()

# Create a spool with patches that have a large gap
time = patch.get_coord("time")
one_hour = dc.to_timedelta64(3600)
patch2 = patch.update_coords(time_min=time.max() + one_hour)
spool = dc.spool([patch, patch2])

# chunk rightfully wouldn't merge these patches, but concatenate will.
merged = spool.concatenate(time=None)
print(merged[0].coords)






map


The map method applies a function to all patches in the spool. It provides an efficient way to process large datasets, especially when combined with clients (aka executors).


For example, calculating the maximum value for each channel (distance) for 5 second increments with 1 second overlap can be done like so:




import dascore as dc
spool = dc.get_example_spool()

# define function for mapping to each patch
def get_dist_max(patch):
    """Function which will be mapped to each patch in spool."""
    return patch.aggregate("time", "max")

# chunk and apply function
map_out = spool.chunk(time=5, overlap=1).map(get_dist_max)

# combine output back into a single patch
agg_patch = dc.spool(map_out).concatenate(time=None)[0]

print(agg_patch)















DASCore Patch ⚡
---------------
➤ Coordinates (distance: 300, time: 5)
    *distance: CoordRange( min: 0 max: 299 step: 1 shape: (300,) dtype: int64 units: m )
    *time: CoordPartial( shape: (5,) dtype: None )
➤ Data (float64)
   [[1.    1.    1.    1.    1.   ]
    [1.    1.    1.    1.    1.   ]
    [0.993 1.    0.993 1.    0.993]
    ...
    [1.    1.    1.    1.    1.   ]
    [0.999 0.999 0.999 0.999 0.999]
    [1.    0.998 1.    0.998 1.   ]]
➤ Attributes
    tag: random
    history: ("aggregate(dim='time',dim_reduce='empty',method='max')", 'drop_private_coords()', 'concatenate')
    category: DAS







See the parallel processing recipe for more examples with map.






Inventory


A spool can carry a DASDAE inventory: a description of the observing system its data was recorded through — the fiber, where it goes, and how the interrogator was configured over time. Attaching one adds nothing to the patches, but makes the names the inventory defines along the fiber available to select and expand_by, and lets enrich copy that metadata onto patches as they are extracted.




import dascore as dc
from dascore.examples import inventory_patch_pair

patch, inventory = inventory_patch_pair()
spool = dc.spool(patch).attach_inventory(inventory)

# The inventory annotates two zones along the fiber, so they can be selected.
assert len(spool.expand_by("zone")) == 2




A spool opened on a directory which carries an inventory under the name .inventory — as a directory, or as .inventory.yaml, .inventory.yml, or .inventory.json — starts out attached to it.