Apply ruff formatting to Python source and test files

gridappsd-field-bus-lib/gridappsd_field_bus/field_interface/field_proxy_forwarder.py

@@ -6,7 +6,8 @@
 
 try:
     from importlib.metadata import version as _pkg_version
-    _STOMP_V8 = int(_pkg_version('stomp-py').split('.')[0]) >= 8
+
+    _STOMP_V8 = int(_pkg_version("stomp-py").split(".")[0]) >= 8
 except Exception:
     _STOMP_V8 = False
 
@@ -102,8 +103,9 @@ def __init__(self, connection_url: str, username: str, password: str, mrid: str)
         self.network.get_all_edges(cim.Substation)
 
         for substation in self.network.graph.get(cim.Substation, {}).values():
-            print(f"Subscribing to Substation: /topic/goss.gridappsd.field.{substation.mRID}")
-            self.ot_connection.subscribe("/topic/goss.gridappsd.field." + substation.mRID, self.on_message_from_ot)
+            mrid = substation.mRID  # type: ignore[attr-defined]
+            print(f"Subscribing to Substation: /topic/goss.gridappsd.field.{mrid}")
+            self.ot_connection.subscribe("/topic/goss.gridappsd.field." + mrid, self.on_message_from_ot)
 
         # self.ot_connection.subscribe(topics.BASE_FIELD_TOPIC, self.on_message_from_ot)
 

gridappsd-python-lib/gridappsd/goss.py

@@ -67,10 +67,15 @@
 _log: Logger = logging.getLogger(inspect.getmodulename(__file__))
 
 # stomp.py 8.x changed listener callbacks from (headers, body) to (frame)
-_stomp_major = int(getattr(_stomp_module, '__version__', (0,))[0]) if isinstance(getattr(_stomp_module, '__version__', None), tuple) else 0
+_stomp_major = (
+    int(getattr(_stomp_module, "__version__", (0,))[0])
+    if isinstance(getattr(_stomp_module, "__version__", None), tuple)
+    else 0
+)
 try:
     from importlib.metadata import version as _pkg_version
-    _stomp_major = int(_pkg_version('stomp-py').split('.')[0])
+
+    _stomp_major = int(_pkg_version("stomp-py").split(".")[0])
 except Exception:
     pass
 _STOMP_V8 = _stomp_major >= 8

gridappsd-python-lib/tests/test_simulation.py

@@ -1,64 +1,65 @@
-#import json
-#import logging
-#import os
-#import sys
-#import time
-#import pytest
-#from datetime import datetime, timezone
+import json
+import logging
+import os
+import sys
+import time
+import pytest
+from datetime import datetime, timezone
 
-#logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
 
-#from gridappsd import GridAPPSD, topics as t
-#from gridappsd.simulation import Simulation, PowerSystemConfig, SimulationArgs, SimulationConfig
+from gridappsd import GridAPPSD, topics as t
+from gridappsd.simulation import Simulation, PowerSystemConfig, SimulationArgs, SimulationConfig
 
-#simulation_is_complete = False
-#measurements_received = 0
+simulation_is_complete = False
+measurements_received = 0
 
-#@pytest.fixture
-#def createGadObject():
-#    gad_user = os.environ.get('GRIDAPPSD_USER')
-#    if gad_user is None:
-#        os.environ['GRIDAPPSD_USER'] = 'system'
-#    gad_password = os.environ.get('GRIDAPPSD_PASSWORD')
-#    if gad_password is None:
-#        os.environ['GRIDAPPSD_PASSWORD'] = 'manager'
-#    return GridAPPSD()
+@pytest.fixture
+def createGadObject():
+   gad_user = os.environ.get('GRIDAPPSD_USER')
+   if gad_user is None:
+       os.environ['GRIDAPPSD_USER'] = 'system'
+   gad_password = os.environ.get('GRIDAPPSD_PASSWORD')
+   if gad_password is None:
+       os.environ['GRIDAPPSD_PASSWORD'] = 'manager'
+   return GridAPPSD()
 
