Modules

Top-level package for ElHub Python SDK.

acknolwedgment

Acknowledgement related functions

After the market party has persisted the received messages, an Acknowledgement message referring to the polled data set must be sent back to Elhub on the polling service to confirm that the messages has been received.

Note: Elhub will send a 200 response regardless of the validity of the Original Business Document Reference. To confirm that the acknowlegment was sucessfull poll again and check if the request have been acknowledged.

acknowledge_poll(client, history, sender_gsn, original_business_document_reference, process_role=<ROLES.THIRD_PARTY: 'AG'>)

Metering Values | Market processes WSDL

Parameters:

Name	Type	Description	Default
history	HistoryPlugin		required
client	Client		required
meter_identificator			required
sender_gsn	str		required
original_business_document_reference	str		required
process_role	ROLES		<ROLES.THIRD_PARTY: 'AG'>

Source code in elhub_sdk/acknolwedgment.py

def acknowledge_poll(
    client: zeep.Client,
    history: HistoryPlugin,
    sender_gsn: str,
    original_business_document_reference: str,
    process_role: ROLES = ROLES.THIRD_PARTY,
) -> bool:
    """
    Metering Values | Market processes WSDL
    Args:
        history:
        client:
        meter_identificator:
        sender_gsn:
        original_business_document_reference:
        process_role:

    Returns:

    """

    factory = client.type_factory('ns4')
    eh_request = factory.Acknowledgement(
        Header={
            'Identification': uuid.uuid4(),
            'DocumentType': {
                '_value_1': DOCUMENT_TYPE_UN_CEFACT.QUERY.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'Creation': f'{datetime.utcnow().strftime(TIME_FORMAT)}',
            'PhysicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalRecipientEnergyParty': {
                'Identification': {'_value_1': ELHUB_GSN, 'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value}
            },
        },
        ProcessEnergyContext={  # https://dok.elhub.no/ediel2/general#General-Process
            'EnergyBusinessProcess': {
                '_value_1': BSR_IDS.METERING_VALUES.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.ELHUB.value,
            },
            'EnergyBusinessProcessRole': {
                '_value_1': process_role.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'EnergyIndustryClassification': "23",
        },
        PayloadResponseEvent={
            'StatusType': {
                '_value_1': STATUS_TYPE.ACCEPTED.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'OriginalBusinessDocumentReference': original_business_document_reference,
        },
    )

    try:
        response = client.service.Acknowledge(eh_request)
        if history.last_received:
            return True
        logger.error(f"Unknown error: {response}")
    except Exception as ex:
        logger.exception(ex)
    return False

client

ElHub API client

APIClient

Main ElHub API client class

Source code in elhub_sdk/client.py

class APIClient:
    """
    Main ElHub API client class
    """

    @staticmethod
    def get_client(
        environment: ElHubEnvironment, service: ElHubService, key_file: str = "", cert_file: str = ""
    ) -> Tuple[Client, HistoryPlugin]:
        """
        Get a production client and a history plugin
        Args:
            environment: the environment to use in Elhub
            service: the service to use in Elhub
            key_file: the key file to use
            cert_file: the certificate file to use
        Returns:

        """
        return APIClient._get_zeep_client(
            wsdl=os.path.join(os.path.dirname(__file__), "wsdl", environment, "2.3", "wsdl", service),
            secure=True,
            key_file=key_file,
            cert_file=cert_file,
        )

    @staticmethod
    def _get_zeep_client(
        wsdl, secure: bool = True, key_file: str = "", cert_file: str = ""
    ) -> Tuple[Client, HistoryPlugin]:
        """
        Returns a client
        Args:
            wsdl:
            key_file:
            cert_file:
            secure:

        Returns:

        """
        history = HistoryPlugin()

        client_settings = Settings(strict=False)
        binary_signature = None
        if secure:
            binary_signature = BinarySignatureTimestamp(
                key_file=key_file,
                certfile=cert_file,
            )

        client = Client(wsdl=wsdl, plugins=[history], wsse=binary_signature, settings=client_settings)
        return client, history

    @staticmethod
    def get_zeep_client(
        wsdl, secure: bool = True, key_file: str = "", cert_file: str = ""
    ) -> Tuple[Client, HistoryPlugin]:
        """
        Deprecated: Please use get_client instead
        Returns a client
        Args:
            wsdl:
            key_file:
            cert_file:
            secure:

        Returns:

        """
        return APIClient._get_zeep_client(wsdl, secure, key_file, cert_file)

get_client(environment, service, key_file='', cert_file='') staticmethod

Get a production client and a history plugin

Parameters:

Name	Type	Description	Default
environment	ElHubEnvironment	the environment to use in Elhub	required
service	ElHubService	the service to use in Elhub	required
key_file	str	the key file to use	''
cert_file	str	the certificate file to use	''

Source code in elhub_sdk/client.py

@staticmethod
def get_client(
    environment: ElHubEnvironment, service: ElHubService, key_file: str = "", cert_file: str = ""
) -> Tuple[Client, HistoryPlugin]:
    """
    Get a production client and a history plugin
    Args:
        environment: the environment to use in Elhub
        service: the service to use in Elhub
        key_file: the key file to use
        cert_file: the certificate file to use
    Returns:

    """
    return APIClient._get_zeep_client(
        wsdl=os.path.join(os.path.dirname(__file__), "wsdl", environment, "2.3", "wsdl", service),
        secure=True,
        key_file=key_file,
        cert_file=cert_file,
    )

get_zeep_client(wsdl, secure=True, key_file='', cert_file='') staticmethod

Returns a client

Parameters:

Name	Type	Description	Default
wsdl			required
key_file	str		''
cert_file	str		''
secure	bool		True

Source code in elhub_sdk/client.py

@staticmethod
def get_zeep_client(
    wsdl, secure: bool = True, key_file: str = "", cert_file: str = ""
) -> Tuple[Client, HistoryPlugin]:
    """
    Deprecated: Please use get_client instead
    Returns a client
    Args:
        wsdl:
        key_file:
        cert_file:
        secure:

    Returns:

    """
    return APIClient._get_zeep_client(wsdl, secure, key_file, cert_file)

BinarySignatureTimestamp (BinarySignature)

Add timestamp in signature

Source code in elhub_sdk/client.py

class BinarySignatureTimestamp(BinarySignature):
    """
    Add timestamp in signature
    """

    def apply(self, envelope, headers):
        """
        Apply header
        Args:
            envelope:
            headers:

        Returns:

        """
        security = utils.get_security_header(envelope)

        created = datetime.utcnow()
        expired = created + timedelta(seconds=1 * 60)

        timestamp = utils.WSU('Timestamp')
        timestamp.append(utils.WSU('Created', created.replace(microsecond=0).isoformat() + 'Z'))
        timestamp.append(utils.WSU('Expires', expired.replace(microsecond=0).isoformat() + 'Z'))

        security.append(timestamp)
        super().apply(envelope, headers)
        return envelope, headers

    def verify(self, envelope):
        """
        Args:
            envelope:

        Returns:

        """
        return envelope

apply(self, envelope, headers)

Apply header

Parameters:

Name	Type	Description	Default
envelope			required
headers			required

Source code in elhub_sdk/client.py

def apply(self, envelope, headers):
    """
    Apply header
    Args:
        envelope:
        headers:

    Returns:

    """
    security = utils.get_security_header(envelope)

    created = datetime.utcnow()
    expired = created + timedelta(seconds=1 * 60)

    timestamp = utils.WSU('Timestamp')
    timestamp.append(utils.WSU('Created', created.replace(microsecond=0).isoformat() + 'Z'))
    timestamp.append(utils.WSU('Expires', expired.replace(microsecond=0).isoformat() + 'Z'))

    security.append(timestamp)
    super().apply(envelope, headers)
    return envelope, headers

verify(self, envelope)

Parameters:

Name	Type	Description	Default
envelope			required

Source code in elhub_sdk/client.py

def verify(self, envelope):
    """
    Args:
        envelope:

    Returns:

    """
    return envelope

ElHubEnvironment (str, Enum)

ElHub environment, either prod or test

Source code in elhub_sdk/client.py

class ElHubEnvironment(str, Enum):
    """
    ElHub environment, either prod or test
    """

    PRODUCTION = "prod"
    TEST = "test"

ElHubService (str, Enum)

ElHub services pointing to the correct WSDL file

Source code in elhub_sdk/client.py

class ElHubService(str, Enum):
    """
    ElHub services pointing to the correct WSDL file
    """

    MARKET_PROCESSES = "MarketProcesses - WS-Security.wsdl"
    METERING_VALUES = "MeteringValues - WS-Security.wsdl"
    POOL_MARKET_PROCESSES = "PollMarketProcesses - WS-Security.wsdl"
    POOL_METERING_VALUES = "PollMeteringValues - WS-Security.wsdl"
    QUERY = "Query - WS-Security.wsdl"

constants

Constants used

consumption

Consumption related functions

The data will be made available on the polling service. If the query is related to metering values (applies to the RequestDataFromElhubRequest message and depends on the specified query type), the responses (including Acknowledgement) will be made available on the PollMeteringValues service.

poll_consumption(client, history, sender_gsn, process_role=<ROLES.THIRD_PARTY: 'AG'>)

Poll WSDL

Parameters:

Name	Type	Description	Default
client	Client		required
history	HistoryPlugin		required
sender_gsn	str		required
process_role	ROLES		<ROLES.THIRD_PARTY: 'AG'>

Source code in elhub_sdk/consumption.py

def poll_consumption(
    client: zeep.Client, history: HistoryPlugin, sender_gsn: str, process_role: ROLES = ROLES.THIRD_PARTY
) -> Optional[str]:
    """
    Poll WSDL
    Args:
        client:
        history:
        sender_gsn:
        process_role:

    Returns:

    """

    factory = client.type_factory('ns8')
    request = factory.PollForData(
        Header={
            'Identification': uuid.uuid4(),
            'DocumentType': {
                '_value_1': DOCUMENT_TYPE_UN_CEFACT.QUERY.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'Creation': f'{datetime.utcnow().strftime(TIME_FORMAT)}',
            'PhysicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalRecipientEnergyParty': {
                'Identification': {'_value_1': ELHUB_GSN, 'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value}
            },
        },
        ProcessEnergyContext={
            'EnergyBusinessProcess': {'_value_1': 'POLL', 'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.ELHUB.value},
            'EnergyBusinessProcessRole': {
                '_value_1': process_role.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'EnergyIndustryClassification': ENERGY_INDUSTRY_CLASSIFICATION.ELECTRICITY_SUPPLY.value,
        },
        Payload={},
    )
    try:
        response = client.service.PollForData(request)
        if "ResultDataSet" in response and history.last_received:
            xml_response = ET.tostring(history.last_received["envelope"], encoding="unicode")
            return xml_response
        logger.error(f"Unknown response: {response}")
    except zeep.exceptions.Fault as ex:
        logger.error(f"Bad response: {ex}")

    return None

request_consumption(client, history, meter_identificator, sender_gsn, start, end, process_role=<ROLES.THIRD_PARTY: 'AG'>)

Query WSDL

Parameters:

Name	Type	Description	Default
history	HistoryPlugin		required
client	Client		required
meter_identificator	str		required
sender_gsn	str		required
start	datetime		required
end	datetime		required
process_role	ROLES		<ROLES.THIRD_PARTY: 'AG'>

Source code in elhub_sdk/consumption.py

def request_consumption(
    client: zeep.Client,
    history: HistoryPlugin,
    meter_identificator: str,
    sender_gsn: str,
    start: datetime,
    end: datetime,
    process_role: ROLES = ROLES.THIRD_PARTY,
) -> bool:
    """
    Query WSDL
    Args:
        history:
        client:
        meter_identificator:
        sender_gsn:
        start:
        end:
        process_role:

    Returns:

    """

    factory = client.type_factory('ns4')
    eh_request = factory.RequestDataFromElhub(
        Header={
            'Identification': uuid.uuid4(),
            'DocumentType': {
                '_value_1': DOCUMENT_TYPE_UN_CEFACT.QUERY.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'Creation': f'{datetime.utcnow().strftime(TIME_FORMAT)}',
            'PhysicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalRecipientEnergyParty': {
                'Identification': {'_value_1': ELHUB_GSN, 'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value}
            },
        },
        ProcessEnergyContext={  # https://dok.elhub.no/ediel2/general#General-Process
            'EnergyBusinessProcess': {
                '_value_1': BSR_IDS.METERING_VALUES.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.ELHUB.value,
            },
            'EnergyBusinessProcessRole': {
                '_value_1': process_role.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'EnergyIndustryClassification': "23",
        },
        PayloadMPEvent={
            'QueryTypeCode': QUERY_MARKET.METERING_VALUES_TIME_SERIES.value,
            'Period': {'Start': f'{start.strftime(TIME_FORMAT)}', 'End': f'{end.strftime(TIME_FORMAT)}'},
            'MeteringPointUsedDomainLocation': {
                'Identification': {
                    '_value_1': meter_identificator,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
        },
    )

    try:
        response = client.service.RequestDataFromElhub(eh_request)
        if history.last_received:
            return True
        logger.error(f"Unknown error: {response}")
    except Exception as ex:
        logger.exception(ex)
    return False

enrollment

Processes related to the enrollment of customers

get_meter_characteristics(client, history, meter_identificator, sender_gsn, process_role=<ROLES.THIRD_PARTY: 'AG'>)

Parameters:

Name	Type	Description	Default
client	Client	zeep client	required
history	HistoryPlugin	zeep history	required
meter_identificator	str	meter identificator	required
sender_gsn	str	sender gsn	required
process_role	ROLES	process role	<ROLES.THIRD_PARTY: 'AG'>

Source code in elhub_sdk/enrollment.py

def get_meter_characteristics(
    client: zeep.Client,
    history: zeep.plugins.HistoryPlugin,
    meter_identificator: str,
    sender_gsn: str,
    process_role: ROLES = ROLES.THIRD_PARTY,
) -> Optional[str]:
    """
    https://dok.elhub.no/ediel2/requestupfrontmeteringpointcharacteristics
    Args:
        client: zeep client
        history: zeep history
        meter_identificator: meter identificator
        sender_gsn: sender gsn
        process_role: process role

    Returns:

    """

    factory = client.type_factory('ns1')
    eh_request = factory.RequestUpfrontMeteringPointCharacteristics(
        Header={
            'Identification': uuid.uuid4(),
            'DocumentType': {
                '_value_1': DOCUMENT_TYPE_EBIX.CHANGE_DATA.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.EBIX.value,
            },
            'Creation': f'{datetime.utcnow().strftime(TIME_FORMAT)}',
            'PhysicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalRecipientEnergyParty': {
                'Identification': {'_value_1': ELHUB_GSN, 'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value}
            },
        },
        ProcessEnergyContext={  # https://dok.elhub.no/ediel2/general#General-Process
            'EnergyBusinessProcess': {
                '_value_1': BSR_IDS.METERING_POINT_CHARACTERISTICS.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.ELHUB.value,
            },
            'EnergyBusinessProcessRole': {
                '_value_1': process_role.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'EnergyIndustryClassification': "23",
        },
        PayloadMPEvent={
            'MeteringPointUsedDomainLocation': {
                'Identification': {
                    '_value_1': meter_identificator,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
        },
    )

    try:
        response = client.service.RequestUpfrontMeteringPointCharacteristics(eh_request)
        if history.last_received and "ResponseUpfrontMeteringPointCharacteristics" in response:
            xml_response = ET.tostring(history.last_received["envelope"], encoding="unicode")
            return xml_response
        logger.error(f"Unknown error: {response}")
    except Exception as ex:
        logger.exception(ex)

    return None

enums

Enums, mainly to create SOAP requests taken from various places in Elhub documentation

BSR_IDS (Enum)

Query type https://dok.elhub.no/ediel2/elhub-brs-identifications

Source code in elhub_sdk/enums.py

class BSR_IDS(Enum):
    """
    Query type
    https://dok.elhub.no/ediel2/elhub-brs-identifications
    """

    METERING_VALUES = "BRS-NO-315"
    THIRD_PARTY = "BRS-NO-622"
    METERING_POINT_CHARACTERISTICS = "BRS-NO-611"

DOCUMENT_TYPE_EBIX (Enum)

Metering data: https://dok.elhub.no/ediel2/document-type

Source code in elhub_sdk/enums.py

class DOCUMENT_TYPE_EBIX(Enum):
    """
    Metering data: https://dok.elhub.no/ediel2/document-type
    """

    CHANGE_DATA = "E10"
    METERING_DATA = "E13"
    MASTER_DATA = "E07"

DOCUMENT_TYPE_UN_CEFACT (Enum)

Quaery types

Source code in elhub_sdk/enums.py

class DOCUMENT_TYPE_UN_CEFACT(Enum):
    """
    Quaery types
    """

    QUERY = "21"

ENERGY_INDUSTRY_CLASSIFICATION (Enum)

Industry codes https://dok.elhub.no/ediel2/general

Source code in elhub_sdk/enums.py

class ENERGY_INDUSTRY_CLASSIFICATION(Enum):
    """
    Industry codes
    https://dok.elhub.no/ediel2/general
    """

    ELECTRICITY_SUPPLY = "23"

LIST_AGENCY_IDENTIFIER (Enum)

Agency identifier

Source code in elhub_sdk/enums.py

class LIST_AGENCY_IDENTIFIER(Enum):
    """
    Agency identifier
    """

    EBIX = '260'
    UN_CEFACT = '6'
    ELHUB = '89'

QUERY_MARKET (Enum)

Source code in elhub_sdk/enums.py

class QUERY_MARKET(Enum):
    """
    https://dok.elhub.no/ediel2/5-query-from-market-parties
    """

    METERING_VALUES_TIME_SERIES = "MVTS"

ROLES (Enum)

Roles

https://dok.elhub.no/ediel2/roles-and-domains

Source code in elhub_sdk/enums.py

class ROLES(Enum):
    """
    Roles:
        https://dok.elhub.no/ediel2/roles-and-domains
    """

    THIRD_PARTY = "AG"
    BALANCE_SUPPLIER = "DDQ"

SCHEME_AGENCY_IDENTIFIER (Enum)

Agency identifiers

Source code in elhub_sdk/enums.py

class SCHEME_AGENCY_IDENTIFIER(Enum):
    """
    Agency identifiers
    """

    GS1 = '9'

STATUS_TYPE (Enum)

Status Type ref: https://dok.elhub.no/ediel2/acknowledgement

Source code in elhub_sdk/enums.py

class STATUS_TYPE(Enum):
    """
    Status Type
    ref: https://dok.elhub.no/ediel2/acknowledgement
    """

    ACCEPTED = '39'
    REJECTED = '41'

THIRD_PARTY_ACTION (Enum)

Determines the action you want to do

Source code in elhub_sdk/enums.py

class THIRD_PARTY_ACTION(Enum):
    """
    Determines the action you want to do
    """

    ADD = "Add"
    DELETE = "Delete"
    UPDATE = "Update"

settings

Settings module

third_party

Manages third party requests

It mainly uses the endpoint UpdateThirdPartyAccessRequest

ns7:UpdateThirdPartyAccess(Header: ns2:Elhub_HeaderType, ProcessEnergyContext: ns2:Elhub_EnergyContextType, PayloadMPEvent: ns2:Elhub_UpdateThirdPartyAccessType)

Elhub_UpdateThirdPartyAccessType(

UpdateIndicator: ns3: AddDeleteUpdate, MeteringPointUsedDomainLocation: ns2:Elhub_MeteringPointType, ConsumerInvolvedCustomerParty: ns2:Elhub_EndUserType)

request_action(client, history, sender_gsn, meter_identificator, action, extended_storage=False)

Parameters:

Name	Type	Description	Default
history	HistoryPlugin	zeep history	required
client	Client	zeep client	required
sender_gsn	str	sender gsn	required
meter_identificator	str	meter identificator	required
action	THIRD_PARTY_ACTION	action to perform	required
extended_storage		extended storage	False

Source code in elhub_sdk/third_party.py

def request_action(
    client: zeep.Client,
    history: zeep.plugins.HistoryPlugin,
    sender_gsn: str,
    meter_identificator: str,
    action: THIRD_PARTY_ACTION,
    extended_storage=False,
):
    """

    Args:
        history: zeep history
        client: zeep client
        sender_gsn: sender gsn
        meter_identificator: meter identificator
        action: action to perform
        extended_storage: extended storage
    Returns:

    """

    factory = client.type_factory('ns7')

    playload_event: Dict[str, Any] = {
        'UpdateIndicator': action.value,
        'MeteringPointUsedDomainLocation': {
            'Identification': {
                '_value_1': meter_identificator,
                'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
            }
        },
    }

    if extended_storage:
        playload_event['ConsumerInvolvedCustomerParty'] = {'ExtendedStorageMeteringValues': "true"}

    eh_request = factory.UpdateThirdPartyAccess(
        Header={
            'Identification': uuid.uuid4(),
            'DocumentType': {
                '_value_1': DOCUMENT_TYPE_EBIX.CHANGE_DATA.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.EBIX.value,
            },
            'Creation': f'{datetime.utcnow().strftime(TIME_FORMAT)}',
            'PhysicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalSenderEnergyParty': {
                'Identification': {
                    '_value_1': sender_gsn,
                    'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value,
                }
            },
            'JuridicalRecipientEnergyParty': {
                'Identification': {'_value_1': ELHUB_GSN, 'schemeAgencyIdentifier': SCHEME_AGENCY_IDENTIFIER.GS1.value}
            },
        },
        ProcessEnergyContext={  # https://dok.elhub.no/ediel2/general#General-Process
            'EnergyBusinessProcess': {
                '_value_1': BSR_IDS.THIRD_PARTY.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.ELHUB.value,
            },
            'EnergyBusinessProcessRole': {
                '_value_1': ROLES.THIRD_PARTY.value,
                'listAgencyIdentifier': LIST_AGENCY_IDENTIFIER.UN_CEFACT.value,
            },
            'EnergyIndustryClassification': "23",
        },
        PayloadMPEvent=playload_event,
    )

    try:
        response = client.service.UpdateThirdPartyAccess(eh_request)
        return response
    except Exception as ex:
        logger.exception(ex)
    return False