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, load_model: 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, load_model: 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.

  • load_model (bool) – If the model should be read after connection is established. 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.

property mc: ModelCache

Get the model cache for the current session.

Return ods.ModelCache:

ModelCache object containing the cached application model.

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.

property security: Security

Get the security information for the current session.

Raises:

requests.HTTPError – If security info retrieval fails.

Return ods.Security:

Security object containing permissions and roles.

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.

security

Helper class for ASAM ODS Security configuration

class odsbox.security.Security(con_i: ConI)[source]

Bases: object

This class offers the ASAM ODS server security API

class Level(*values)[source]

Bases: IntFlag

This class defines the security levels for the ASAM ODS server. The levels are defined as bit flags found in the security_level attribute of the entities.

static _generate_next_value_(name, start, count, last_values)

Generate the next value when not given.

name: the name of the member start: the initial start value or None count: the number of existing members last_values: the last value assigned or None

class Right(*values)[source]

Bases: IntFlag

This class defines the security rights for the ASAM ODS server. The rights are defined as bit flags found in the security api. In the API they are used as integer values.

static _generate_next_value_(name, start, count, last_values)

Generate the next value when not given.

name: the name of the member start: the initial start value or None count: the number of existing members last_values: the last value assigned or None

initial_rights(security_write_request: SecurityWriteRequest) → None[source]

This method sets the initial rights for newly created instances. :param security_write_request: The security write request :raises requests.HTTPError: If fails.

security_read(security_read_request: SecurityReadRequest) → SecurityInfo[source]

This method reads the security configuration from the ASAM ODS server. :param security_read_request: The security read request :raises requests.HTTPError: If fails. :return: The security information

security_update(security_write_request: SecurityWriteRequest) → None[source]

This method updates the security configuration on the ASAM ODS server. :param security_write_request: The security write request :raises requests.HTTPError: If fails.