-#def test_createSimulations(createGadObject):
-#    gadObj = createGadObject
-#    response = gadObj.query_model_info()
-#    models = response.get("data", {}).get("models", {})
-#    start_time = int(datetime(year=2025, month=1, day=1, hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc).timestamp())
-#    simulationArgs = SimulationArgs(start_time=f"{start_time}",
-#                                    duration="120",
-#                                    run_realtime=False,
-#                                    pause_after_measurements=False)
-#    sim_config = SimulationConfig(simulation_config=simulationArgs)
-#    modelsToRun = [
-#        "49AD8E07-3BF9-A4E2-CB8F-C3722F837B62", # IEEE 13 Node Test Feeder
-#        "C1C3E687-6FFD-C753-582B-632A27E28507"  # IEEE 123 Node Test Feeder
-#    ]
-#    for m in models:
-#        if m.get("modelId") not in modelsToRun:
-#            continue
-#        line_name = m.get("modelId")
-#        subregion_name = m.get("subRegionId")
-#        region_name = m.get("regionId")
-#        psc = PowerSystemConfig(Line_name=line_name,
-#                                SubGeographicalRegion_name=subregion_name,
-#                                GeographicalRegion_name=region_name)
-#        sim_config.power_system_configs.append(psc)
-#    sim_obj = Simulation(gapps=gadObj, run_config=sim_config)
-#    def on_measurement(sim, ts, m):
-#        global measurements_received
-#        measurements_received += 1
-#    def on_simulation_complete(sim):
-#        global simulation_is_complete
-#        simulation_is_complete = True
-#    sim_obj.add_onmeasurement_callback(on_measurement)
-#    sim_obj.add_oncomplete_callback(on_simulation_complete)
-#    sim_obj.start_simulation()
-#    while not simulation_is_complete:
-#        time.sleep(1)
-#    assert measurements_received == 1
-#    gadObj.disconnect()
+@pytest.mark.integration
+def test_createSimulations(createGadObject):
+   gadObj = createGadObject
+   response = gadObj.query_model_info()
+   models = response.get("data", {}).get("models", {})
+   start_time = int(datetime(year=2025, month=1, day=1, hour=0, minute=0, second=0, microsecond=0, tzinfo=timezone.utc).timestamp())
+   simulationArgs = SimulationArgs(start_time=f"{start_time}",
+                                   duration="120",
+                                   run_realtime=False,
+                                   pause_after_measurements=False)
+   sim_config = SimulationConfig(simulation_config=simulationArgs)
+   modelsToRun = [
+       "49AD8E07-3BF9-A4E2-CB8F-C3722F837B62", # IEEE 13 Node Test Feeder
+       "C1C3E687-6FFD-C753-582B-632A27E28507"  # IEEE 123 Node Test Feeder
+   ]
+   for m in models:
+       if m.get("modelId") not in modelsToRun:
+           continue
+       line_name = m.get("modelId")
+       subregion_name = m.get("subRegionId")
+       region_name = m.get("regionId")
+       psc = PowerSystemConfig(Line_name=line_name,
+                               SubGeographicalRegion_name=subregion_name,
+                               GeographicalRegion_name=region_name)
+       sim_config.power_system_configs.append(psc)
+   sim_obj = Simulation(gapps=gadObj, run_config=sim_config)
+   def on_measurement(sim, ts, m):
+       global measurements_received
+       measurements_received += 1
+   def on_simulation_complete(sim):
+       global simulation_is_complete
+       simulation_is_complete = True
+   sim_obj.add_onmeasurement_callback(on_measurement)
+   sim_obj.add_oncomplete_callback(on_simulation_complete)
+   sim_obj.start_simulation()
+   while not simulation_is_complete:
+       time.sleep(1)
+   assert measurements_received == 1
+   gadObj.disconnect()

pixi.toml

@@ -84,8 +84,8 @@ py314 = { features = ["dev", "py314"], solve-group = "py314" }
 # ──────────────────────────────────────────────────────────────────────────────
 # Testing
 # ──────────────────────────────────────────────────────────────────────────────
-test = { cmd = "pytest gridappsd-python-lib/tests -v", description = "Run main library tests" }
-test-cov = { cmd = "pytest gridappsd-python-lib/tests -v --cov=gridappsd --cov-report=term-missing --cov-report=html", description = "Run tests with coverage" }
+test = { cmd = "pytest gridappsd-python-lib/tests -v -m 'not integration'", description = "Run main library tests" }
+test-cov = { cmd = "pytest gridappsd-python-lib/tests -v -m 'not integration' --cov=gridappsd --cov-report=term-missing --cov-report=html", description = "Run tests with coverage" }
 test-field-bus = { cmd = "pytest gridappsd-field-bus-lib/tests -v", description = "Run field bus tests" }
 test-all = { depends-on = ["test", "test-field-bus"], description = "Run all tests" }
 test-integration = { cmd = "pytest gridappsd-python-lib/tests -v -m integration", description = "Run integration tests only" }