Nobil data source

Author: johan12345

src/evmap_backend/api.py

@@ -7,6 +7,8 @@
 from ninja.orm import register_field
 
 from evmap_backend.chargers.models import ChargingSite
+from evmap_backend.data_sources import UpdateMethod
+from evmap_backend.data_sources.models import UpdateState
 from evmap_backend.data_sources.registry import get_data_source
 
 api = NinjaAPI(urls_namespace="evmap")
@@ -35,7 +37,7 @@ def sites(request, sw_lat: float, sw_lng: float, ne_lat: float, ne_lng: float):
 @api.post("/push/{data_source}")
 def push(request, data_source: str):
     data_source = get_data_source(data_source)
-    if not data_source.supports_push:
+    if not UpdateMethod.HTTP_PUSH in data_source.supported_update_methods:
         raise ValueError("Data source does not support push")
 
     data_source.verify_push(request)
@@ -47,4 +49,6 @@ def push(request, data_source: str):
         body = gzip.decompress(body)
 
     data_source.process_push(body)
+
+    UpdateState(data_source=data_source.id, push=True).save()
     logging.info(f"Successfully processed push for {data_source}")

src/evmap_backend/chargers/models.py

@@ -75,6 +75,7 @@ class ConnectorTypes(models.TextChoices):
         TYPE_3A = "Type 3A", "Type 3A"
         TYPE_3C = "Type 3C", "Type 3C"
         CHADEMO = "CHAdeMO", "CHAdeMO"
+        MCS = "MCS", "MCS"
 
         SCHUKO = "Schuko", "Schuko (Type F)"
         # TODO: add other household plugs

src/evmap_backend/data_sources/__init__.py

@@ -25,6 +25,20 @@ class DataType(Enum):
     """
 
 
+class UpdateMethod(Enum):
+    PULL = 1
+    """Data can be retrieved by calling load_data from a cron job"""
+
+    HTTP_PUSH = 2
+    """
+    Data source an call our server at /push/{data_source_id} to push new data.
+    verify_push and process_push have to be implemented.
+    """
+
+    BACKGROUND_SERVICE = 3
+    """A background service needs to run continuously (e.g., with an MQTT connection) to receive updates"""
+
+
 class DataSource(ABC):
     @property
     @abstractmethod
@@ -38,7 +52,7 @@ def supported_data_types(self) -> List[DataType]:
 
     @property
     @abstractmethod
-    def supports_push(self) -> bool:
+    def supported_update_methods(self) -> List[UpdateMethod]:
         pass
 
     @abstractmethod

src/evmap_backend/data_sources/admin.py

@@ -0,0 +1,13 @@
+from django.contrib import admin
+
+from evmap_backend.data_sources.models import UpdateState
+
+
+class UpdateStateAdmin(admin.ModelAdmin):
+    readonly_fields = ["data_source", "last_update", "push"]
+    list_display = ["data_source", "last_update", "push"]
+    list_filter = ["push"]
+
+
+# Register your models here.
+admin.site.register(UpdateState, UpdateStateAdmin)

src/evmap_backend/data_sources/datex2/source.py

@@ -9,7 +9,7 @@
 from cryptography.x509.verification import PolicyBuilder, Store
 from django.http import HttpRequest
 
-from evmap_backend.data_sources import DataSource, DataType
+from evmap_backend.data_sources import DataSource, DataType, UpdateMethod
 from evmap_backend.data_sources.datex2.parser.json import Datex2JsonParser
 from evmap_backend.data_sources.datex2.parser.xml import Datex2XmlParser
 from evmap_backend.settings import BASE_DIR
@@ -22,8 +22,8 @@ def supported_data_types(self) -> List[DataType]:
         return [DataType.STATIC]
 
     @property
-    def supports_push(self) -> bool:
-        return False
+    def supported_update_methods(self) -> List[UpdateMethod]:
+        return [UpdateMethod.PULL]
 
     @abstractmethod
     def get_data(self) -> str:
@@ -56,8 +56,8 @@ def _parse_data(self, root: str):
 
 class BaseMobilithekDatex2DataSource(BaseDatex2DataSource):
     @property
-    def supports_push(self) -> bool:
-        return True
+    def supported_update_methods(self) -> List[UpdateMethod]:
+        return [UpdateMethod.PULL, UpdateMethod.HTTP_PUSH]
 
     def verify_push(self, request: HttpRequest):
         if "X-Forwarded-Client-Cert" not in request.headers:

src/evmap_backend/data_sources/eliso/source.py

@@ -6,7 +6,7 @@
 from tqdm import tqdm
 
 from evmap_backend.chargers.models import Chargepoint, ChargingSite, Connector
-from evmap_backend.data_sources import DataSource, DataType
+from evmap_backend.data_sources import DataSource, DataType, UpdateMethod
 from evmap_backend.sync import sync_chargers
 
 API_URL = "https://mobilithek.info:8443/mobilithek/api/v1.0/subscription"
@@ -77,8 +77,8 @@ def supported_data_types(self) -> List[DataType]:
         return [DataType.STATIC]
 
     @property
-    def supports_push(self) -> bool:
-        return False
+    def supported_update_methods(self) -> List[UpdateMethod]:
+        return [UpdateMethod.PULL]
 
     def load_data(self):
         root = get_mobilithek_data()

src/evmap_backend/data_sources/management/commands/load_data_source.py

@@ -1,5 +1,6 @@
 from django.core.management import BaseCommand
 
+from evmap_backend.data_sources.models import UpdateState
 from evmap_backend.data_sources.registry import get_data_source, list_available_sources
 
 
@@ -22,6 +23,7 @@ def handle(self, *args, **options):
             )
 
             data_source.load_data()
+            UpdateState(data_source=data_source.id, push=False).save()
 
             self.stdout.write(
                 self.style.SUCCESS(f"Successfully loaded data from source: {source_id}")

src/evmap_backend/data_sources/migrations/0001_initial.py

@@ -0,0 +1,24 @@
+# Generated by Django 5.2.6 on 2025-12-13 23:19
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+
+    initial = True
+
+    dependencies = []
+
+    operations = [
+        migrations.CreateModel(
+            name="UpdateState",
+            fields=[
+                (
+                    "data_source",
+                    models.CharField(max_length=255, primary_key=True, serialize=False),
+                ),
+                ("last_update", models.DateTimeField(auto_now=True)),
+                ("push", models.BooleanField()),
+            ],
+        ),
+    ]

src/evmap_backend/data_sources/migrations/__init__.py

@@ -0,0 +1,9 @@
+from django.db import models
+
+
+class UpdateState(models.Model):
+    data_source = models.CharField(
+        max_length=255, primary_key=True, blank=False, null=False
+    )
+    last_update = models.DateTimeField(auto_now=True)
+    push = models.BooleanField(blank=False, null=False)