pyprediktormapclient package

Subpackages

• pyprediktormapclient.dwh package
  • Subpackages
    • pyprediktormapclient.dwh.context package
      • Submodules
      • pyprediktormapclient.dwh.context.enercast module
      • pyprediktormapclient.dwh.context.plant module
      • pyprediktormapclient.dwh.context.solcast module
      • Module contents
  • Submodules
  • pyprediktormapclient.dwh.db module
    • Db
  • pyprediktormapclient.dwh.dwh module
    • DWH
      • DWH.connection
      • DWH.cursor
      • DWH.version()
  • pyprediktormapclient.dwh.idwh module
    • IDWH
      • IDWH.execute()
      • IDWH.fetch()
      • IDWH.version()
  • Module contents

Submodules

pyprediktormapclient.analytics_helper module

class pyprediktormapclient.analytics_helper.AnalyticsHelper(input: List)

Bases: object

Your data as a Pandas DataFrame, but with a specific formatting and some nifty functions. Put the data from ModelIndex in here and move further on by adding data and from other ModelIndex calls or live or historical data from OPC UA.

Columns in the normalizes dataframe are:

• Id

• Name

• Type

• Props

  • DisplayName

  • Value

• Vars

  • DisplayName

  • Id

Parameters:

input (List) – The return from a ModelIndex call function

dataframe

The normalized dataframe

Type:

pandas.DataFrame

Returns:

An instance of the class with some resources and attributes

ID_PATTERN = '^\\d+:\\d+:\\S+$'

list_of_ids() → list

Extracts the values in the column “Id” to a list of unique IDs.

Returns:

Unique IDs

Return type:

list

list_of_names() → list

Extracts the values in the column “Name” to a list of unique names.

Returns:

Unique names

Return type:

list

list_of_types() → list

Extracts the values in the column “Type” to a list of unique types.

Returns:

Unique types

Return type:

list

list_of_variable_names() → list

Explodes the content of the column “Vars” and extracts the values from DisplayName into a list of unique values.

Returns:

Unique variable names

Return type:

list

namespaces_as_list(list_of_dicts: List) → List

Takes the output of a get_namespace_array() request from ModelIndex and generates a list of strings that can be used for the OPC UA Values API.

Parameters:

