Overview

The Metadata class can be used in order to organize lot’s of data stream files. For example you might have a project generated by the Analyzer4D Software and you want to search through all data files in order to find all streams with a certain compression. You could go through each file and check out if the compression matches your requirement which takes a lot of time. The cache creates a database that caches all the metadata available in the qass.tools.analyzer.buffer_parser.Buffer class. You can then query the cache very fast and programmatically in Python!

Example

In this example we create an instance of the cache and synchronize it with the directory my/directory. We then create a template BufferMetadata object and use it with the cache to query for all buffers with a compression_frq = 8. You can use all properties that are in BufferMetadata.properties. The BufferMetadataCache.get_matching_buffers() method returns a list of Buffer objects that are in this case sorted by their process number (as specified in the query).

from qass.tools.analyzer.buffer_metadata_cache import (
    BufferMetadataCache as BMC, 
    BufferMetadata as BM, 
    select
    )


cache = BMC()
cache.synchronize_directory("my/directory")
results = cache.get_matching_buffers(select(BM).filter(BM.compression_frq==8).order_by(BM.process))

qass.tools.analyzer.buffer_metadata_cache

Classes:

Name	Description
BufferMetadata	This class acts as a template for buffer files. It's properties represent all available metadata of a buffer file.
BufferMetadataCache	This class acts as a Cache for Buffer Metadata. It uses a database session with a buffer_metadata table to map

__Base module-attribute

__Base = declarative_base()

__all__ module-attribute

__all__ = ['BufferMetadataCache', 'BufferMetadata']

BufferEnum

Bases: TypeDecorator

Methods:

Name	Description
__init__
process_bind_param
process_result_value

Attributes:

Name	Type	Description
impl

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

