Modules

odsbox

con_i

Helper for ASAM ODS HTTP API conI session

Example:

from odsbox.con_i import ConI

with ConI(
    url="http://localhost:8087/api",
    auth=("sa", "sa")
) as con_i:
    units = con_i.query_data({"AoUnit": {}})

class odsbox.con_i.ConI(url: str = 'http://localhost:8080/api', auth: AuthBase | Tuple[str, str] = ('sa', 'sa'), context_variables: ContextVariables | dict | None = None, verify_certificate: bool = True)[source]

Bases: object

This is a helper to hold an ASAM ODS HTTP API ConI session.

Example:

from odsbox.con_i import ConI

with ConI(
    url="http://localhost:8087/api",
    auth=("sa", "sa")
) as con_i:
    units = con_i.query_data({"AoUnit": {}})

__init__(url: str = 'http://localhost:8080/api', auth: AuthBase | Tuple[str, str] = ('sa', 'sa'), context_variables: ContextVariables | dict | None = None, verify_certificate: bool = True)[source]

Create a session object keeping track of ASAM ODS session URL named conI.

Example:

from odsbox.con_i import ConI

# basic auth
with ConI(
    url="http://localhost:8087/api",
    auth=("sa", "sa")
) as con_i:
    units = con_i.query_data({"AoUnit": {}})

Example:

import requests
from odsbox.con_i import ConI

class BearerAuth(requests.auth.AuthBase):
    def __init__(self, token):
        self.token = token
    def __call__(self, r):
        r.headers["authorization"] = "Bearer " + self.token
        return r

# bearer auth
with ConI(
    url="http://localhost:8087/api",
    auth=BearerAuth("YOUR_BEARER_TOKEN")
) as con_i:
    units = con_i.query_data({"AoUnit": {}})

Parameters:

url (str) – base url of the ASAM ODS API of a given server. An example is “http://localhost:8080/api”.
auth (requests.auth.AuthBase) – An auth object to be used for the used requests package. For basic auth (“USER”, “PASSWORD”) can be used.
context_variables (ods.ContextVariables | dict | None) – If context variables are necessary for the connection they are passed here. It defaults to None.
verify_certificate (bool) – If no certificate is provided for https insecure access can be enabled. It defaults to True.

Raises:

requests.HTTPError – If connection to ASAM ODS server fails.

asampath_create(instance: Instance) → AsamPath[source]

Create an persistent string representing the instance.

Parameters:: instance (ods.Instance) – Instance to be get AsamPath for.
Raises:: requests.HTTPError – If creation fails.
Return ods.AsamPath:: The AsamPath that represents the instance.

asampath_resolve(asam_path: AsamPath) → Instance[source]

Use the persistent string to get back the instance.

Parameters:: asam_path (ods.AsamPath) – AsamPath to be resolved.
Raises:: requests.HTTPError – If path could not be resolved.
Return ods.Instance:: Instance represented by AsamPath.

basemodel_read() → BaseModel[source]

Read the ODS base model version used by the server.

Raises:: requests.HTTPError – If reading base model fails.
Return ods.BaseModel:: used server base model.

con_i_url() → str[source]

Get the ASAM ODS session URL used to work with this session.

Return str:: The ASAM ODS session URL

context_read(pattern_or_filter: ContextVariablesFilter | str = '*') → ContextVariables[source]

Read the session context variables.

Parameters:: pattern_or_filter (ods.ContextVariablesFilter | str) – Context variable filter as str or ContextVariablesFilter. It defaults to “*” to return all variables.
Raises:: requests.HTTPError – If something went wrong.
Return ods.ContextVariables:: ContextVariables where the name matches the filter.

context_update(context_variables: ContextVariables)[source]

Set context variables for current session. This will set context variables for the given session. If new session is created they will fall back to their default.

Parameters:: context_variables (ods.ContextVariables) – ContextVariables to be set or updated.
Raises:: requests.HTTPError – If something went wrong.

data_copy(copy_request: CopyRequest) → Instance[source]

Copy an Instance and its related children.

Parameters:: copy_request (ods.CopyRequest) – Define instance to be copied.
Raises:: requests.HTTPError – If copy fails.
Return ods.Instance:: Newly created instance

