Clean up orphaned transactions

If a CP is reset during a transaction, the system will end up with an orphaned transaction that is not closed.

This middleware closes any previous orphaned transactions for that CP, using the last seen meter value for tx.meter_stop

README.md

@@ -2,7 +2,7 @@ Levity: An extensible OCPP server and EVSE management platform.
 
 # Install
 
-Levity runs well on a t4a.small instance (2GB RAM)
+Levity runs well on a t4a.small instance (2GB of RAM)
 
 ## Full install (non-TLS)
 

be/ocpp/models/transaction.py

@@ -2,6 +2,7 @@
 
 from ocpp.models.charge_point import ChargePoint
 from ocpp.types.stop_reason import StopReason
+from ocpp.utils.date import utc_now
 
 
 class Transaction(models.Model):
@@ -15,3 +16,12 @@ class Transaction(models.Model):
     stop_reason = models.CharField(
         max_length=64, choices=StopReason.choices(), null=True, blank=True
     )
+
+    def stop(self, reason: StopReason, meter_stop: int):
+        self.meter_stop = meter_stop
+        self.stop_reason = reason
+        self.stopped_at = utc_now()
+        self.save(update_fields=["meter_stop", "stop_reason", "stopped_at"])
+        charge_point = self.charge_point
+        charge_point.last_tx_stop_at = utc_now()
+        charge_point.save(update_fields=["last_tx_stop_at"])

be/ocpp/services/charge_point_message_handler.py

@@ -4,6 +4,9 @@
 from ocpp.models.message import Message
 from ocpp.services.charge_point_service import ChargePointService
 from ocpp.services.ocpp.anon.auto_remote_start import AutoRemoteStartMiddleware
+from ocpp.services.ocpp.automation.orphaned_transaction import (
+    OrphanedTransactionMiddleware,
+)
 from ocpp.services.ocpp.base import ResponseMiddleware, OCPPRequest
 from ocpp.services.ocpp.core.authorize import AuthorizeMiddleware
 from ocpp.services.ocpp.core.boot_notification import BootNotificationMiddleware
@@ -37,7 +40,10 @@
     ],
     (Action.Heartbeat, MessageType.call): [HeartbeatMiddleware],
     (Action.MeterValues, MessageType.call): [MeterValuesMiddleware],
