Shapeshifter library for Python

This is a Python implementation of the ShapeShifter UFTP protocol.

pip install shapeshifter-uftp

$ pip install .
$ pip install .[dev]
$ pytest .

Running the Shapeshifter Service

The shapeshifter protocol requires you to run a server that can receive HTTP POST requests. You have the following options.

• If you already have a FastAPI application and you’d like to run the Shapeshifter service inside that application, see Running in FastAPI.

• If you already have a Flask application and you’d like to run the Shapeshifter service inside that, see Running in Flask.

• If you don’t have an existing application, or you want to run the Shapeshifter interface stand alone (perhaps inside its own Docker container), see Running stand-alone. Since the service is implemented as a standard WSGI service, you can choose several web servers like nginx, apache2 or gunicorn, which all support WSGI interfaces.

from a2wsgi import WSGIMiddleware
from fastapi import FastAPI

from shapeshifter_uftp.service import ShapeshifterAgrService


class MyShapeshifterService(ShapeshifterAgrService):
    # Your implementation of the service goes here, see next section
    ...


# Create an instance of the Shapeshifter Service
shapeshifter_service = MyShapeshifterService(..., path="/api/v3/message")

# Your existing FastAPI application
app = FastAPI()

# Mount the Shapeshifte service inside your existing FastAPI application
app.mount("/shapeshifter", shapeshifter_service.app)

...

from flask import Flask, DispatcherMiddleware

from shapeshifter_uftp.service import ShapeshifterAgrService


class MyShapeshifterService(ShapeshifterAgrService):
    # Your implementation of the service goes here, see next section
    ...

my_app = Flask(__name__)

shapeshifter_service = MyShapeshifterService(..., path="/api/v3/message")

app = DispatcherMiddleware(
    my_app,
    {"/shapeshifter": shapeshifter_service}
)

from shapeshifter_uftp.service import ShapeshifterAgrService


class MyShapeshifterService(ShapeshifterAgrService):
    # Your implementation of the service goes here, see next section
    ...

service = MyShapeshifterService(...)
service.run()

from shapeshifter_uftp.service import ShapeshifterAgrService

class MyShapeshifterService(ShapeshifterAgrService):
    # Your implementation of the service goes here, see next section
    ...

service = MyShapeshifterService(..., path="/api/v3/message")

<VirtualHost *:80>

    ServerName www.example.com

    WSGIPythonHome /app/.venv/bin/python
    WSGIScriptAlias /shapeshifter /app/main.py
    WSGICallableObject service

    <Directory /app>
        Require all granted
    </Directory>

</VirtualHost>

# main.py
from shapeshifter_uftp.service import ShapeshifterAgrService

class MyShapeshifterService(ShapeshifterAgrService):
    # Your implementation of the service goes here, see next section
    ...

service = MyShapeshifterService(...)

gunicorn main:service

Implementing your Shapeshifter Service

Shapehifter always requires the use of a Client and a Service, because all responses are asynchronous.

You choose the server class based on your role in the Shapeshifter conversation, and subclass the abstract implementation of the appropriate service. You then add implementations for each of the messages that you can receive, in order to post-process them according to you own business logic.

This is an overview of the roles, the base class, and the methods you must implement:

If you are an Aggregator (also known as a CSP), you can use this setup:

from datetime import datetime, timedelta, timezone

from shapeshifter_uftp import ShapeshifterAgrService
from shapeshifter_uftp.uftp import (FlexOffer, FlexOfferOption,
                                    FlexOfferOptionISP, FlexRequest,
                                    FlexRequestResponse, FlexOrder, FlexOrderResponse,
                                    AcceptedRejected)
from xsdata.models.datatype import XmlDate


