[DEV 3374] Add example for feeder load analysis

changelog.md

@@ -51,15 +51,17 @@
   * Translating a CIM network model into a pandapower model
   * Requesting a PowerFactory model through the SDK
   * Manipulating the current state of the network, including swapping a zone open point.
+  * Added Example for requesting feeder load analysis study through EAS client.
 
 ### Enhancements
 * Limited power factory demo to 1 job at a time.
 * Added model download function to power factory demo
 * restrict installation to supported Python versions from 3.9 to 3.11
+* update request power factory models to use new authentication method
 
 ### Fixes
 * None.
 
 ### Notes
-* Support `zepben.eas` up to 0.16.0.
-* Support `zepben.evolve` up to 0.44.0.
+* Support `zepben.eas` up to 0.19.0.
+* Support `zepben.evolve` up to 0.48.0.

src/zepben/examples/request_feeder_load_analysis_study.py

@@ -0,0 +1,46 @@
+#  Copyright 2025 Zeppelin Bend Pty Ltd
+#
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+import asyncio
+import sys
+
+from zepben.eas.client.feeder_load_analysis_input import FeederLoadAnalysisInput
+
+from zepben.examples.utils import get_config_dir, get_client
+
+
+async def main(argv):
+    # See connecting_to_grpc_service.py for examples of each connect function
+    config_dir = get_config_dir(argv)
+    print("Connecting to EAS..")
+    eas_client = get_client(config_dir)
+    print("Connection established..")
+    # Fire off a feeder load analysis study
+    feeder_load_analysis_token = await eas_client.async_run_feeder_load_analysis_report(
+        FeederLoadAnalysisInput(
+            feeders=["feeder1", "feeder2"],
+            substations=None,
+            sub_geographical_regions=None,
+            geographical_regions=None,
+            start_date="2022-04-01",
+            end_date="2022-12-31",
+            fetch_lv_network=True,
+            process_feeder_loads=True,
+            process_coincident_loads=True,
+            aggregate_at_feeder_level=False,
+            output="Test"
+        )
+    )
+
+    print(f"Feeder Load Analysis study: {feeder_load_analysis_token['data']['runFeederLoadAnalysis']}")
+
+    # Feeder Load Analysis Study results can be retrieved from back end storage set up with EAS.
+
+    await eas_client.aclose()
+
+
+if __name__ == "__main__":
+    loop = asyncio.get_event_loop()
+    loop.run_until_complete(main(sys.argv))

src/zepben/examples/request_power_factory_models.py

@@ -34,11 +34,8 @@
 # graphQL endpoint access settings
 network_endpoint = 'https://{url}/api/network/graphql'
 api_endpoint = 'https://{url}/api/graphql'
-audience = "https://{url}/"
-issuer = "issuer_domain"
-client_id = 'client_id'
-username = 'username'
-password = 'password'
+#Generate a personal access token from the energy work bench UI and replace <Token>.
+token = 'Bearer <Token>'
 
 ### EXAMPLE QUERY ONLY ###
 # This is an example GraphQL query for the full network hierarchy. This is not used as part of this
@@ -70,10 +67,8 @@
     }
 }
 '''
-token_fetcher = get_token_fetcher(audience=audience, issuer=issuer, client_id=client_id, username=username, password=password)
-tft = token_fetcher.fetch_token()
-network_transport = RequestsHTTPTransport(url=network_endpoint, headers={'Authorization': tft})
-api_transport = RequestsHTTPTransport(url=api_endpoint, headers={'Authorization': tft})
+network_transport = RequestsHTTPTransport(url=network_endpoint, headers={'Authorization': token})
+api_transport = RequestsHTTPTransport(url=api_endpoint, headers={'Authorization': token})
 
 network_client = Client(transport=network_transport)
 api_client = Client(transport=api_transport)
@@ -261,7 +256,7 @@ def download_model(model_number):
     model_status = result['powerFactoryModelById']['state']
     match model_status:
         case 'COMPLETED':
-            model = requests.get(model_url, headers={'Authorization': tft})
+            model = requests.get(model_url, headers={'Authorization': token})
             open(os.path.join(output_dir, file_name) + ".pfd", 'wb').write(model.content)
             print(file_name + ".pfd saved at " + output_dir)
         case "CREATION":

src/zepben/examples/sample_auth_config.json

@@ -0,0 +1,11 @@
+{
+  "eas_server": {
+    "host": "<URL>",
+    "port": 1234,
+    "protocol": "https",
+    "client_id": "<EAS Client ID>",
+    "access_token": "<Access Token>",
+    "verify_certificate": false,
+    "ca_filename": null
+  }
+}

src/zepben/examples/utils.py

@@ -0,0 +1,39 @@
+#  Copyright 2025 Zeppelin Bend Pty Ltd
+#
+#  This Source Code Form is subject to the terms of the Mozilla Public
+#  License, v. 2.0. If a copy of the MPL was not distributed with this
+#  file, You can obtain one at https://mozilla.org/MPL/2.0/.
+import json
+from typing import Dict
+
+from zepben.eas.client.eas_client import EasClient
+
+import logging
+logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s: %(message)s')
+logger = logging.getLogger()
+
+
+# Default config dir is where the sample_auth_config sits.
+def get_config_dir(argv):
+    return argv[1] if len(argv) > 1 else "."
+
+
+def read_json_config(config_file_path: str) -> Dict:
+    file = open(config_file_path)
+    config_dict = json.load(file)
+    file.close()
+    return config_dict
+
+
+def get_client(config_dir):
+    # Change sample_auth_config.json to any other file name
+    auth_config = read_json_config(f"{config_dir}/sample_auth_config.json")
+
+    return EasClient(
+        host=auth_config["eas_server"]["host"],
+        port=auth_config["eas_server"]["port"],
+        protocol=auth_config["eas_server"]["protocol"],
+        access_token=auth_config["eas_server"]["access_token"],
+        verify_certificate=auth_config["eas_server"].get("verify_certificate", True),
+        ca_filename=auth_config["eas_server"].get("ca_filename")
+    )