data_create(data: DataMatrices) → List[int][source]

Create new ASAM ODS instances or write bulk data.

Parameters:: data (ods.DataMatrices) – Matrices containing columns for instances to be created.
Raises:: requests.HTTPError – If creation fails.
Return List[int]:: list of ids created from your request.

data_delete(data: DataMatrices) → None[source]

Delete existing instances.

Parameters:: data (ods.DataMatrices) – Matrices containing columns for instances to be deleted. The id column is used to identify the instances to be deleted.
Raises:: requests.HTTPError – If delete fails.

data_read(select_statement: SelectStatement) → DataMatrices[source]

Query ods server for content.

Parameters:: select_statement (ods.SelectStatement) – Query given as ASAM ODS SelectStatement.
Raises:: requests.HTTPError – If query fails.
Return ods.DataMatrices:: The DataMatrices representing the result. It will contain one ods.DataMatrix for each returned entity type.

data_read_jaquel(jaquel: str | dict) → DataMatrices[source]

Query ods server for content.

Parameters:: jaquel (str | dict) – Query given as JAQueL query (dict or str).
Raises:: requests.HTTPError – If query fails.
Return ods.DataMatrices:: The DataMatrices representing the result. It will contain one ods.DataMatrix for each returned entity type.

data_update(data: DataMatrices) → None[source]

Update existing instances.

Parameters:: data (ods.DataMatrices) – Matrices containing columns for instances to be updated. The id column is used to identify the instances to be updated.
Raises:: requests.HTTPError – If update fails.

logout()[source]

Close the attached session at the ODS server by calling delete on the session URL and closing the requests session.

Raises:: requests.HTTPError – If delete the ASAM ODS session fails.

model() → Model[source]

Get the cache ODS server model. This model will return the cached application model related to your session.

Return ods.Model:: The application model of the ASAM ODS server.

model_check() → None[source]

Check if stored application model is consistent.

Raises:: requests.HTTPError – If model contains errors.

model_delete(model_parts: Model, update_model: bool = True) → None[source]

Delete application model content.

Parameters:

model_parts (ods.Model) – define model parts to be deleted.
update_model (bool) – determine if the model cache of this ConI instance should be updated by reading the whole model again. It defaults to True.

Raises:

requests.HTTPError – If model update fails.

model_read() → Model[source]

Read the model from server and update cached version.

Raises:: requests.HTTPError – If model read fails.
Return ods.Model:: The application model of the server.

model_update(model_parts: Model, update_model: bool = True) → None[source]

Update application model content. This method is used to modify existing items or create new ones.

Parameters:

model_parts (ods.Model) – parts of the model to be updated or created.
update_model (bool) – determine if the model cache of this ConI instance should be updated by reading the whole model again. It defaults to True.

Raises:

requests.HTTPError – If model update fails.

n_m_relation_read(identifier: NtoMRelationIdentifier) → NtoMRelatedInstances[source]

Read n-m relations for a defined instance.

Parameters:: identifier (ods.NtoMRelationIdentifier) – identify n to m relation to be read.
Raises:: requests.HTTPError – If read fails.
Return ods.NtoMRelatedInstances:: Return n to m related instances that were queried.

n_m_relation_write(related_instances: NtoMWriteRelatedInstances) → None[source]

Update, delete or create n-m relations for given instance pairs.

Raises:: requests.HTTPError – If write fails.
Parameters:: related_instances (ods.NtoMWriteRelatedInstances) – related instances to be updated, deleted or created.

ods_post_request(relative_url_part: str, message: Message | None = None, timeout: float = 600.0) → Response[source]

Do ODS post call with the given relative URL.

Parameters:

relative_url_part (str) – url part that is joined to conI URL using /.
message (Message | None) – protobuf message to be send, defaults to None.
timeout (float) – maximal time to wait for response.

Raises:

requests.HTTPError – If status code is not 200 or 201.

Return requests.Response:

requests response if successful.

password_update(password_update: PasswordUpdate) → None[source]

Update the password of the defined user.

