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.

file_access(file_identifier: FileIdentifier) → str[source]

Get file access URL for file content.

Parameters:

file_identifier (ods.FileIdentifier) – Define content to be accessed. Might be an AoFile or a DT_BLOB attribute.

Raises:

  • requests.HTTPError – If something went wrong.

  • ValueError – If no file location provided by server.

Return str:

The server file URL.

file_access_delete(file_identifier: FileIdentifier) → None[source]

Delete file content from server.

Parameters:

file_identifier (ods.FileIdentifier) – Define content to be deleted. Might be an AoFile or a DT_BLOB attribute.

Raises:

  • requests.HTTPError – If something went wrong.

  • ValueError – If no open session.

file_access_download(file_identifier: FileIdentifier, target_file_or_folder: str, overwrite_existing: bool = False, default_filename: str = 'download.bin') → str[source]

Read file content from server.

Parameters:

  • file_identifier (ods.FileIdentifier) – Define content to be read. Might be an AoFile or a DT_BLOB attribute.

  • target_file_or_folder (str) – Path to save the file content to. If pointing to an existing folder. Original filename will be used. Full path is returned.

  • overwrite_existing (bool) – If existing files should be overwritten. It defaults to False.

  • default_filename (str) – Default filename if no filename is provided by server. It defaults to “download.bin”.

Raises:

  • requests.HTTPError – If something went wrong.

  • FileExistsError – If file already exists and ‘overwrite_existing’ is False.

  • ValueError – If no open session.

Return str:

file path of saved file.

file_access_upload(file_identifier: FileIdentifier, source_file_path: str) → None[source]

Upload file content to server.

Parameters:

  • file_identifier (ods.FileIdentifier) – Define content to be written. Might be an AoFile or a DT_BLOB attribute.

  • source_file_path (str) – Path to the file to be uploaded.

Raises:

  • requests.HTTPError – If something went wrong.

  • FileNotFoundError – If source file was not found.

  • ValueError – If no open session.

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, headers: dict[str, str] | None = None) → 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, **kwargs) → 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.

  • kwargs – additional arguments passed to to_pandas.

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, name_separator: str = '.') → 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.

  • name_separator (str) – separator used to concatenate entity and attribute names to define column name.

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[.AGGREGATE].

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.