openrvdas.logger.transforms.geofence_transform

  1#!/usr/bin/env python3
  2"""Read lat/lon from passed records and compare to a geofence loaded at initialization
  3time. Emit pre-defined messages if lat/lon transition between inside and outside of
  4fence.
  5
  6EEZ files in GML format can be downloaded from https://marineregions.org/eezsearch.php
  7
  8Sample logger that switches modes when entering/exiting EEZ:
  9```
 10# Read parsed DASRecords from UDP
 11readers:
 12  class: UDPReader
 13  kwargs:
 14    port: 6224
 15# Look for lat/lon values in the DASRecords and emit appropriate commands
 16# when entering/leaving EEZ. Note that EEZ files in GML format can be
 17# downloaded from https://marineregions.org/eezsearch.php.
 18transforms:
 19  - class: GeofenceTransform
 20    module: logger.transforms.geofence_transform
 21    kwargs:
 22      latitude_field_name: s330Latitude,
 23      longitude_field_name: s330Longitude
 24      boundary_file_name: /tmp/eez.gml
 25      leaving_boundary_message: set_active_mode write+influx
 26      entering_boundary_message: set_active_mode no_write+influx
 27# Send the messages that we get from geofence to the LoggerManager
 28writers:
 29  - class: LoggerManagerWriter
 30    module: logger.writers.logger_manager_writer
 31    kwargs:
 32      database: django
 33      allowed_prefixes:
 34        - 'set_active_mode '
 35        - 'sleep '
 36```
 37Some questions:
 38- Should messages be emitted when first record is received? That is, when transform
 39  first fires up, should it send the "entering_boundary_message" if the first record
 40  it receives indicates it's inside? DECISION: Yes.
 41
 42NOTE: optional parameter distance_from_boundary is in degrees. Computing the appropriate
 43value in km/nm is nontrivial and requires figuring out the right UTM projection for each
 44location and recomputing it for each point and switching when lat/lon moved to a new UTM
 45projection area, possibly resulting in discontinuities. Simpler and less error-prone
 46to just require degrees.
 47"""
 48import logging
 49import os
 50import time
 51
 52from typing import Union
 53from logger.utils.das_record import DASRecord  # noqa: E402
 54from logger.transforms.transform import Transform  # noqa: E402
 55
 56# Load the transform-specific packages we need
 57import_errors = False
 58try:
 59    import geopandas as gpd
 60except ImportError:
 61    import_errors = True
 62try:
 63    from shapely.geometry import Point
 64except ImportError:
 65    import_errors = True
 66
 67import_pandas_errors = False
 68try:
 69    import pandas as pd
 70except ImportError:
 71    import_pandas_errors = True
 72
 73
 74################################################################################
 75class GeofenceTransform(Transform):
 76    """Class that reads lat/lon from passed records and compare to a geofence loaded at
 77    initialization time. Emit pre-defined messages if lat/lon transition between inside
 78    and outside of fence.
 79    """
 80    def __init__(self,
 81                 latitude_field_name,
 82                 longitude_field_name,
 83                 boundary_file_name=None,
 84                 boundary_dir_name=None,
 85                 distance_from_boundary_in_degrees=0,
 86                 leaving_boundary_message=None,
 87                 entering_boundary_message=None,
 88                 seconds_between_checks=0,
 89                 **kwargs):
 90        """
 91        latitude_field_name
 92        longitude_field_name
 93                Field names to read for lat/lon values.# what fields to listen
 94                to for lat/lon values. Format is assumed to be decimal, with
 95                negative values representing south latitude and west longitude.
 96
 97        boundary_file_name
 98                Path to file from which to load GML boundary definition..
 99
100        boundary_dir_name
101                Path to directory from which to load multiple GML boundary definitions.
102
103        distance_from_boundary_in_degrees
104                Optional distance from boundary to place the fence, in degrees.
105                Negative means inside the boundary.
106
107        leaving_boundary_message
108                Optional message to emit when boundary is crossed, outbound
109        entering_boundary_message,
110                Optional message to emit when boundary is crossed, inbound
111
112        seconds_between_checks
113                Optional number of seconds to wait between doing checks,
114                computation overhead
115        """
116        # Only throw this error if user tries to actually use this code
117        if import_errors:
118            raise ImportError('GeofenceTransform requires installation of geopandas and '
119                              'shapely packages. Please run "pip install geopandas shapely" '
120                              'and retry.')
121
122        super().__init__(**kwargs)  # processes 'quiet' and type hints
123
124        self.latitude_field_name = latitude_field_name
125        self.longitude_field_name = longitude_field_name
126        self.leaving_boundary_message = leaving_boundary_message
127        self.entering_boundary_message = entering_boundary_message
128        self.seconds_between_checks = seconds_between_checks
129
130        # Once we start receiving data, this will either be True or False
131        self.last_position_inside = None
132
133        self.last_check = 0  # timestamp: the last time we checked
134
135        if (boundary_file_name and boundary_dir_name
136                or not (boundary_file_name or boundary_dir_name)):
137            raise ValueError('Please specify a boundary_file_name OR a boundary_dir_name '
138                             'containing multiple gml files.')
139
140        eez_data = None
141
142        if boundary_file_name:
143            # Load the EEZ data from the GML file
144            eez_data = gpd.read_file(boundary_file_name)
145
146        if boundary_dir_name:
147            if import_pandas_errors:
148                raise ImportError('GeofenceTransform using "boundary_dir_name" requires '
149                                  'installation of the pandas packages. Please run "pip '
150                                  'install pandas" and retry.')
151
152            # List all GML files in the directory
153            gml_files = [os.path.join(boundary_dir_name, file)
154                         for file in os.listdir(boundary_dir_name)
155                         if file.endswith('.gml')]
156
157            # Initialize an empty list to store GeoDataFrames
158            dfs = []
159
160            # Read each GML file into a GeoDataFrame and store it in the list
161            for gml in gml_files:
162                dfs.append(gpd.read_file(gml))
163
164            # Combine all GeoDataFrames into a single GeoDataFrame
165            combined_gdf = gpd.GeoDataFrame(pd.concat(dfs, ignore_index=True), crs=dfs[0].crs)
166
167            # Combine all polygons into a single polygon
168            # Handle deprecation of unary_union in newer geopandas/shapely versions
169            try:
170                union_polygon = combined_gdf.union_all()
171            except AttributeError:
172                # Fallback for older versions
173                union_polygon = combined_gdf.unary_union
174
175            # Convert the union polygon into a GeoDataFrame
176            eez_data = gpd.GeoDataFrame(geometry=[union_polygon])
177
178        # Buffer the country's EEZ by distance in degrees
179        self.buffered_eez = eez_data.buffer(distance_from_boundary_in_degrees)
180
181    ############################
182    def _get_lat_lon(self, record: Union[DASRecord, dict]):
183        """If the DASRecord or dict contains a lat/lon pair, return it as a tuple,
184        otherwise return (None, None)."""
185
186        if type(record) is dict:
187            # Is it a simple dict with lat/lon defined at the top level?
188            lat = record.get(self.latitude_field_name, None)
189            lon = record.get(self.longitude_field_name, None)
190            if lat is not None and lon is not None:
191                return (lat, lon)
192
193            # Is it a dict with a 'fields' subdict?
194            if record.get('fields') is not None:
195                lat = record['fields'].get(self.latitude_field_name, None)
196                lon = record['fields'].get(self.longitude_field_name, None)
197                if lat is not None and lon is not None:
198                    return (lat, lon)
199
200            # No lat/lon pairs we can find in this dict
201            return (None, None)
202
203        # Maybe they've passed us a DASRecord
204        if type(record) is DASRecord:
205            lat = record.fields.get(self.latitude_field_name, None)
206            lon = record.fields.get(self.longitude_field_name, None)
207            if lat is not None and lon is not None:
208                return (lat, lon)
209            else:
210                return (None, None)
211
212    # Define a function to check if a point is within N nautical miles of the geofenced area
213    def _is_inside_boundary(self, lat, lon):
214        # Note that Point() takes longitude as first arg, not latitude
215        point = Point(lon, lat)
216        return self.buffered_eez.contains(point).any()
217
218    ############################
219    def transform(self, record: Union[DASRecord, dict]):
220        """Look for the named lat/lon fields in the passed dict. If the previous
221        lat/lon pair was on one side of the geofence and this lat/lon pair is on
222        the other, return the appropriate string defined in either
223        leaving_boundary_message or entering_boundary_message. Otherwise, return
224        None.
225
226        record
227                A DASRecord, dict of {field_name: field_value} pairs, or a list of
228                DASRecords/dicts in which to look for the specified latitude_field_name
229                and longitude_field_name.
230        """
231        # See if it's something we can process, and if not, try digesting
232        if not self.can_process_record(record):  # BaseModule
233            return self.digest_record(record)  # BaseModule
234
235        # If we've checked too recently, skip check. Note that because this decision
236        # is made for computational efficiency rather than data efficiency, it is made
237        # based on system time, not the timestamp of the record.
238        now = time.time()
239        time_since_last = now - self.last_check
240        if self.seconds_between_checks and time_since_last < self.seconds_between_checks:
241            logging.debug(f'Only {time_since_last} seconds since last GeofenceTransform check; '
242                          f'less than the {self.seconds_between_checks} required.')
243            return None
244
245        # Does this record have a lat/lon?
246        (lat, lon) = self._get_lat_lon(record)
247        if lat is None or lon is None:
248            return None
249
250        # We have a lat and lon, so we're going ahead and checking
251        self.last_check = now
252
253        is_inside = self._is_inside_boundary(lat, lon)
254        if is_inside == self.last_position_inside:
255            return None
256
257        self.last_position_inside = is_inside
258        if is_inside:
259            logging.info('GeofenceTransform entered boundary')
260            logging.info(f'Issuing message: {self.entering_boundary_message}')
261            return self.entering_boundary_message
262        else:
263            logging.info('GeofenceTransform exited boundary')
264            logging.info(f'Issuing message: {self.leaving_boundary_message}')
265            return self.leaving_boundary_message