Parameters:: password_update (ods.PasswordUpdate) – Defines for which user the password should eb updated.
Raises:: requests.HTTPError – If something went wrong.

query_data(query: str | dict | SelectStatement, enum_as_string: bool = False, date_as_timestamp: bool = False) → DataFrame[source]

Query ods server for content and return the results as Pandas DataFrame

Parameters:

query (str | dict | ods.SelectStatement) – Query given as JAQueL query (dict or str) or as an ASAM ODS SelectStatement.
enum_as_string (bool) – columns of type DT_ENUM are returned as int values. If this is set to True the int values are mapped to the corresponding string values.
date_as_timestamp (bool) – columns of type DT_DATE or DS_DATE are returned as string. If this is set to True the strings are converted to pandas Timestamp.

Raises:

requests.HTTPError – If query fails.

Return DataFrame:

The DataMatrices as Pandas.DataFrame. The columns are named as ENTITY_NAME.ATTRIBUTE_NAME. IsNull values are not marked invalid.

transaction() → Transaction[source]

Open a transaction object to be used in a with clause

Example:

with con_i.transaction() as transaction:
    # do writing
    transaction.commit()

Raises:: requests.HTTPError – If creation of transaction fails.
Return Transaction:: transaction object that will abort automatically if commit is not called.

transaction_abort() → None[source]: Abort transaction created before. :raises requests.HTTPError: If creation of transaction fails.

transaction_commit() → None[source]: Commit transaction created before. :raises requests.HTTPError: If creation of transaction fails.

transaction_create() → None[source]: Open a transaction for writing. :raises requests.HTTPError: If creation of transaction fails.

valuematrix_read(request: ValueMatrixRequestStruct) → DataMatrices[source]

Read bulk data from a submatrix or measurement. Submatrix access can also be done using data-read.

Parameters:: request (ods.ValueMatrixRequestStruct) – Define measurement or submatrix to create ASAM ODS ValueMatrix for.
Raises:: requests.HTTPError – If ValueMatrix access fails.
Return ods.DataMatrices:: DataMatrices containing the bulk data for the request.

transaction

helper for handling transactions

class odsbox.transaction.Transaction(con_i: ConI)[source]

Bases: object

Class helps to keep track of transactions. If no commit is called it will abort the transaction if with sections is left.

abort() → None[source]: Aborts the transaction.

commit() → None[source]: Commit the transaction.

unit_catalog

Helper class for Units in ASAM ODS

class odsbox.unit_catalog.UnitCatalog(con_i: ConI)[source]

Bases: object

This class caches the units stored in the ASAM ODS server. If a Unit does not exist it is created with a physical dimension unknown.

create(unit_name: str) → int[source]

Create a unit by its case sensitive name using an unknown physical dimension.

Parameters:: unit_name (str) – Case sensitive name of an unit.
Return int:: The unit id if the unit exists.

get(unit_name: str) → int | None[source]

Get a unit by its case sensitive name.

Parameters:: unit_name (str) – Case sensitive name of an unit.
Return int | None:: The unit id if the unit exists. Else None is returned.

get_or_create(unit_name: str) → int[source]

Get a unit by its case sensitive name or create one using an unknown physical dimension.

Parameters:: unit_name (str) – Case sensitive name of an unit.
Return int:: The unit id if the unit exists.

model_cache

helps working with the ASAM ODS model

class odsbox.model_cache.ModelCache(model: Model)[source]

Bases: object

The ods.Model object returned from ods server needs some utilities to work with it. This cache functionality useful for daily work.

aid(entity_or_name: str | Entity) → int[source]

Determine the application element id of an entity by its name.

Parameters:: entity_or_name (str) – entity object or case sensitive application name to lookup
Raises:: ValueError – If the entity does not exist.
Return int:: The ApplicationElementId of the entity.

attribute(entity_or_name: str | Entity, application_or_base_name: str) → Attribute[source]

This is a convenience method to find an attribute. It will first check for an attribute with the given application name and afterwards check for an attribute with the given base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity or case insensitive name of an entity
application_or_base_name (str) – case insensitive name to lookup

Raises:

ValueError – If attribute does not.

