Added feed manager (#4)

flightradar24_client/dump1090_aircrafts.py

@@ -10,6 +10,7 @@
     ATTR_TRACK, ATTR_UPDATED, ATTR_SPEED, ATTR_CALLSIGN, ATTR_ALTITUDE, \
     ATTR_MODE_S, ATTR_LONGITUDE, ATTR_LATITUDE, ATTR_LON, ATTR_LAT, ATTR_HEX, \
     ATTR_FLIGHT
+from flightradar24_client.feed_manager import FeedManagerBase
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -19,6 +20,20 @@
 URL_TEMPLATE = "http://{}:{}/data/aircraft.json"
 
 
+class Dump1090AircraftsFeedManager(FeedManagerBase):
+    """Feed Manager for Dump1090 Aircrafts feed."""
+
+    def __init__(self, generate_callback, update_callback, remove_callback,
+                 coordinates, filter_radius=None, url=None,
+                 hostname=DEFAULT_HOSTNAME, port=DEFAULT_PORT, loop=None,
+                 session=None):
+        """Initialize the NSW Rural Fire Services Feed Manager."""
+        feed = Dump1090AircraftsFeedAggregator(
+            coordinates, filter_radius=filter_radius, url=url,
+            hostname=hostname, port=port, loop=loop, session=session)
+        super().__init__(feed, generate_callback, update_callback,
+                         remove_callback)
+
 class Dump1090AircraftsFeedAggregator(FeedAggregator):
     """Aggregates date received from the feed over a period of time."""
 

flightradar24_client/feed_manager.py

@@ -0,0 +1,77 @@
+"""
+Base class for the feed manager.
+
+This allows managing feeds and their entries throughout their life-cycle.
+"""
+import logging
+
+from flightradar24_client import UPDATE_OK
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class FeedManagerBase:
+    """Generic Feed manager."""
+
+    def __init__(self, feed, generate_callback, update_callback,
+                 remove_callback, persistent_timestamp=False):
+        """Initialise feed manager."""
+        self._feed = feed
+        self.feed_entries = {}
+        self._managed_external_ids = set()
+        self._generate_callback = generate_callback
+        self._update_callback = update_callback
+        self._remove_callback = remove_callback
+        self._persistent_timestamp = persistent_timestamp
+
+    def __repr__(self):
+        """Return string representation of this feed."""
+        return '<{}(feed={})>'.format(
+            self.__class__.__name__, self._feed)
+
+    async def update(self):
+        """Update the feed and then update connected entities."""
+        status, feed_entries = await self._feed.update()
+        if status == UPDATE_OK:
+            _LOGGER.warning("Data retrieved %s", feed_entries)
+            # Keep a copy of all feed entries for future lookups by entities.
+            self.feed_entries = feed_entries
+            # For entity management the external ids from the feed are used.
+            feed_external_ids = set(self.feed_entries)
+            remove_external_ids = self._managed_external_ids.difference(
+                feed_external_ids)
+            self._remove_entities(remove_external_ids)
+            update_external_ids = self._managed_external_ids.intersection(
+                feed_external_ids)
+            self._update_entities(update_external_ids)
+            create_external_ids = feed_external_ids.difference(
+                self._managed_external_ids)
+            self._generate_new_entities(create_external_ids)
+        else:
+            _LOGGER.warning(
+                "Update not successful, no data received from %s", self._feed)
+            # Remove all entities.
+            self._remove_entities(self._managed_external_ids.copy())
+            # Remove all feed entries and managed external ids.
+            self.feed_entries.clear()
+            self._managed_external_ids.clear()
+
+    def _generate_new_entities(self, external_ids):
+        """Generate new entities for events."""
+        for external_id in external_ids:
+            self._generate_callback(external_id)
+            _LOGGER.debug("New entity added %s", external_id)
+            self._managed_external_ids.add(external_id)
+
+    def _update_entities(self, external_ids):
+        """Update entities."""
+        for external_id in external_ids:
+            _LOGGER.debug("Existing entity found %s", external_id)
+            self._update_callback(external_id)
+
+    def _remove_entities(self, external_ids):
+        """Remove entities."""
+        for external_id in external_ids:
+            _LOGGER.debug("Entity not current anymore %s", external_id)
+            self._managed_external_ids.remove(external_id)
+            self._remove_callback(external_id)

flightradar24_client/fr24_flights.py

@@ -9,6 +9,7 @@
 from flightradar24_client.consts import ATTR_VERT_RATE, ATTR_SQUAWK, \
     ATTR_TRACK, ATTR_UPDATED, ATTR_SPEED, ATTR_CALLSIGN, \
     ATTR_ALTITUDE, ATTR_MODE_S, ATTR_LONGITUDE, ATTR_LATITUDE
+from flightradar24_client.feed_manager import FeedManagerBase
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -18,6 +19,21 @@
 URL_TEMPLATE = "http://{}:{}/flights.json"
 
 
+class Flightradar24FlightsFeedManager(FeedManagerBase):
+    """Feed Manager for Flightradar24 Flights feed."""
+
+    def __init__(self, generate_callback, update_callback, remove_callback,
+                 coordinates, filter_radius=None, url=None,
+                 hostname=DEFAULT_HOSTNAME, port=DEFAULT_PORT, loop=None,
+                 session=None):
+        """Initialize the NSW Rural Fire Services Feed Manager."""
+        feed = Flightradar24FlightsFeedAggregator(
+            coordinates, filter_radius=filter_radius, url=url,
+            hostname=hostname, port=port, loop=loop, session=session)
+        super().__init__(feed, generate_callback, update_callback,
+                         remove_callback)
+
+
 class Flightradar24FlightsFeedAggregator(FeedAggregator):
     """Aggregates date received from the feed over a period of time."""
 

tests/fixtures/fr24-flights-3.json

@@ -0,0 +1,8 @@
+{
+  "x7c1469": ["7C1469", -33.8880,151.2435,167,2950,183,"4040",0,"","",1540539591,"","","",0,-64,"QFA456"],
+  "x7c1c5d": ["7C1C5D", -33.2635,151.2188,182,12725,241,"1117",0,"","",1540539591,"","","",0,-576,"VOZ1192"],
+  "x7c5304": ["7C5304", -34.3867,150.6932,234,17775,230,"4024",0,"","",1540539590,"","","",0,1088,""],
+  "x7c6b28": ["7C6B28", -32.5470,150.9698,268,22175,265,"1140",0,"","",1540539591,"","","",0,-1600,"JST423"],
+  "x7c6b39": ["7C6B39", 0.0000,0.0000,0,5000,0,"3665",0,"","",1540539564,"","","",0,0,""],
+  "x7c6b29": ["7C6B29", -32.6581,150.0709,268,22175,265,"1140",0,"","",1540539591,"","","",0,-1600,"JST423"]
+}

tests/test_dump1090_aircrafts.py

@@ -8,7 +8,7 @@
 from flightradar24_client import FeedEntry
 from flightradar24_client.consts import UPDATE_OK, UPDATE_ERROR
 from flightradar24_client.dump1090_aircrafts import Dump1090AircraftsFeed, \
-    Dump1090AircraftsFeedAggregator
+    Dump1090AircraftsFeedAggregator, Dump1090AircraftsFeedManager
 from tests.utils import load_fixture
 
 
@@ -178,6 +178,52 @@ def test_feed_aggregator_filter_radius(self, mock_session):
         self.assertIsNotNone(entries)
         assert len(entries) == 1
 
+    @aioresponses()
+    def test_feed_manager(self, mock_session):
+        """Test the feed manager."""
+        loop = asyncio.get_event_loop()
+        home_coordinates = (-31.0, 151.0)
+        mock_session.get('http://localhost:8888/data/aircraft.json',
+                         status=200,
+                         body=load_fixture('dump1090-aircrafts-1.json'))
+
+        # This will just record calls and keep track of external ids.
+        generated_entity_external_ids = []
+        updated_entity_external_ids = []
+        removed_entity_external_ids = []
+
+        def _generate_entity(external_id):
+            """Generate new entity."""
+            generated_entity_external_ids.append(external_id)
+
+        def _update_entity(external_id):
+            """Update entity."""
+            updated_entity_external_ids.append(external_id)
+
+        def _remove_entity(external_id):
+            """Remove entity."""
+            removed_entity_external_ids.append(external_id)
+
+        feed_manager = Dump1090AircraftsFeedManager(_generate_entity,
+                                                    _update_entity,
+                                                    _remove_entity,
+                                                    home_coordinates,
+                                                    None)
+        assert repr(feed_manager) == "<Dump1090AircraftsFeedManager(feed=" \
+                                     "<Dump1090AircraftsFeedAggregator" \
+                                     "(feed=<Dump1090AircraftsFeed(" \
+                                     "home=(-31.0, 151.0), " \
+                                     "url=http://localhost:8888/" \
+                                     "data/aircraft.json, " \
+                                     "radius=None)>)>)>"
+        loop.run_until_complete(feed_manager.update())
+        entries = feed_manager.feed_entries
+        self.assertIsNotNone(entries)
+        assert len(entries) == 4
+        assert len(generated_entity_external_ids) == 4
+        assert len(updated_entity_external_ids) == 0
+        assert len(removed_entity_external_ids) == 0
+
     def test_entry_without_data(self):
         """Test simple entry without data."""
         entry = FeedEntry(None, None)

tests/test_fr24_flights.py

@@ -7,7 +7,7 @@
 
 from flightradar24_client.consts import UPDATE_OK, UPDATE_ERROR
 from flightradar24_client.fr24_flights import Flightradar24FlightsFeed, \
-    Flightradar24FlightsFeedAggregator
+    Flightradar24FlightsFeedAggregator, Flightradar24FlightsFeedManager
 from flightradar24_client import FeedEntry
 from tests.utils import load_fixture
 
@@ -184,6 +184,92 @@ def test_feed_aggregator_update_error(self, mock_session):
         assert status == UPDATE_ERROR
         self.assertIsNone(entries)
 
+    @aioresponses()
+    def test_feed_manager(self, mock_session):
+        """Test the feed manager."""
+        loop = asyncio.get_event_loop()
+        home_coordinates = (-31.0, 151.0)
+        mock_session.get('http://localhost:8754/flights.json', status=200,
+                         body=load_fixture('fr24-flights-1.json'))
+
+        # This will just record calls and keep track of external ids.
+        generated_entity_external_ids = []
+        updated_entity_external_ids = []
+        removed_entity_external_ids = []
+
+        def _generate_entity(external_id):
+            """Generate new entity."""
+            generated_entity_external_ids.append(external_id)
+
+        def _update_entity(external_id):
+            """Update entity."""
+            updated_entity_external_ids.append(external_id)
+
+        def _remove_entity(external_id):
+            """Remove entity."""
+            removed_entity_external_ids.append(external_id)
+
+        feed_manager = Flightradar24FlightsFeedManager(_generate_entity,
+                                                       _update_entity,
+                                                       _remove_entity,
+                                                       home_coordinates,
+                                                       None)
+        assert repr(feed_manager) == "<Flightradar24FlightsFeedManager(feed=" \
+                                     "<Flightradar24FlightsFeedAggregator" \
+                                     "(feed=<Flightradar24FlightsFeed(" \
+                                     "home=(-31.0, 151.0), " \
+                                     "url=http://localhost:8754/" \
+                                     "flights.json, " \
+                                     "radius=None)>)>)>"
+        loop.run_until_complete(feed_manager.update())
+        entries = feed_manager.feed_entries
+        self.assertIsNotNone(entries)
+        assert len(entries) == 5
+        assert len(generated_entity_external_ids) == 5
+        assert len(updated_entity_external_ids) == 0
+        assert len(removed_entity_external_ids) == 0
+
+        feed_entry = entries['7C1469']
+        assert feed_entry.external_id == "7C1469"
+        assert feed_entry.coordinates == (-33.7779, 151.1324)
+        assert repr(feed_entry) == "<FeedEntry(id=7C1469)>"
+
+        # Simulate an update with several changes.
+        generated_entity_external_ids.clear()
+        updated_entity_external_ids.clear()
+        removed_entity_external_ids.clear()
+
+        mock_session.get('http://localhost:8754/flights.json', status=200,
+                         body=load_fixture('fr24-flights-3.json'))
+
+        loop.run_until_complete(feed_manager.update())
+        entries = feed_manager.feed_entries
+        self.assertIsNotNone(entries)
+        assert len(entries) == 5
+        assert len(generated_entity_external_ids) == 1
+        assert len(updated_entity_external_ids) == 4
+        assert len(removed_entity_external_ids) == 1
+
+        feed_entry = entries['7C1469']
+        assert feed_entry.external_id == "7C1469"
+        assert feed_entry.coordinates == (-33.8880, 151.2435)
+
+        # Simulate an update with no data.
+        generated_entity_external_ids.clear()
+        updated_entity_external_ids.clear()
+        removed_entity_external_ids.clear()
+
+        mock_session.get('http://localhost:8754/flights.json', status=500,
+                         body='ERROR')
+
+        loop.run_until_complete(feed_manager.update())
+        entries = feed_manager.feed_entries
+
+        assert len(entries) == 0
+        assert len(generated_entity_external_ids) == 0
+        assert len(updated_entity_external_ids) == 0
+        assert len(removed_entity_external_ids) == 5
+
     def test_entry_without_data(self):
         """Test simple entry without data."""
         entry = FeedEntry(None, None)