Update windborne ingestor to fetch per mission and add it to API

django_project/gap/admin/main.py

@@ -86,7 +86,7 @@ class MeasurementAdmin(admin.ModelAdmin):
     """Measurement admin."""
 
     list_display = (
-        'station', 'dataset_attribute', 'date_time', 'value'
+        'station', 'dataset_attribute', 'date_time', 'value', 'station_history'
     )
     list_filter = ('station',)
     search_fields = ('name',)

django_project/gap/admin/station.py

@@ -10,6 +10,7 @@
 from gap.models import (
     Station, StationType, StationHistory
 )
+from gap.tasks.station import assign_history_of_stations_to_measurement
 
 
 @admin.register(StationType)
@@ -19,6 +20,14 @@
     pass
 
 
+@admin.action(description='Assign history to measurement')
+def assign_history_to_measurement(modeladmin, request, queryset):
+    """Assign history to measurement."""
+    assign_history_of_stations_to_measurement.delay(
+        list(queryset.values_list('id', flat=True))
+    )
+
+
 @admin.register(Station)
 class StationAdmin(admin.ModelAdmin):
     """Station admin."""
@@ -28,6 +37,7 @@
     )
     list_filter = ('provider', 'station_type', 'country')
     search_fields = ('code', 'name')
+    actions = (assign_history_to_measurement,)
 
 
 @admin.register(StationHistory)

django_project/gap/fixtures/4.dataset_type.json

@@ -43,10 +43,10 @@
     "model": "gap.datasettype",
     "pk": 5,
     "fields": {
-      "name": "Airborne Observational",
+      "name": "WindBorne Observational",
       "description": "",
       "type": "historical",
-      "variable_name": "default"
+      "variable_name": "windborne_observational"
     }
   },
   {

django_project/gap/ingestor/wind_borne_systems.py

@@ -23,7 +23,7 @@
 
 PROVIDER = 'WindBorne Systems'
 STATION_TYPE = 'Balloon'
-DATASET_TYPE = 'Airborne Observational'
+DATASET_TYPE = 'WindBorne Observational'
 DATASET_NAME = 'WindBorne Balloons Observations'
 USERNAME_ENV_NAME = 'WIND_BORNE_SYSTEMS_USERNAME'
 PASSWORD_ENV_NAME = 'WIND_BORNE_SYSTEMS_PASSWORD'
@@ -44,11 +44,12 @@
         if not self.password:
             raise EnvIsNotSetException(PASSWORD_ENV_NAME)
 
-    def measurements(self, since=None) -> (list, int, bool):
+    def measurements(self, mission_id, since=None) -> (list, int, bool):
         """Return measurements, since and has_next_page."""
         params = {
             'include_ids': True,
-            'include_mission_name': True
+            'include_mission_name': True,
+            'mission_id': mission_id
         }
         if since:
             params['since'] = since
@@ -67,6 +68,18 @@
             f'{response.status_code}: {response.text} : {response.url}'
         )
 
+    def missions(self) -> list:
+        """Return missions."""
+        response = requests.get(
+            f'{self.base_url}/missions.json',
+            auth=HTTPBasicAuth(self.username, self.password)
+        )
+        if response.status_code == 200:
+            return response.json()['missions']
+        raise Exception(
+            f'{response.status_code}: {response.text} : {response.url}'
+        )
+
 
 class WindBorneSystemsIngestor(BaseIngestor):
     """Ingestor for WindBorneSystems."""
@@ -96,68 +109,95 @@
         for dataset_attr in self.dataset.datasetattribute_set.all():
             self.attributes[dataset_attr.source] = dataset_attr
 
+    def mission_ids(self, api: WindBorneSystemsAPI):
+        """Mission ids."""
+        mission_ids = list(
+            Station.objects.filter(
+                provider=self.provider,
+                station_type=self.station_type
+            ).values_list('code', flat=True)
+        )
+        # From API
+        mission_ids.extend([mission['id'] for mission in api.missions()])
+        return list(set(mission_ids))
+
     def run(self):
         """Run the ingestor."""
         api = WindBorneSystemsAPI()
+        additional_config = self.session.additional_config
+        global_since = additional_config.get('since', None)
+
+        # Run for every mission id
+        missions = self.mission_ids(api)
+        for mission_id in missions:
+            print(f'Checking mission {mission_id}')
 