Return ods.Model.Attribute:

The found attribute.

attribute_by_base_name(entity_or_name: str | Entity, attribute_base_name: str) → Attribute[source]

Get the attribute by base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity object or case sensitive application name to lookup
attribute_base_name (str) – case insensitive name of the base model element.

Raises:

ValueError – If the attribute does not exist.

Return ods.Model.Attribute:

Corresponding attribute.

attribute_no_throw(entity_or_name: str | Entity, application_or_base_name: str) → Attribute | None[source]

This is a convenience method to find an attribute. It will first check for an attribute with the given application name and afterwards check for an attribute with the given base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity or case insensitive name of an entity
application_or_base_name (str) – case insensitive name to lookup

Raises:

ValueError – If entity does not exist.

Return ods.Model.Attribute | None:

The found attribute or None.

entity(entity_name: str) → Entity[source]

Get the entity name.

Parameters:: entity_name (str) – case insensitive name of an entity.
Raises:: ValueError – If the entity does not exist.

entity_by_aid(aid: int) → Entity[source]

Get the entity by its ApplicationElementId(aid).

Parameters:: aid (int) – ApplicationElementId of an entity to lookup.
Raises:: ValueError – If the entity does not exist.
Return ods.Model.Entity:: Entity corresponding to given aid.

entity_by_base_name(entity_base_name: str) → Entity[source]

Get the entity by its base name.

Parameters:: entity_base_name (str) – case insensitive name of the base model element.
Raises:: ValueError – If the entity does not exist.

enumeration(enumeration_name: str) → Enumeration[source]

Get enumeration by its name.

Parameters:: enumeration_name (str) – case insensitive name of the application model enumeration.
Raises:: ValueError – If the enumeration does not exist.
Return ods.Model.Enumeration:: Corresponding enumeration.

enumeration_key_to_value(enumeration_or_name: str | Enumeration, lookup_key: str) → int[source]

Convert an enumeration integer value into its string representation.

Parameters:

enumeration_or_name (str | ods.Model.Enumeration) – ods enumeration or its case insensitive name
lookup_key (str) – case insensitive string key value to check

Raises:

ValueError – If the enumeration does not exist or does not contain the key.

Return str:

Int representation of string value.

enumeration_value_to_key(enumeration_or_name: str | Enumeration, lookup_value: int) → str[source]

Convert an enumeration value into its string representation.

Parameters:

enumeration_or_name (str | ods.Model.Enumeration) – ods enumeration or its case insensitive name
lookup_value (int) – integer value to check

Raises:

ValueError – If the enumeration does not exist or does not contain value.

Return str:

String representation of int value.

model() → Model[source]

Get the attached ASAM ODS model

Return ods.Model:: The used model.

relation(entity_or_name: str | Entity, application_or_base_name: str) → Relation[source]

This is a convenience method to find a relation. It will first check for a relation with the given application name and afterwards check for a relation with the given base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity or case insensitive name of an entity
application_or_base_name (str) – case insensitive name to lookup

Raises:

ValueError – If relation does not exist.

Return ods.Model.Relation:

The found relation.

relation_by_base_name(entity_or_name: str | Entity, relation_base_name: str) → Relation[source]

Get the relation by base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity object or case sensitive application name to lookup
relation_base_name (str) – case insensitive name of the base model element.

Raises:

ValueError – If the relation does not exist.

Return ods.Model.Relation:

Corresponding relation.

relation_no_throw(entity_or_name: str | Entity, application_or_base_name: str) → Relation | None[source]

This is a convenience method to find a relation. It will first check for a relation with the given application name and afterwards check for a relation with the given base name.

Parameters:

entity_or_name (str | ods.Model.Entity) – entity or case insensitive name of an entity
application_or_base_name (str) – case insensitive name to lookup

Raises:

ValueError – If entity does not exist.

Return ods.Model.Relation | None:

The relation or None if it does not exist.

jaquel

Used to convert JAQueL queries to ASAM ODS SelectStatements

odsbox.jaquel.jaquel_to_ods(model: Model, jaquel_query: str | dict) → Tuple[Entity, SelectStatement][source]