list_of_dicts (List) – A list in of dicts like [{‘Idx’: 0, ‘Uri’: ‘http://opcfoundation.org/UA/’}, etc]

Returns:

A list of strings containing the URIs

Return type:

List

normalize()

Normalize column names in the dataframe class attribute. Different ModelIndex calls has different column names but this will be normalized with this function according to the class docs.

Returns:

Nothing, but normalizes the instance “dataframe”

properties_as_dataframe() → DataFrame

Explodes the column “Props” into a new dataframe. Column names will be.

• Id (same as from the original dataframe)

• Name (same as from the original dataframe)

• Type (same as from the original dataframe)

• Property (from the exploded value DisplayName)

• Value (from the exploded value Value)

Returns:

A new dataframe with all properties as individual rows

Return type:

pandas.DataFrame

split_id(id: Any) → dict

variables_as_dataframe() → DataFrame

Explodes the column “Vars” into a new dataframe. Column names will be.

• Id (same as from the original dataframe)

• Name (same as from the original dataframe)

• Type (same as from the original dataframe)

• VariableId (from the exploded value Id)

• VariableName (from the exploded value DisplayName)

Returns:

A new dataframe with all variables as individual rows

Return type:

pandas.DataFrame

variables_as_list(include_only: List = []) → List

Extracts variables as a list. If there are names listed in the include_only argument, only variables matching that name will be encluded.

Parameters:

include_only (list) – A list of variable names (str) that should be included

Returns:

Unique types

Return type:

list

pyprediktormapclient.auth_client module

class pyprediktormapclient.auth_client.AUTH_CLIENT(rest_url: Url, username: str, password: str)

Bases: object

Helper functions to authenticate with Ory.

Parameters:

• rest_url (str) – The complete url of the OPC UA Values REST API. E.g. “http://127.0.0.1:13371/”

• opcua_url (str) – The complete url of the OPC UA Server that is passed on to the REST server. E.g. “opc.tcp://127.0.0.1:4872”

• namespaces (list) – An optional but recommended ordered list of namespaces so that IDs match

Returns:

Object

check_if_token_has_expired() → bool

Check if token has expired.

get_login_id() → None

Request login token from Ory.

get_login_token() → None

Request login token from Ory.

request_new_ory_token() → None

Request Ory token.

class pyprediktormapclient.auth_client.Ory_Login_Structure(*, method: str, identifier: str, password: str)

Bases: BaseModel

identifier: str

method: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'identifier': FieldInfo(annotation=str, required=True), 'method': FieldInfo(annotation=str, required=True), 'password': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

password: str

class pyprediktormapclient.auth_client.Token(*, session_token: str, expires_at: datetime | None = None)

Bases: BaseModel

expires_at: datetime | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {'arbitrary_types_allowed': True}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'expires_at': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'session_token': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

classmethod remove_nanoseconds(v)

session_token: str

pyprediktormapclient.model_index module

class pyprediktormapclient.model_index.ModelIndex(url: Url, auth_client: object = None, session: Session = None)

Bases: object

Helper functions to access the ModelIndex API server.

Parameters:

url (str) – The URL of the ModelIndex server with the trailing slash

class Config

Bases: object

arbitrary_types_allowed = True

check_auth_client(content)

get_namespace_array() → str

Get the namespace array.

Returns:

the JSON returned from the server

Return type:

str

get_object_ancestors(type_name: str, ids: List, domain: str) → str

Function to get object ancestors.

Parameters:

• type_name (str) – the ancestor parent type

• ids (list) – a list of ids you want the ancestors for

• domain (str) – Either PV_Assets or PV_Serves

Returns:

A json-formatted string with ancestors data of selected object (or None if the type is not found)

get_object_descendants(type_name: str, ids: List, domain: str) → str

A function to get object descendants.

Parameters:

• type_name (str) – type_name of a descendant

• ids (list) – a list of ids you want the descendants for

• domain (str) – PV_Assets or PV_Serves

Returns:

A json-formatted string with descendats data of selected object (or None if the type is not found)

get_object_type_id_from_name(type_name: str) → str

Get object type id from type name.

Parameters:

type_name (str) – type name

Returns:

the type id that corresponds with the id (None if not found)

Return type:

str

get_object_types() → str

get_objects_of_type(type_name: str) → str

Function to get all the types of an object.

Parameters:

type_name (str) – type name

Returns:

A json-formatted string with the objects (or None if the type is not found)

json_serial(obj)

JSON serializer for objects not serializable by default json code.

pyprediktormapclient.opc_ua module

class pyprediktormapclient.opc_ua.AsyncIONotebookHelper

Bases: object

static run_coroutine(coroutine)

class pyprediktormapclient.opc_ua.Config

Bases: object

arbitrary_types_allowed = True

class pyprediktormapclient.opc_ua.HistoryValue(*, Value: SubValue)

Bases: BaseModel

Helper class to parse all values api’s.

Variables:

Value: SubValue - Containing Type and Body (value) of the variable. Described in SubValue class.

Value: SubValue

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'Value': FieldInfo(annotation=SubValue, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.OPC_UA(rest_url: Url, opcua_url: Url, namespaces: List = None, auth_client: object = None, session: Session = None)

Bases: object

Helper functions to access the OPC UA REST Values API server.

Parameters:

• rest_url (str) – The complete url of the OPC UA Values REST API. E.g. “http://127.0.0.1:13371/”

• opcua_url (str) – The complete url of the OPC UA Server that is passed on to the REST server. E.g. “opc.tcp://127.0.0.1:4872”

• namespaces (list) – An optional but recommended ordered list of namespaces so that IDs match

Returns:

Object

class Config

Bases: object

arbitrary_types_allowed = True

check_auth_client(content)

check_if_ory_session_token_is_valid_refresh()

Check if the session token is still valid.

get_historical_aggregated_values(*args, **kwargs)

async get_historical_aggregated_values_asyn(start_time: datetime, end_time: datetime, pro_interval: int, agg_name: str, variable_list: List[str], **kwargs) → DataFrame

Request historical aggregated values from the OPC UA server.

get_historical_raw_values(*args, **kwargs)

async get_historical_raw_values_asyn(start_time: datetime, end_time: datetime, variable_list: List[str], limit_start_index: int | None = None, limit_num_records: int | None = None, **kwargs) → DataFrame

Request raw historical values from the OPC UA server.

async get_historical_values(start_time: datetime, end_time: datetime, variable_list: List[str], endpoint: str, prepare_variables: Callable[[List[str]], List[dict]], additional_params: dict = None, max_data_points: int = 10000, max_retries: int = 3, retry_delay: int = 5, max_concurrent_requests: int = 30) → DataFrame

Generic method to request historical values from the OPC UA server with batching.

get_values(variable_list: List[Variables]) → List

Request realtime values from the OPC UA server.

Parameters:

variable_list (list) – A list of variables you want, containing keys “Id”, “Namespace” and “IdType”

Returns:

The input variable_list extended with “Timestamp”, “Value”, “ValueType”, “StatusCode” and “StatusSymbol” (all defaults to None)

Return type:

list

json_serial(obj)

JSON serializer for objects not serializable by default json code.

write_historical_values(variable_list: List[WriteHistoricalVariables]) → List

Request to write realtime values to the OPC UA server.

Parameters:

variable_list (list) – A list of variables you want, containing keys “Id”, “Namespace”, “Values” and “IdType”. Values must be in descending order of the timestamps.

Returns:

The input variable_list extended with “Timestamp”, “Value”, “ValueType”, “StatusCode” and “StatusSymbol” (all defaults to None)

Return type:

list

write_values(variable_list: List[WriteVariables]) → List

Request to write realtime values to the OPC UA server.

Parameters:

variable_list (list) – A list of variables you want to write to with the value, timestamp and quality, containing keys “Id”, “Namespace”, “Values” and “IdType”.

Returns:

The input variable_list extended with “Timestamp”, “Value”, “ValueType”, “StatusCode” and “StatusSymbol” (all defaults to None)

Return type:

list

class pyprediktormapclient.opc_ua.StatsCode(*, Code: int | None = None, Symbol: str | None = None)

Bases: BaseModel

Helper class to parse all values api’s.

Variables:

Code: Optional[int] - Status code, described in https://reference.opcfoundation.org/v104/Core/docs/Part8/A.4.3/ Symbol: Optional[str] - String value for status code, described in https://reference.opcfoundation.org/v104/Core/docs/Part8/A.4.3/

Code: int | None

Symbol: str | None

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'Code': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'Symbol': FieldInfo(annotation=Union[str, NoneType], required=False, default=None)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.SubValue(*, Type: int, Body: str | float | int | bool)

Bases: BaseModel

Helper class to parse all values api’s.

Variables:

Type: int - Type of variable, e.g. 12. string, 11. float, etc. list of types described in https://reference.opcfoundation.org/Core/Part6/v104/5.1.2/ Body: Union[str, float, int, bool] - The value of the varible, should match type.

Body: str | float | int | bool

Type: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'Body': FieldInfo(annotation=Union[str, float, int, bool], required=True), 'Type': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.Value(*, Value: SubValue, SourceTimestamp: datetime, SourcePicoseconds: int | None = None, ServerTimestamp: datetime | None = None, ServerPicoseconds: int | None = None, StatusCode: StatsCode | None = None)

Bases: BaseModel

Helper class to parse all values api’s.

Variables:

Value: SubValue - Containing Type and Body (value) of the variable. Described in SubValue class. SourceTimestamp: str - Timestamp of the source, e.g. when coming from an API the timestamp returned from the API for the varaible is the sourcetimestamp. SourcePicoseconds: Optional[int] - Picoseconds for the timestamp of the source if there is a need for a finer granularity, e.g. if samples are sampled in picosecond level or more precision is needed. ServerTimestamp: Optional[str] - Timestamp for the server, normally this is assigned by the server. ServerPicoseconds: Optional[int] - Picoseconds for the timestamp on the server, normally this is assigned by the server. StatusCode: StatusCode - Status code, described in https://reference.opcfoundation.org/v104/Core/docs/Part8/A.4.3/

ServerPicoseconds: int | None

ServerTimestamp: datetime | None

SourcePicoseconds: int | None

SourceTimestamp: datetime

StatusCode: StatsCode | None

Value: SubValue

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'ServerPicoseconds': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'ServerTimestamp': FieldInfo(annotation=Union[datetime, NoneType], required=False, default=None), 'SourcePicoseconds': FieldInfo(annotation=Union[int, NoneType], required=False, default=None), 'SourceTimestamp': FieldInfo(annotation=datetime, required=True), 'StatusCode': FieldInfo(annotation=Union[StatsCode, NoneType], required=False, default=None), 'Value': FieldInfo(annotation=SubValue, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.Variables(*, Id: str, Namespace: int, IdType: int)

Bases: BaseModel

Helper class to parse all values api’s. Variables are described in https://reference.opcfoundation.org/v104/Core/docs/Part3/8.2.1/

Variables:

Id: str - Id of the signal, e.g. SSO.EG-AS.WeatherSymbol Namespace: int - Namespace on the signal, e.g. 2. IdType: int - IdTypes described in https://reference.opcfoundation.org/v104/Core/docs/Part3/8.2.3/.

Id: str

IdType: int

Namespace: int

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'Id': FieldInfo(annotation=str, required=True), 'IdType': FieldInfo(annotation=int, required=True), 'Namespace': FieldInfo(annotation=int, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.WriteHistoricalVariables(*, NodeId: Variables, PerformInsertReplace: int, UpdateValues: List[Value])

Bases: BaseModel

Helper class for write historical values api.

Variables:

NodeId (str): The complete node’id for the variable PerformInsertReplace (int): Historical insertion method 1. Insert, 2. Replace 3. Update, 4. Remove UpdateValues (list): List of values to update for the node’id. Time must be in descending order.

NodeId: Variables

PerformInsertReplace: int

UpdateValues: List[Value]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'NodeId': FieldInfo(annotation=Variables, required=True), 'PerformInsertReplace': FieldInfo(annotation=int, required=True), 'UpdateValues': FieldInfo(annotation=List[Value], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.WriteReturn(*, Id: str, Value: str, TimeStamp: str, Success: bool)

Bases: BaseModel

Helper class to collect API output with API input to see successfull writes for nodes.

Variables:

Id: str - The Id of the signal Value: str - The written value of the signal TimeStamp: str - THe SourceTimestamp of the written signal Success: bool - Success flag for the write operation

Id: str

Success: bool

TimeStamp: str

Value: str

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'Id': FieldInfo(annotation=str, required=True), 'Success': FieldInfo(annotation=bool, required=True), 'TimeStamp': FieldInfo(annotation=str, required=True), 'Value': FieldInfo(annotation=str, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.WriteVariables(*, NodeId: Variables, Value: Value)

Bases: BaseModel

Helper class for write values api.

Variables:

NodeId: Variables - The complete node’id for the variable Value: Value - The value to update for the node’id.

NodeId: Variables

Value: Value

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'NodeId': FieldInfo(annotation=Variables, required=True), 'Value': FieldInfo(annotation=Value, required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

class pyprediktormapclient.opc_ua.WriteVariablesResponse(*, SymbolCodes: List[StatsCode])

Bases: BaseModel

Helper class for write historical values api.

Variables:

SymbolCodes: List[StatusCode] - A list of class StatusCode, described in StatusCode class.

SymbolCodes: List[StatsCode]

model_computed_fields: ClassVar[Dict[str, ComputedFieldInfo]] = {}

A dictionary of computed field names and their corresponding ComputedFieldInfo objects.

model_config: ClassVar[ConfigDict] = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

model_fields: ClassVar[Dict[str, FieldInfo]] = {'SymbolCodes': FieldInfo(annotation=List[StatsCode], required=True)}

Metadata about the fields defined on the model, mapping of field names to [FieldInfo][pydantic.fields.FieldInfo] objects.

This replaces Model.__fields__ from Pydantic V1.

pyprediktormapclient.shared module

class pyprediktormapclient.shared.Config

Bases: object

arbitrary_types_allowed = True

pyprediktormapclient.shared.request_from_api(rest_url: Url, method: Literal['GET', 'POST'], endpoint: str, data: str = None, params: dict = None, headers: dict = None, extended_timeout: bool = False, session: Session = None) → str

Function to perform the request to the ModelIndex server.

Parameters:

• rest_url (str) – The URL with trailing shash

• method (str) – “GET” or “POST”

• endpoint (str) – The last part of the url (without the leading slash)

• data (str) – defaults to None but can contain the data to send to the endpoint

• headers (str) – default to None but can contain the headers og the request

Returns:

The result if successfull

Return type:

JSON

Module contents