class BufferEnum(TypeDecorator):
    impl = String

    def __init__(self, enumtype, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self._enumtype = enumtype

    def process_bind_param(self, value, dialect):
        if value is None:
            return None
        return value.name

    def process_result_value(self, value, dialect):
        if value is None:
            return None
        return self._enumtype[str(value)]

impl class-attribute instance-attribute

impl = String

__init__

__init__(enumtype, *args, **kwargs)

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def __init__(self, enumtype, *args, **kwargs):
    super().__init__(*args, **kwargs)
    self._enumtype = enumtype

process_bind_param

process_bind_param(value, dialect)

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def process_bind_param(self, value, dialect):
    if value is None:
        return None
    return value.name

process_result_value

process_result_value(value, dialect)

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def process_result_value(self, value, dialect):
    if value is None:
        return None
    return self._enumtype[str(value)]

BufferMetadata

Bases: __Base

This class acts as a template for buffer files. It's properties represent all available metadata of a buffer file. This class is used internally as a database model and can be instantiated to provide a template for a buffer file by populating desired properties and passing the object to the cache which will in turn create a query based on this object.

Methods:

Name	Description
buffer_to_metadata	Converts a Buffer object to a BufferMetadata database object by copying all the @properties from the Buffer
filepath

Attributes:

Name	Type	Description
__tablename__
adc_type
analyzer_version
avg_frq
avg_time
bit_resolution
bytes_per_sample
channel
compression_frq
compression_time
datakind
datamode
datatype
db_count
db_header_size
db_sample_count
db_size
db_spec_count
directory_path
fft_log_shift
filename
frq_bands
frq_end
frq_per_band
frq_start
full_blocks
header_hash
header_size
id
machine_id
opening_error
partnumber
preamp_gain
process
process_date_time
process_time
project_id
properties
sample_count
spec_count
spec_duration
streamno

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

class BufferMetadata(__Base):
    """This class acts as a template for buffer files. It's properties represent all available metadata of a buffer file.
    This class is used internally as a database model and can be instantiated to provide a template for a buffer file by
    populating desired properties and passing the object to the cache which will in turn create a query based on this object.
    """

    __tablename__ = "buffer_metadata"
    properties = (
        "id",
        "project_id",
        "directory_path",
        "filename",
        "header_size",
        "process",
        "channel",
        "datamode",
        "datakind",
        "datatype",
        "process_time",
        "process_date_time",
        "db_header_size",
        "bytes_per_sample",
        "db_count",
        "full_blocks",
        "db_size",
        "db_sample_count",
        "frq_bands",
        "db_spec_count",
        "compression_frq",
        "compression_time",
        "avg_time",
        "avg_frq",
        "spec_duration",
        "frq_start",
        "frq_end",
        "frq_per_band",
        "sample_count",
        "spec_count",
        "adc_type",
        "bit_resolution",
        "fft_log_shift",
        "streamno",
        "preamp_gain",
        "analyzer_version",
        "partnumber",
        "header_hash",
    )

    id = Column(Integer, Identity(start=1), primary_key=True)
    project_id = Column(BigInteger, index=True)
    directory_path = Column(String, nullable=False, index=True)
    filename = Column(String, nullable=False)
    header_hash = Column(String(64), index=True)
    machine_id = Column(String, nullable=True)
    header_size = Column(Integer)
    process = Column(Integer, index=True)
    channel = Column(Integer, index=True)
    datamode = Column(
        BufferEnum(Buffer.DATAMODE), index=True
    )  # this is an ENUM in buffer_parser
    datakind = Column(BufferEnum(Buffer.DATAKIND))  # this is an ENUM in buffer_parser
    datatype = Column(BufferEnum(Buffer.DATATYPE))  # this is an ENUM in buffer_parser
    process_time = Column(BigInteger)
    process_date_time = Column(String)
    db_header_size = Column(Integer)
    bytes_per_sample = Column(Integer)
    db_count = Column(Integer)
    full_blocks = Column(Integer)
    db_size = Column(Integer)
    db_sample_count = Column(Integer)
    frq_bands = Column(Integer)
    db_spec_count = Column(Integer)
    compression_frq = Column(Integer, index=True)
    compression_time = Column(Integer, index=True)
    avg_time = Column(Integer, index=True)
    avg_frq = Column(Integer, index=True)
    spec_duration = Column(Float)
    frq_start = Column(Integer)
    frq_end = Column(Integer)
    frq_per_band = Column(Float)
    sample_count = Column(BigInteger)
    spec_count = Column(BigInteger)
    adc_type = Column(
        BufferEnum(Buffer.ADCTYPE)
    )  # TODO this is an ENUM in buffer_parser
    bit_resolution = Column(Integer)
    fft_log_shift = Column(Integer)
    streamno = Column(Integer)
    preamp_gain = Column(Integer)
    analyzer_version = Column(String)
    partnumber = Column(String)

    opening_error = Column(String, nullable=True)

    @hybrid_property
    def filepath(self):
        return str(Path(str(self.directory_path)) / self.filename)

    @staticmethod
    def buffer_to_metadata(buffer):
        """Converts a Buffer object to a BufferMetadata database object by copying all the @properties from the Buffer
        object putting them in the BufferMetadata object

        Parameters
        ----------
        buffer : Buffer

        Returns
        -------
        buffer_metadata: BufferMetadata
        """
        file = Path(buffer.filepath)
        directory_path = str(file.parent)
        buffer_metadata = BufferMetadata(
            filename=file.name, directory_path=directory_path
        )
        for prop in BufferMetadata.properties:
            try:  # try to map all the buffer properties and skip on error
                setattr(
                    buffer_metadata, prop, getattr(buffer, prop)
                )  # get the @property method and execute it
            except Exception:
                continue
        return buffer_metadata

__tablename__ class-attribute instance-attribute

__tablename__ = 'buffer_metadata'

adc_type class-attribute instance-attribute

adc_type = Column(BufferEnum(ADCTYPE))

analyzer_version class-attribute instance-attribute

analyzer_version = Column(String)

avg_frq class-attribute instance-attribute

avg_frq = Column(Integer, index=True)

avg_time class-attribute instance-attribute

avg_time = Column(Integer, index=True)

bit_resolution class-attribute instance-attribute

bit_resolution = Column(Integer)

bytes_per_sample class-attribute instance-attribute

bytes_per_sample = Column(Integer)

channel class-attribute instance-attribute

channel = Column(Integer, index=True)

compression_frq class-attribute instance-attribute

compression_frq = Column(Integer, index=True)

compression_time class-attribute instance-attribute

compression_time = Column(Integer, index=True)

datakind class-attribute instance-attribute

datakind = Column(BufferEnum(DATAKIND))

datamode class-attribute instance-attribute

datamode = Column(BufferEnum(DATAMODE), index=True)

datatype class-attribute instance-attribute

datatype = Column(BufferEnum(DATATYPE))

db_count class-attribute instance-attribute

db_count = Column(Integer)

db_header_size class-attribute instance-attribute

db_header_size = Column(Integer)

db_sample_count class-attribute instance-attribute

db_sample_count = Column(Integer)

db_size class-attribute instance-attribute

db_size = Column(Integer)

db_spec_count class-attribute instance-attribute

db_spec_count = Column(Integer)

directory_path class-attribute instance-attribute

directory_path = Column(String, nullable=False, index=True)

fft_log_shift class-attribute instance-attribute

fft_log_shift = Column(Integer)

filename class-attribute instance-attribute

filename = Column(String, nullable=False)

frq_bands class-attribute instance-attribute

frq_bands = Column(Integer)

frq_end class-attribute instance-attribute

frq_end = Column(Integer)

frq_per_band class-attribute instance-attribute

frq_per_band = Column(Float)

frq_start class-attribute instance-attribute

frq_start = Column(Integer)

full_blocks class-attribute instance-attribute

full_blocks = Column(Integer)

header_hash class-attribute instance-attribute

header_hash = Column(String(64), index=True)

header_size class-attribute instance-attribute

header_size = Column(Integer)

id class-attribute instance-attribute

id = Column(Integer, Identity(start=1), primary_key=True)

machine_id class-attribute instance-attribute

machine_id = Column(String, nullable=True)

opening_error class-attribute instance-attribute

opening_error = Column(String, nullable=True)

partnumber class-attribute instance-attribute

partnumber = Column(String)

preamp_gain class-attribute instance-attribute

preamp_gain = Column(Integer)

process class-attribute instance-attribute

process = Column(Integer, index=True)

process_date_time class-attribute instance-attribute

process_date_time = Column(String)

process_time class-attribute instance-attribute

process_time = Column(BigInteger)

project_id class-attribute instance-attribute

project_id = Column(BigInteger, index=True)

properties class-attribute instance-attribute

properties = ('id', 'project_id', 'directory_path', 'filename', 'header_size', 'process', 'channel', 'datamode', 'datakind', 'datatype', 'process_time', 'process_date_time', 'db_header_size', 'bytes_per_sample', 'db_count', 'full_blocks', 'db_size', 'db_sample_count', 'frq_bands', 'db_spec_count', 'compression_frq', 'compression_time', 'avg_time', 'avg_frq', 'spec_duration', 'frq_start', 'frq_end', 'frq_per_band', 'sample_count', 'spec_count', 'adc_type', 'bit_resolution', 'fft_log_shift', 'streamno', 'preamp_gain', 'analyzer_version', 'partnumber', 'header_hash')

sample_count class-attribute instance-attribute

sample_count = Column(BigInteger)

spec_count class-attribute instance-attribute

spec_count = Column(BigInteger)

spec_duration class-attribute instance-attribute

spec_duration = Column(Float)

streamno class-attribute instance-attribute

streamno = Column(Integer)

buffer_to_metadata staticmethod

buffer_to_metadata(buffer)

Converts a Buffer object to a BufferMetadata database object by copying all the @properties from the Buffer object putting them in the BufferMetadata object

Parameters:

Name	Type	Description	Default
buffer	Buffer		required

Returns:

Name	Type	Description
buffer_metadata	BufferMetadata

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

@staticmethod
def buffer_to_metadata(buffer):
    """Converts a Buffer object to a BufferMetadata database object by copying all the @properties from the Buffer
    object putting them in the BufferMetadata object

    Parameters
    ----------
    buffer : Buffer

    Returns
    -------
    buffer_metadata: BufferMetadata
    """
    file = Path(buffer.filepath)
    directory_path = str(file.parent)
    buffer_metadata = BufferMetadata(
        filename=file.name, directory_path=directory_path
    )
    for prop in BufferMetadata.properties:
        try:  # try to map all the buffer properties and skip on error
            setattr(
                buffer_metadata, prop, getattr(buffer, prop)
            )  # get the @property method and execute it
        except Exception:
            continue
    return buffer_metadata

filepath

filepath()

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

@hybrid_property
def filepath(self):
    return str(Path(str(self.directory_path)) / self.filename)

BufferMetadataCache

This class acts as a Cache for Buffer Metadata. It uses a database session with a buffer_metadata table to map metadata to files on the disk. The cache can be queried a lot faster than manually opening a lot of buffer files.

Methods:

Name	Description
__init__
add_files_to_cache	Add buffer files to the cache by providing the complete filepaths
get_buffer_metadata_query	Converts a BufferMetadata object to a complete query. Every property of the object will be converted into
get_matching_buffers	Calls get_matching_files and converts the result to Buffer objects
get_matching_files	Query the Cache for all files matching the properties that selected by the query object.
get_matching_metadata	Query the cache for all BufferMetadata database entries matching
get_non_synchronized_files	calculate the difference between the set of files and the set of synchronized files
remove_files_from_cache	Remove synchronized files from the cache
synchronize_directory	synchronize the buffer files in the given paths with the database matching the regex pattern

Attributes:

Name	Type	Description
BufferMetadata
Buffer_cls
Session
engine

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

class BufferMetadataCache:
    """This class acts as a Cache for Buffer Metadata. It uses a database session with a buffer_metadata table to map
    metadata to files on the disk. The cache can be queried a lot faster than manually opening a lot of buffer files.
    """

    BufferMetadata = BufferMetadata

    def __init__(self, db_url="sqlite:///:memory:", Buffer_cls=Buffer):
        self.engine = create_engine(db_url)
        BufferMetadata.metadata.create_all(self.engine)
        self.Session = sessionmaker(bind=self.engine)
        self.Buffer_cls = Buffer_cls

    def synchronize_directory(
        self,
        *paths,
        sync_subdirectories=True,
        regex_pattern="^.*[p][0-9]*[c][0-9]{1}[b][0-9]{2}",
        verbose=1,
        delete_stale_entries=False,
        machine_id=None,
        glob_pattern="*p*c?b*",
    ):
        """synchronize the buffer files in the given paths with the database matching the regex pattern

        Parameters
        ----------
        paths: str
            The absolute paths to the directory
        sync_subdirectories: bool, optional
            When True synchronize all of the subdirectories recursively, defaults to True
        regex_pattern: str, optional
            The regex pattern validating the buffer naming format (matched on file.name)
        verbose: int, optional
            verbosity level. 0 = no feedback, 1 = progress bar
        machine_id: str, optional
            An optional identifier for a certain machine to enable synchronization of different platforms
        glob_pattern: str, optional
            The pattern forwarded to Path.glob. This pattern acts as a preselection for the
            files retrieved for the regex pattern
        """
        pattern = re.compile(regex_pattern)
        for path in paths:
            if sync_subdirectories:
                files = [
                    file
                    for file in Path(path).resolve().rglob(glob_pattern)
                    if file.is_file() and pattern.match(file.name)
                ]
            else:
                files = [
                    file
                    for file in Path(path).resolve().glob(glob_pattern)
                    if file.is_file() and pattern.match(file.name)
                ]
            unsynchronized_files, synchronized_missing_buffers = (
                self.get_non_synchronized_files(files, machine_id)
            )
            if delete_stale_entries:
                self.remove_files_from_cache(
                    synchronized_missing_buffers, verbose=verbose
                )
            self.add_files_to_cache(
                unsynchronized_files,
                verbose=verbose,
                machine_id=machine_id,
                check_synced=False,
            )

    def get_matching_metadata(self, query: Select) -> List[BufferMetadata]:
        """Query the cache for all BufferMetadata database entries matching

        Parameters
        ----------
        query: Select
            A sqlalchemy select statement specifying the properties of the BufferMetadata objects

        Returns
        -------
        list[BufferMetadata]
            A list with the matching BufferMetadata objects
        """
        with self.Session() as session:
            matching_metadata = list(session.scalars(query).all())
        return matching_metadata

    def get_matching_files(self, query: Select) -> List[str]:
        """Query the Cache for all files matching the properties that selected by the query object.
        The usage of the buffer_metadata, filter_functions and sort_key is deprecated and will be removed in
        two minor versions. Use the sqlalchemy query parameter instead.

        Example
        -------
        Returns all buffer filepaths with channel = 1, A frequency compression of 4,
        processes above 100 sorted by the process number
        ```py
        BufferMetadataCache.get_matching_files(
            select(BM).filter(BM.channel==1, BM.compression_freq==4, BM.process > 100)
        )
        ```

        Parameters
        ----------
        query: Select
            A sqlalchemy select statement specifying the properties of the BufferMetadata objects

        Returns
        -------
        list[str]
            A list with the paths to the buffer files that match the buffer_metadata
        """
        matching_metadata = self.get_matching_metadata(query)
        return [str(m.filepath) for m in matching_metadata]

    def get_matching_buffers(self, query: Select) -> List[Buffer]:
        """Calls get_matching_files and converts the result to Buffer objects

        Returns
        -------
        list[Buffer]
            List of Buffer objects
        """
        files = self.get_matching_files(query)
        buffers = []
        for file in files:
            try:
                with self.Buffer_cls(file) as buffer:
                    buffers.append(buffer)
            except Exception:
                pass
        return buffers

    def get_non_synchronized_files(
        self,
        files: Sequence[Path],
        machine_id: Union[str, None] = None,
    ) -> Tuple[List[Path], List[Path]]:
        """calculate the difference between the set of files and the set of synchronized files

        Parameters
        ----------
        files: Sequence[Path]
            Sequence of files to check whether they are already in the cache
        machine_id: Union[str, None]
            machine identifier to provide extra context for the location of the files

        Returns
        -------
        tuple[list[Path], list[Path]]
            The set of files that are not synchronized, and the database entries that exist but the file is not present anymore
        """
        file_set = set(files)
        with self.Session() as session:
            synchronized_buffers = set(
                Path(str(buffer.filepath))
                for buffer in session.query(self.BufferMetadata)
                .filter(BufferMetadata.machine_id == machine_id)
                .all()
            )
        unsynchronized_files = [
            Path(p) for p in file_set.difference(synchronized_buffers)
        ]
        synchronized_missing_buffers = [
            Path(p) for p in synchronized_buffers.difference(file_set)
        ]
        return unsynchronized_files, synchronized_missing_buffers

    def add_files_to_cache(
        self,
        files: Sequence[Path],
        verbose: int = 0,
        batch_size: int = 1000,
        machine_id: Union[str, None] = None,
        check_synced: bool = True,
    ):
        """Add buffer files to the cache by providing the complete filepaths
        If a file (determined by filename and directory) is already synchronized it will be skipped

        Parameters
        ----------
        files: Sequence[Path]
            complete filepaths that are added to the cache. The filepath is used with the Buffer class to open a buffer and extract the header information.
        verbose: int, optional
            verbosity level. 0 = no feedback, 1 = progress bar
        batch_size: int, optional
            The batch size after which the cache will commit a batch to the database
        machine_id: str, optional
            A unique identifier for a different machine
        check_synced: bool, optional
            Whether to check if the files that are about to be added are already synced to the cache
        """
        if check_synced:
            files, _ = self.get_non_synchronized_files(files, machine_id)
        with self.Session() as session:
            with Pool() as pool:
                params = [(self.Buffer_cls, f) for f in files]
                f_gen = pool.imap_unordered(_create_metadata, params)
                f_gen = (
                    tqdm(f_gen, desc="Adding Buffers", total=len(params))
                    if verbose > 0 and len(params) > 0
                    else f_gen
                )
                for i, metadata in enumerate(filter(lambda m: m is not None, f_gen)):
                    metadata: BufferMetadata
                    assert metadata is not None, "BufferMetadata object is None"
                    metadata.machine_id = machine_id
                    session.add(metadata)
                    if i % batch_size == 0:
                        session.commit()
            session.commit()

    def remove_files_from_cache(self, files: List[Path], verbose=0):
        """Remove synchronized files from the cache

        Parameters
        ----------
        files: list[Path]
            complete filepaths that are present in the cache
        verbose: int, optional
            verbosity level. 0 = no feedback, 1 = progress bar
        """
        with self.Session() as session:
            file_iter = (
                tqdm(files, desc="Removing File Entries")
                if verbose > 0 and len(files) > 0
                else files
            )
            for file in file_iter:
                try:
                    with self.Buffer_cls(file) as b:
                        pass
                    entry = (
                        session.query(self.BufferMetadata)
                        .filter(self.BufferMetadata.header_hash == b.header_hash)
                        .one()
                    )
                    if not entry:
                        continue
                    session.delete(entry)
                except Exception as e:
                    session.rollback()
                    raise e
            session.commit()

    def get_buffer_metadata_query(self, buffer_metadata):
        """Converts a `BufferMetadata` object to a complete query. Every property of the object will be converted into
        SQL and returned as a `sqlalchemy.orm.query.FromStatement` object

        Parameters
        ----------
        buffer_metadata : BufferMetadata
            The template BufferMetadata object.

        Returns
        -------
        sqlalchemy.orm.query.FromStatement
            The sqlalchemy query object
        """
        q = "SELECT * FROM buffer_metadata WHERE opening_error IS NULL AND "
        for prop in self.BufferMetadata.properties:
            prop_value = getattr(buffer_metadata, prop)
            if prop_value is not None:
                if isinstance(prop_value, str):
                    prop_value = f"'{prop_value}'"
                elif isinstance(prop_value, Enum):
                    prop_value = f"'{prop_value.name}'"
                q += f"{prop} = {prop_value} AND "
        q = q[:-4]  # prune the last AND
        return select(BufferMetadata).from_statement(text(q))

BufferMetadata class-attribute instance-attribute

BufferMetadata = BufferMetadata

Buffer_cls instance-attribute

Buffer_cls = Buffer_cls

Session instance-attribute

Session = sessionmaker(bind=engine)

engine instance-attribute

engine = create_engine(db_url)

__init__

__init__(db_url='sqlite:///:memory:', Buffer_cls=Buffer)

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def __init__(self, db_url="sqlite:///:memory:", Buffer_cls=Buffer):
    self.engine = create_engine(db_url)
    BufferMetadata.metadata.create_all(self.engine)
    self.Session = sessionmaker(bind=self.engine)
    self.Buffer_cls = Buffer_cls

add_files_to_cache

add_files_to_cache(files: Sequence[Path], verbose: int = 0, batch_size: int = 1000, machine_id: Union[str, None] = None, check_synced: bool = True)

Add buffer files to the cache by providing the complete filepaths If a file (determined by filename and directory) is already synchronized it will be skipped

Parameters:

Name	Type	Description	Default
files	Sequence[Path]	complete filepaths that are added to the cache. The filepath is used with the Buffer class to open a buffer and extract the header information.	required
verbose	int	verbosity level. 0 = no feedback, 1 = progress bar	0
batch_size	int	The batch size after which the cache will commit a batch to the database	1000
machine_id	Union[str, None]	A unique identifier for a different machine	None
check_synced	bool	Whether to check if the files that are about to be added are already synced to the cache	True

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def add_files_to_cache(
    self,
    files: Sequence[Path],
    verbose: int = 0,
    batch_size: int = 1000,
    machine_id: Union[str, None] = None,
    check_synced: bool = True,
):
    """Add buffer files to the cache by providing the complete filepaths
    If a file (determined by filename and directory) is already synchronized it will be skipped

    Parameters
    ----------
    files: Sequence[Path]
        complete filepaths that are added to the cache. The filepath is used with the Buffer class to open a buffer and extract the header information.
    verbose: int, optional
        verbosity level. 0 = no feedback, 1 = progress bar
    batch_size: int, optional
        The batch size after which the cache will commit a batch to the database
    machine_id: str, optional
        A unique identifier for a different machine
    check_synced: bool, optional
        Whether to check if the files that are about to be added are already synced to the cache
    """
    if check_synced:
        files, _ = self.get_non_synchronized_files(files, machine_id)
    with self.Session() as session:
        with Pool() as pool:
            params = [(self.Buffer_cls, f) for f in files]
            f_gen = pool.imap_unordered(_create_metadata, params)
            f_gen = (
                tqdm(f_gen, desc="Adding Buffers", total=len(params))
                if verbose > 0 and len(params) > 0
                else f_gen
            )
            for i, metadata in enumerate(filter(lambda m: m is not None, f_gen)):
                metadata: BufferMetadata
                assert metadata is not None, "BufferMetadata object is None"
                metadata.machine_id = machine_id
                session.add(metadata)
                if i % batch_size == 0:
                    session.commit()
        session.commit()

get_buffer_metadata_query

get_buffer_metadata_query(buffer_metadata)

Converts a BufferMetadata object to a complete query. Every property of the object will be converted into SQL and returned as a sqlalchemy.orm.query.FromStatement object

Parameters:

Name	Type	Description	Default
buffer_metadata	BufferMetadata	The template BufferMetadata object.	required

Returns:

Type	Description
FromStatement	The sqlalchemy query object

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_buffer_metadata_query(self, buffer_metadata):
    """Converts a `BufferMetadata` object to a complete query. Every property of the object will be converted into
    SQL and returned as a `sqlalchemy.orm.query.FromStatement` object

    Parameters
    ----------
    buffer_metadata : BufferMetadata
        The template BufferMetadata object.

    Returns
    -------
    sqlalchemy.orm.query.FromStatement
        The sqlalchemy query object
    """
    q = "SELECT * FROM buffer_metadata WHERE opening_error IS NULL AND "
    for prop in self.BufferMetadata.properties:
        prop_value = getattr(buffer_metadata, prop)
        if prop_value is not None:
            if isinstance(prop_value, str):
                prop_value = f"'{prop_value}'"
            elif isinstance(prop_value, Enum):
                prop_value = f"'{prop_value.name}'"
            q += f"{prop} = {prop_value} AND "
    q = q[:-4]  # prune the last AND
    return select(BufferMetadata).from_statement(text(q))

get_matching_buffers

get_matching_buffers(query: Select) -> List[Buffer]

Calls get_matching_files and converts the result to Buffer objects

Returns:

Type	Description
list[Buffer]	List of Buffer objects

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_matching_buffers(self, query: Select) -> List[Buffer]:
    """Calls get_matching_files and converts the result to Buffer objects

    Returns
    -------
    list[Buffer]
        List of Buffer objects
    """
    files = self.get_matching_files(query)
    buffers = []
    for file in files:
        try:
            with self.Buffer_cls(file) as buffer:
                buffers.append(buffer)
        except Exception:
            pass
    return buffers

get_matching_files

get_matching_files(query: Select) -> List[str]

Query the Cache for all files matching the properties that selected by the query object. The usage of the buffer_metadata, filter_functions and sort_key is deprecated and will be removed in two minor versions. Use the sqlalchemy query parameter instead.

Example

Returns all buffer filepaths with channel = 1, A frequency compression of 4, processes above 100 sorted by the process number

BufferMetadataCache.get_matching_files(
    select(BM).filter(BM.channel==1, BM.compression_freq==4, BM.process > 100)
)

Parameters:

Name	Type	Description	Default
query	Select	A sqlalchemy select statement specifying the properties of the BufferMetadata objects	required

Returns:

Type	Description
list[str]	A list with the paths to the buffer files that match the buffer_metadata

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_matching_files(self, query: Select) -> List[str]:
    """Query the Cache for all files matching the properties that selected by the query object.
    The usage of the buffer_metadata, filter_functions and sort_key is deprecated and will be removed in
    two minor versions. Use the sqlalchemy query parameter instead.

    Example
    -------
    Returns all buffer filepaths with channel = 1, A frequency compression of 4,
    processes above 100 sorted by the process number
    ```py
    BufferMetadataCache.get_matching_files(
        select(BM).filter(BM.channel==1, BM.compression_freq==4, BM.process > 100)
    )
    ```

    Parameters
    ----------
    query: Select
        A sqlalchemy select statement specifying the properties of the BufferMetadata objects

    Returns
    -------
    list[str]
        A list with the paths to the buffer files that match the buffer_metadata
    """
    matching_metadata = self.get_matching_metadata(query)
    return [str(m.filepath) for m in matching_metadata]

get_matching_metadata

get_matching_metadata(query: Select) -> List[BufferMetadata]

Query the cache for all BufferMetadata database entries matching

Parameters:

Name	Type	Description	Default
query	Select	A sqlalchemy select statement specifying the properties of the BufferMetadata objects	required

Returns:

Type	Description
list[BufferMetadata]	A list with the matching BufferMetadata objects

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_matching_metadata(self, query: Select) -> List[BufferMetadata]:
    """Query the cache for all BufferMetadata database entries matching

    Parameters
    ----------
    query: Select
        A sqlalchemy select statement specifying the properties of the BufferMetadata objects

    Returns
    -------
    list[BufferMetadata]
        A list with the matching BufferMetadata objects
    """
    with self.Session() as session:
        matching_metadata = list(session.scalars(query).all())
    return matching_metadata

get_non_synchronized_files

get_non_synchronized_files(files: Sequence[Path], machine_id: Union[str, None] = None) -> Tuple[List[Path], List[Path]]

calculate the difference between the set of files and the set of synchronized files

Parameters:

Name	Type	Description	Default
files	Sequence[Path]	Sequence of files to check whether they are already in the cache	required
machine_id	Union[str, None]	machine identifier to provide extra context for the location of the files	None

Returns:

Type	Description
tuple[list[Path], list[Path]]	The set of files that are not synchronized, and the database entries that exist but the file is not present anymore

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_non_synchronized_files(
    self,
    files: Sequence[Path],
    machine_id: Union[str, None] = None,
) -> Tuple[List[Path], List[Path]]:
    """calculate the difference between the set of files and the set of synchronized files

    Parameters
    ----------
    files: Sequence[Path]
        Sequence of files to check whether they are already in the cache
    machine_id: Union[str, None]
        machine identifier to provide extra context for the location of the files

    Returns
    -------
    tuple[list[Path], list[Path]]
        The set of files that are not synchronized, and the database entries that exist but the file is not present anymore
    """
    file_set = set(files)
    with self.Session() as session:
        synchronized_buffers = set(
            Path(str(buffer.filepath))
            for buffer in session.query(self.BufferMetadata)
            .filter(BufferMetadata.machine_id == machine_id)
            .all()
        )
    unsynchronized_files = [
        Path(p) for p in file_set.difference(synchronized_buffers)
    ]
    synchronized_missing_buffers = [
        Path(p) for p in synchronized_buffers.difference(file_set)
    ]
    return unsynchronized_files, synchronized_missing_buffers

remove_files_from_cache

remove_files_from_cache(files: List[Path], verbose=0)

Remove synchronized files from the cache

Parameters:

Name	Type	Description	Default
files	List[Path]	complete filepaths that are present in the cache	required
verbose		verbosity level. 0 = no feedback, 1 = progress bar	0

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def remove_files_from_cache(self, files: List[Path], verbose=0):
    """Remove synchronized files from the cache

    Parameters
    ----------
    files: list[Path]
        complete filepaths that are present in the cache
    verbose: int, optional
        verbosity level. 0 = no feedback, 1 = progress bar
    """
    with self.Session() as session:
        file_iter = (
            tqdm(files, desc="Removing File Entries")
            if verbose > 0 and len(files) > 0
            else files
        )
        for file in file_iter:
            try:
                with self.Buffer_cls(file) as b:
                    pass
                entry = (
                    session.query(self.BufferMetadata)
                    .filter(self.BufferMetadata.header_hash == b.header_hash)
                    .one()
                )
                if not entry:
                    continue
                session.delete(entry)
            except Exception as e:
                session.rollback()
                raise e
        session.commit()

synchronize_directory

synchronize_directory(*paths, sync_subdirectories=True, regex_pattern='^.*[p][0-9]*[c][0-9]{1}[b][0-9]{2}', verbose=1, delete_stale_entries=False, machine_id=None, glob_pattern='*p*c?b*')

synchronize the buffer files in the given paths with the database matching the regex pattern

Parameters:

Name	Type	Description	Default
paths		The absolute paths to the directory	()
sync_subdirectories		When True synchronize all of the subdirectories recursively, defaults to True	True
regex_pattern		The regex pattern validating the buffer naming format (matched on file.name)	'^.*[p][0-9]*[c][0-9]{1}[b][0-9]{2}'
verbose		verbosity level. 0 = no feedback, 1 = progress bar	1
machine_id		An optional identifier for a certain machine to enable synchronization of different platforms	None
glob_pattern		The pattern forwarded to Path.glob. This pattern acts as a preselection for the files retrieved for the regex pattern	'*p*c?b*'

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def synchronize_directory(
    self,
    *paths,
    sync_subdirectories=True,
    regex_pattern="^.*[p][0-9]*[c][0-9]{1}[b][0-9]{2}",
    verbose=1,
    delete_stale_entries=False,
    machine_id=None,
    glob_pattern="*p*c?b*",
):
    """synchronize the buffer files in the given paths with the database matching the regex pattern

    Parameters
    ----------
    paths: str
        The absolute paths to the directory
    sync_subdirectories: bool, optional
        When True synchronize all of the subdirectories recursively, defaults to True
    regex_pattern: str, optional
        The regex pattern validating the buffer naming format (matched on file.name)
    verbose: int, optional
        verbosity level. 0 = no feedback, 1 = progress bar
    machine_id: str, optional
        An optional identifier for a certain machine to enable synchronization of different platforms
    glob_pattern: str, optional
        The pattern forwarded to Path.glob. This pattern acts as a preselection for the
        files retrieved for the regex pattern
    """
    pattern = re.compile(regex_pattern)
    for path in paths:
        if sync_subdirectories:
            files = [
                file
                for file in Path(path).resolve().rglob(glob_pattern)
                if file.is_file() and pattern.match(file.name)
            ]
        else:
            files = [
                file
                for file in Path(path).resolve().glob(glob_pattern)
                if file.is_file() and pattern.match(file.name)
            ]
        unsynchronized_files, synchronized_missing_buffers = (
            self.get_non_synchronized_files(files, machine_id)
        )
        if delete_stale_entries:
            self.remove_files_from_cache(
                synchronized_missing_buffers, verbose=verbose
            )
        self.add_files_to_cache(
            unsynchronized_files,
            verbose=verbose,
            machine_id=machine_id,
            check_synced=False,
        )

get_declarative_base

get_declarative_base()

Getter for the declarative Base that is used by the class:BufferMetadataCache.

Returns:

Type	Description
declarative base class

Source code in src/qass/tools/analyzer/buffer_metadata_cache.py

def get_declarative_base():
    """Getter for the declarative Base that is used by the :py:class:`BufferMetadataCache`.

    Returns
    -------
    declarative base class
    """
    return __Base