Convert a given JAQueL query into an ASAM ODS SelectStatement.

Parameters:

model (ods.Model) – application model to be used for conversion.
jaquel_query (str | dict) – JAQueL query as dict or json string.

Raises:

SyntaxError – If contains syntactical errors.
ValueError – If conversion fail.
json.decoder.JSONDecodeError – If JSON string contains syntax errors.

Return Tuple[ods.Model.Entity, ods.SelectStatement]:

A tuple defining the target entity and the ASAM ODS SelectStatement

datamatrices_to_pandas

ods works with datamatrices object. This utilities converts them to an pandas DataFrame for ease of use.

odsbox.datamatrices_to_pandas.to_pandas(data_matrices: DataMatrices, model_cache: ModelCache | None = None, enum_as_string: bool = False, date_as_timestamp: bool = False) → DataFrame[source]

Converts data in an ASAM ODS DataMatrices into a pandas DataFrame.

Parameters:

data_matrices (ods.DataMatrices) – matrices to be converted.
model_cache (ModelCache | None) – ModelCache is used to do enum conversion
enum_as_string (bool) – columns of type DT_ENUM or DS_ENUM are returned as int values. If this is set to True the model_cache is used to map the int values to the corresponding string values.
date_as_timestamp (bool) – columns of type DT_DATE or DS_DATE are returned as string. If this is set to True the strings are converted to pandas Timestamp.

Return pd.DataFrame:

A pandas DataFrame containing all the single matrices in a single frame. The columns are named by the schema ENTITY_NAME.ATTRIBUTE_NAME.

odsbox.datamatrices_to_pandas.unknown_array_values(unknown_array: UnknownArray, date_as_timestamp: bool = False) → list[source]

Get the values of an UnknownArray as list

Parameters:

unknown_array (ods.DataMatrix.Column.UnknownArray) – ASAM ODS unknown array to transport array of any
date_as_timestamp (bool) – columns of type DT_DATE or DS_DATE are returned as string. If this is set to True the strings are converted to pandas Timestamp.

Raises:

ValueError – If standard is extended by new types

Return list | np.array:

list containing the values of the Unknown array.

submatrix_to_pandas

converts a submatrix delivered as a datamatrix protobuf object into a pandas DataFrame.

odsbox.submatrix_to_pandas.submatrix_to_pandas(con_i: ConI, submatrix_iid: int, date_as_timestamp: bool = False) → DataFrame[source]

Loads an ASAM ODS SubMatrix and returns it as a pandas DataFrame.

Parameters:

con_i (ConI) – ASAM ODS server session.
submatrix_iid (int) – id of an submatrix to be retrieved.
date_as_timestamp (bool) – columns of type DT_DATE or DS_DATE are returned as string. If this is set to True the strings are converted to pandas Timestamp.

Return pd.DataFrame:

A pandas DataFrame containing the values of the localcolumn as pandas columns. The name of the localcolumn is used as pandas column name. The flags are ignored.

asam_time

datetime is represented as string in ASAM ODS formatted using YYYYMMDDHHMMSSFFF. Here you find some helpers.

odsbox.asam_time.from_pd_timestamp(timestamp: Timestamp, length: int = 17) → str[source]

Convert a pandas Timestamp to a string formatted as asamtime (YYYYMMDDHHMMSSFFF).

Parameters:

timestamp (pd.Timestamp) – The pandas Timestamp to convert. The timezone information given in timestamp is ignored.
length (int) – The desired length of the output string. The final string will be truncated to the specified length. The maximum is 23 including nanoseconds.

Return str:

The asam time representation of the timestamp. For None or pd.NaT an empty string is returned.

odsbox.asam_time.to_pd_timestamp(asam_time: str) → Timestamp[source]

Convert ASAM ODS datetime string to pandas Timestamp.

Parameters:: asam_time (str) – ASAM ODS datetime string to be converted. formatted like YYYYMMDDHHMMSSFFF. It must at least contain YYYYMMDD.
Raises:: requests.SyntaxError – If content is invalid.
Return pd.Timestamp:: Corresponding pandas Timestamp value. For empty string pd.NaT is returned.