`obr.core.queries`#

Module Contents#

Classes#

`query_result`
`Predicates`	Generic enumeration.
`Query`

Functions#

`input_to_query`(→ Query)	converts cli input str to a Query object
`input_to_queries`(→ list[Query])	Convert a json string to list of queries
`execute_query`(→ Query)
`flatten_jobs`(→ dict)	convert a list of jobs to a dictionary
`query_flat_jobs`(→ list[query_result])	Execute queries over a dictionary where the job.id is the key and merged job.docs are the values
`query_to_dict`(→ list[query_result])	Given a list jobs find all jobs for which a query matches
`query_impl`(→ list[dict])	Performs a query and returns a list of records ie for each job the query result
`query_to_records`(→ list[dict])	Given a list jobs find all jobs for which a query matches
`query_to_dataframe`(→ pandas.DataFrame)	Given a list jobs find all jobs for which a query matches
`build_filter_query`(→ list[Query])	This function builds a list of filter queries, where filter queries are queries that request a specific value and has to conform a predicate
`statepoint_get`(statepoint, key)	This function performs a basic recursive query of the statepoint dictionary
`statepoint_query`(statepoint, key, value[, predicate])	This function performs a basic recursive query of the statepoint dictionary
`filter_jobs`(→ list[signac.job.Job])	filter is expected to be a list, string or other iterable of strings in the form of <key><predicate><value>

Attributes#

logger

obr.core.queries.logger#

class obr.core.queries.query_result#

id: str#

result: list[dict]#

sub_keys: list[list[str]]#

__repr__() → str#: Return repr(self).

class obr.core.queries.Predicates#

Bases: enum.Enum

Generic enumeration.

Derive from this class to define new enumerations.

eq = '=='#

neq = '!='#

geq = '>='#

gt = '>'#

leq = '<='#

lt = '<'#

class obr.core.queries.Query#

key: str#

value: Any#

state: dict#

predicate: str = 'eq'#

sub_keys: list#

negate: bool = False#

execute(key, value)#

match()#

__repr__() → str#: Return repr(self).

obr.core.queries.input_to_query(inp: str) → Query#: converts cli input str to a Query object

obr.core.queries.input_to_queries(inp: str) → list[Query]#: Convert a json string to list of queries

obr.core.queries.execute_query(query: Query, key, value, latest_only=True, track_keys=list) → Query#

obr.core.queries.flatten_jobs(jobs: OpenFOAMProject | list[Job]) → dict#: convert a list of jobs to a dictionary

obr.core.queries.query_flat_jobs(jobs: dict[str, dict], queries: list[Query], output, latest_only, strict) → list[query_result]#

Execute queries over a dictionary where the job.id is the key and merged job.docs are the values

Parameters: jobs – a job dictionary ordered by job ids queries – list of queries to run output – Whether to print result to screen latest_only – Take only latest value if resulting value is a list strict – needs all queries to be successful to return a result

obr.core.queries.query_to_dict(jobs: OpenFOAMProject | list[Job], queries: list[Query], output=False, latest_only=True, strict=False) → list[query_result]#

Given a list jobs find all jobs for which a query matches

Flattens list of jobs to a dictionary with merged statepoints and job document first

obr.core.queries.query_impl(jobs: OpenFOAMProject | list[Job], queries: list[Query], output=False, latest_only=True) → list[dict]#: Performs a query and returns a list of records ie for each job the query result

obr.core.queries.query_to_records(jobs: obr.signac_wrapper.operations.OpenFOAMProject, queries: list[Query], latest_only=True, strict=False) → list[dict]#

Given a list jobs find all jobs for which a query matches

Flattens list of jobs to a dictionary with merged statepoints and job document first

obr.core.queries.query_to_dataframe(jobs: obr.signac_wrapper.operations.OpenFOAMProject, queries: list[Query], latest_only=True, strict: bool = False, index: list[str] = [], post_pro: Callable | None = None) → pandas.DataFrame#

Given a list jobs find all jobs for which a query matches

Flattens list of jobs to a dictionary with merged statepoints and job document first Args:

index: A list of strings defining which columns should be used as index post_pro: Function to apply to the DataFrame before creating the index

obr.core.queries.build_filter_query(filters: Iterable[str]) → list[Query]#: This function builds a list of filter queries, where filter queries are queries that request a specific value and has to conform a predicate

obr.core.queries.statepoint_get(statepoint: dict, key: str)#: This function performs a basic recursive query of the statepoint dictionary if the key: value pair is not found in statepoint it recurses into statepoint[“parent”] if present

obr.core.queries.statepoint_query(statepoint: dict, key: str, value, predicate='==')#: This function performs a basic recursive query of the statepoint dictionary if the key: value pair is not found in statepoint it recurses into statepoint[“parent”] if present

obr.core.queries.filter_jobs(project, filter: Iterable[str], output: bool = False) → list[signac.job.Job]#: filter is expected to be a list, string or other iterable of strings in the form of <key><predicate><value>

obr.core.queries#

Module Contents#

Classes#

Functions#

Attributes#

`obr.core.queries`#