-    (Action.StartTransaction, MessageType.call): [StartTransactionMiddleware],
+    (Action.StartTransaction, MessageType.call): [
+        OrphanedTransactionMiddleware,
+        StartTransactionMiddleware,
+    ],
     (Action.StatusNotification, MessageType.call): [
         AutoRemoteStartMiddleware,
         StatusNotificationMiddleware,

be/ocpp/services/ocpp/automation/orphaned_transaction.py

@@ -0,0 +1,37 @@
+import logging
+
+from ocpp.models import Transaction
+from ocpp.services.ocpp.base import OCPPMiddleware, OCPPRequest, OCPPResponse
+from ocpp.types.action import Action
+from ocpp.types.stop_reason import StopReason
+
+logger = logging.getLogger(__name__)
+
+
+class OrphanedTransactionMiddleware(OCPPMiddleware):
+    """
+    When a new transaction is started, close out any unclosed previous transactions for the same CP
+    """
+
+    def handle(self, req: OCPPRequest) -> OCPPResponse:
+        message = req.message
+        assert Action(message.action) == Action.StartTransaction
+        for orphaned_tx in Transaction.objects.filter(
+            charge_point=message.charge_point, stopped_at__isnull=True
+        ):
+            last_meter_value = (
+                orphaned_tx.metervalue_set.filter(
+                    measurand="Energy.Active.Import.Register"
+                )
+                .order_by("timestamp")
+                .last()
+            )
+            meter_stop = last_meter_value.value if last_meter_value else 0
+            orphaned_tx.stop(StopReason.Other, meter_stop)
+            logger.info(
+                "Stopped orphaned transaction %s with meter value %d",
+                orphaned_tx,
+                meter_stop,
+            )
+
+        return self.next.handle(req)

be/ocpp/services/ocpp/core/stop_transaction.py

@@ -2,26 +2,19 @@
 from ocpp.services.ocpp.base import OCPPMiddleware, OCPPRequest, OCPPResponse
 from ocpp.types.authorization_status import AuthorizationStatus
 from ocpp.types.stop_reason import StopReason
-from ocpp.utils.date import utc_now
 
 
 class StopTransactionMiddleware(OCPPMiddleware):
     def handle(self, req: OCPPRequest) -> OCPPResponse:
         message = req.message
         transaction = message.transaction_from_data()
-        transaction.meter_stop = message.data["meterStop"]
-        transaction.stop_reason = StopReason(message.data["reason"])
-        transaction.stopped_at = utc_now()
-        transaction.save(update_fields=["meter_stop", "stop_reason", "stopped_at"])
+        transaction.stop(StopReason(message.data["reason"]), message.data["meterStop"])
         transaction_data = message.data.get("transactionData") or []
         for meter_value in transaction_data:
             for sampled_value in meter_value["sampledValue"]:
                 MeterValue.create_from_json(
                     transaction, meter_value["timestamp"], sampled_value, is_final=True
                 )
-        charge_point = message.charge_point
-        charge_point.last_tx_stop_at = utc_now()
-        charge_point.save(update_fields=["last_tx_stop_at"])
         res = self.next.handle(req)
         res.message.data.update(
             dict(idTagInfo=dict(status=AuthorizationStatus.Accepted)),

be/ocpp/tests/factory.py

@@ -1,8 +1,9 @@
 import factory
 from factory.django import DjangoModelFactory
 
-from ocpp.models import ChargePoint
+from ocpp.models import ChargePoint, Transaction, MeterValue
 from ocpp.types.charge_point_status import ChargePointStatus
+from ocpp.utils.date import utc_now
 
 
 class ChargePointFactory(DjangoModelFactory):
@@ -28,3 +29,30 @@ class ChargePointFactory(DjangoModelFactory):
 
     class Meta:
         model = ChargePoint
+
+
+class TransactionFactory(DjangoModelFactory):
+    charge_point = factory.SubFactory(ChargePointFactory)
+    connector_id = "1"
+    id_tag = "anonymous"
+    meter_start = 0
+    meter_stop = 0
+
+    class Meta:
+        model = Transaction
+
+
+class MeterValueFactory(DjangoModelFactory):
+    timestamp = utc_now()
+    transaction = factory.SubFactory(TransactionFactory)
+    context = ""
+    format = ""
+    location = ""
+    measurand = ""
+    phase = ""
+    unit = ""
+    value = 0
+    is_final = False
+
+    class Meta:
+        model = MeterValue

be/ocpp/tests/services/ocpp/automation/test_orphaned_transaction.py

@@ -0,0 +1,60 @@
+from datetime import timedelta
+from unittest.mock import patch
+
+from django.test import TestCase
+
+from ocpp.models import Message, Transaction
+from ocpp.services.charge_point_message_handler import ChargePointMessageHandler
+from ocpp.tests.factory import ChargePointFactory, TransactionFactory, MeterValueFactory
+from ocpp.utils.date import utc_now
+
+
+@patch(
+    "ocpp.services.charge_point_service.ChargePointService.send_message_to_charge_point"
+)
+class OrphanedTransactionTest(TestCase):
+    def setUp(self) -> None:
+        self.charge_point = ChargePointFactory()
+
+    def test_auto_remote_start(self, send_message_to_charge_point):
+        orphaned_tx = TransactionFactory(
+            charge_point=self.charge_point, started_at=utc_now()
+        )
+        MeterValueFactory(
+            timestamp=utc_now() - timedelta(minutes=1),
+            transaction=orphaned_tx,
+            measurand="Energy.Active.Import.Register",
+            value=5,
+        )
+        MeterValueFactory(
+            timestamp=utc_now(),
+            transaction=orphaned_tx,
+            measurand="Energy.Active.Import.Register",
+            value=10,
+        )
+        message = Message.from_occp(
+            self.charge_point,
+            dict(
+                message=[
+                    2,
+                    "x",
+                    "StartTransaction",
+                    {
+                        "idTag": "x",
+                        "timestamp": "2023-03-30T01:58:52.001Z",
+                        "connectorId": 1,
+                        "meterStart": 0,
+                    },
+                ]
+            ),
+        )
+        ChargePointMessageHandler.handle_message_from_charge_point(message)
+        self.charge_point.refresh_from_db()
+        orphaned_tx.refresh_from_db()
+        assert orphaned_tx.stopped_at
+        assert orphaned_tx.meter_stop == 10
+
+        # make sure it doesn't affect the new transaction
+        assert Transaction.objects.filter(
+            charge_point=self.charge_point, stopped_at__isnull=True, meter_stop=0
+        ).exists()