class DemoAggregator(ShapeshifterAgrService):
    """
    Aggregator service that implements callbacks for
    each of the messages that can be received.
    """

    def process_agr_portfolio_query_response(self, message):
        print(f"Received a message: {message}")

    def process_agr_portfolio_update_response(self, message):
        print(f"Received a message: {message}")

    def process_d_prognosis_response(self, message):
        print(f"Received a message: {message}")

    def process_flex_request(self, message: FlexRequest):
        print(f"Received a message: {message}")

        # Example of how to send a new message after
        # processing an incoming message.
        dso_client = self.dso_client(message.sender_domain)

        # Send the FlexRequestResponse
        dso_client.send_flex_request_response(
            FlexRequestResponse(
                flex_request_message_id=message.message_id,
                conversation_id=message.conversation_id,
                result=AcceptedRejected.ACCEPTED
            )
        )

        # Send the FlexOffer
        dso_client.send_flex_offer(
            FlexOffer(
                flex_request_message_id=message.message_id,
                conversation_id=message.conversation_id,
                isp_duration="PT15M",
                period=XmlDate(2023, 1, 1),
                congestion_point="ean.123456789012",
                expiration_date_time=datetime.now(timezone.utc).isoformat(),
                offer_options=[
                    FlexOfferOption(
                        isps=[FlexOfferOptionISP(power=1, start=1, duration=1)],
                        option_reference="MyOption",
                        price=2.30,
                        min_activation_factor=0.5,
                    )
                ],
            )
        )

    def process_flex_offer_response(self, message: FlexOffer):
        print(f"Received a message: {message}")

    def process_flex_offer_revocation_response(self, message):
        print(f"Received a message: {message}")

    def process_flex_order(self, message: FlexOrder):
        print(f"Received a message: {message}")

        dso_client = self.dso_client(message.sender_domain)
        dso_client.send_flex_order_response(
            FlexOrderResponse(
                flex_order_message_id=message.message_id,
                conversation_id=message.conversation_id,
                result=AcceptedRejected.ACCEPTED
            )
        )

    def process_flex_reservation_update(self, message):
        print(f"Received a message: {message}")

    def process_flex_settlement(self, message):
        print(f"Received a message: {message}")

    def process_metering_response(self, message):
        print(f"Received a message: {message}")


def key_lookup(sender_domain, sender_role):
    """
    Lookup function for public keys, so that incoming
    messages can be verified.
    """
    known_senders = {
        ("dso.demo", "DSO"): "NsTbq/iABU6tbsjriBg/Z5dSfQstulD0GpMI2fLDWec=",
        ("cro.demo", "CRO"): "ySUYU87usErRFKGJafwvVDLGhnBVJCCNYfQvmwv8ObM=",
    }
    return known_senders.get((sender_domain, sender_role))


def endpoint_lookup(sender_domain, sender_role):
    """
    Lookup function for endpoints, so that the service
    knowns where to send responses to.
    """
    known_senders = {
        ("dso.demo", "DSO"): "http://localhost:8081/shapeshifter/api/v3/message",
        ("cro.demo", "CRO"): "http://localhost:8082/shapeshifter/api/v3/message",
    }
    return known_senders.get((sender_domain, sender_role))

aggregator = DemoAggregator(
    sender_domain="aggregator.demo",
    signing_key="mz5XYCNKxpx48K+9oipUhsjBZed3L7rTVKLsWmG1HOqRLIeuGpIa1KAt6AlbVGqJvewd8v1J0uVUTqpGt7F8tw==",
    key_lookup_function=key_lookup,
    endpoint_lookup_function=endpoint_lookup,
    port=8080,
)

# Start the Aggregator Service
aggregator.run_in_thread()

# Create a client object to talk to a DSO
dso_client = aggregator.dso_client("dso.demo")

# Create a Flex Offer Message
flex_offer_message = FlexOffer(
    isp_duration="PT15M",
    period=XmlDate(2023, 1, 1),
    congestion_point="ean.123456789012",
    expiration_date_time=datetime.now(timezone.utc).isoformat(),
    flex_request_message_id=str(uuid4())
    offer_options=[
        FlexOfferOption(
            isps=[FlexOfferOptionISP(power=1, start=1, duration=1)],
            option_reference="MyOption",
            price=2.30,
            min_activation_factor=0.5,
        )
    ],
)

# As a demo, press enter to send another FlexOffer message to the DSO.
while True:
    try:
        input("Press return to send a FlexOffer message to the DSO")
        response = dso_client.send_flex_offer(flex_offer_message)
        print(f"Response was: {response}")
    except:
        aggregator.stop()
        break

from shapeshifter_uftp import ShapeshifterAgrDsoClient, OAuthClient

oauth_client = OAuthClient(
    url="https://oauth.provider.url",
    client_id="my-client-id",
    client_secret="my-client-secret"
)

client = ShapeshifterAgrDsoClient(
    sender_domain="my.aggregator.domain",
    signing_key="abcdef",
    recipient_domain="some.dso",
    recipient_endpoint="https://some.dso.endpoint/shapeshifter/api/v3/message",
    recipient_signing_key="123456",
    oauth_client=oauth_client,
)

# If you use any of the sending methods, the oauth client will
# make sure you're authenticated.
client.send_flex_request_response(...)