-        has_next_page = True
-        since = self.session.additional_config.get('since', None)
-        while has_next_page:
-            observations, since, has_next_page = api.measurements(since)
-
-            # Process if it has observations
-            if len(observations):
-                for observation in observations:
-                    # Get date time
-                    date_time = datetime.fromtimestamp(
-                        observation['timestamp']
-                    )
-                    date_time = timezone.make_aware(
-                        date_time, timezone.get_default_timezone()
-                    )
-
-                    # Points
-                    point = Point(
-                        x=observation['longitude'],
-                        y=observation['latitude'],
-                        srid=4326
-                    )
-                    station, _ = Station.objects.update_or_create(
-                        provider=self.provider,
-                        station_type=self.station_type,
-                        code=observation['mission_id'],
-                        defaults={
-                            'name': observation['mission_name'],
-                            'geometry': point,
-                            'altitude': observation['altitude'],
-                        }
-                    )
-                    StationHistory.objects.update_or_create(
-                        station=station,
-                        date_time=date_time,
-                        defaults={
-                            'geometry': point,
-                            'altitude': observation['altitude'],
-                        }
-                    )
-
-                    # Save the measurements
-                    for variable, dataset_attribute in self.attributes.items():
-                        try:
-                            value = observation[variable]
-                            if value is not None:
-                                Measurement.objects.update_or_create(
-                                    station=station,
-                                    dataset_attribute=dataset_attribute,
-                                    date_time=date_time,
-                                    defaults={
-                                        'value': observation[variable]
-                                    }
-                                )
-                        except KeyError:
-                            pass
-                # Save last since
-                self.session.additional_config = {
-                    'since': since,
-                }
-                self.session.save()
+            mission_since_key = f'since-{mission_id}'
+            mission_since = additional_config.get(
+                mission_since_key, None
+            )
+            since = mission_since if mission_since else global_since
+
+            has_next_page = True
+            while has_next_page:
+                observations, since, has_next_page = api.measurements(
+                    mission_id, since
+                )
+
+                # Process if it has observations
+                if len(observations):
+                    for observation in observations:
+                        # Get date time
+                        date_time = datetime.fromtimestamp(
+                            observation['timestamp']
+                        )
+                        date_time = timezone.make_aware(
+                            date_time, timezone.get_default_timezone()
+                        )
+
+                        # Points
+                        point = Point(
+                            x=observation['longitude'],
+                            y=observation['latitude'],
+                            srid=4326
+                        )
+                        station, _ = Station.objects.update_or_create(
+                            provider=self.provider,
+                            station_type=self.station_type,
+                            code=observation['mission_id'],
+                            defaults={
+                                'name': observation['mission_name'],
+                                'geometry': point,
+                                'altitude': observation['altitude'],
+                            }
+                        )
+                        history, _ = StationHistory.objects.update_or_create(
+                            station=station,
+                            date_time=date_time,
+                            defaults={
+                                'geometry': point,
+                                'altitude': observation['altitude'],
+                            }
+                        )
+
+                        # Save the measurements
+                        for variable, attribute in self.attributes.items():
+                            try:
+                                value = observation[variable]
+                                if value is not None:
+                                    Measurement.objects.update_or_create(
+                                        station=station,
+                                        dataset_attribute=attribute,
+                                        date_time=date_time,
+                                        defaults={
+                                            'value': observation[variable],
+                                            'station_history': history
+                                        }
+                                    )
+                            except KeyError:
+                                pass
+
+                    # Save last since for mission
+                    additional_config[mission_since_key] = since
+                    self.session.additional_config = additional_config
+                    self.session.save()

django_project/gap/migrations/0031_measurement_station_history.py

@@ -0,0 +1,19 @@
+# Generated by Django 4.2.7 on 2024-10-17 09:52
+
+from django.db import migrations, models
+import django.db.models.deletion
+
+
+class Migration(migrations.Migration):
+
+    dependencies = [
+        ('gap', '0030_preferences_ingestor_config'),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name='measurement',
+            name='station_history',
+            field=models.ForeignKey(blank=True, help_text='Station history of the measurement.', null=True, on_delete=django.db.models.deletion.CASCADE, to='gap.stationhistory'),
+        ),
+    ]

django_project/gap/models/measurement.py

@@ -10,7 +10,7 @@
 from core.models.common import Definition
 from gap.models.common import Unit
 from gap.models.dataset import Dataset
-from gap.models.station import Station
+from gap.models.station import Station, StationHistory
 
 
 class Attribute(Definition):
@@ -77,6 +77,15 @@ class Measurement(models.Model):
     date_time = models.DateTimeField()
     value = models.FloatField()
 
+    # Specifically measurement is linked to a station history
+    station_history = models.ForeignKey(
+        StationHistory, on_delete=models.CASCADE,
+        help_text=(
+            'Station history of the measurement.'
+        ),
+        null=True, blank=True
+    )
+
     def __str__(self):
         return f'{self.date_time} - {self.value}'
 

django_project/gap/providers/__init__.py

@@ -5,16 +5,20 @@
 .. note:: Helper for reading NetCDF File
 """
 
+from gap.ingestor.wind_borne_systems import PROVIDER
 from gap.models import Dataset, DatasetStore
-from gap.utils.netcdf import NetCDFProvider
+from gap.providers.airborne_observation import ObservationAirborneDatasetReader
 from gap.providers.cbam import CBAMZarrReader, CBAMNetCDFReader  # noqa
-from gap.providers.salient import SalientNetCDFReader, SalientZarrReader  # noqa
 from gap.providers.observation import ObservationDatasetReader
+from gap.providers.salient import (
+    SalientNetCDFReader, SalientZarrReader
+)  # noqa
 from gap.providers.tio import (
     TomorrowIODatasetReader,
     PROVIDER_NAME as TIO_PROVIDER,
     TioZarrReader
 )
+from gap.utils.netcdf import NetCDFProvider
 
 
 def get_reader_from_dataset(dataset: Dataset):
@@ -32,6 +36,8 @@ def get_reader_from_dataset(dataset: Dataset):
         return SalientZarrReader
     elif dataset.provider.name in ['Tahmo', 'Arable']:
         return ObservationDatasetReader
+    elif dataset.provider.name in [PROVIDER]:
+        return ObservationAirborneDatasetReader
     elif (
         dataset.provider.name == TIO_PROVIDER and
         dataset.store_type == DatasetStore.EXT_API