Add detail by location feature to adaptation and historic

Author: Lyetenth

firststreet/api/adaptation.py

@@ -42,6 +42,52 @@ def get_detail(self, fsids, csv=False, limit=100, output_dir=None):
 
         return product
 
+    def get_details_by_location(self, fsids, location_type, csv=False, limit=100, output_dir=None):
+        """Retrieves adaptation detail product data from the First Street Foundation API given a list of location
+        FSIDs and returns a list of Adaptation Detail objects.
+
+        Args:
+            fsids (list/file): A First Street Foundation IDs or a file of First Street Foundation IDs
+            location_type (str): The location lookup type
+            csv (bool): To output extracted data to a csv or not
+            limit (int): max number of connections to make
+            output_dir (str): The output directory to save the generated csvs
+        Returns:
+            A list of list of Adaptation Summary and Adaptation Detail
+        Raises:
+            InvalidArgument: The location provided is empty
+            TypeError: The location provided is not a string
+        """
+
+        if not location_type:
+            raise InvalidArgument(location_type)
+        elif not isinstance(location_type, str):
+            raise TypeError("location is not a string")
+        elif location_type == 'property':
+            raise InvalidArgument("Property is not a valid location type")
+
+        # Get data from api and create objects
+        api_datas_summary = self.call_api(fsids, "adaptation", "summary", location_type, limit=limit)
+        summary = [AdaptationSummary(api_data) for api_data in api_datas_summary]
+
+        fsids = list(set([adaptation for sum_adap in summary if sum_adap.adaptation for
+                          adaptation in sum_adap.adaptation]))
+
+        if fsids:
+            api_datas_detail = self.call_api(fsids, "adaptation", "detail", None, limit=limit)
+
+        else:
+            api_datas_detail = [{"adaptationId": None}]
+
+        detail = [AdaptationDetail(api_data) for api_data in api_datas_detail]
+
+        if csv:
+            csv_format.to_csv([summary, detail], "adaptation", "summary_detail", location_type, output_dir=output_dir)
+
+        logging.info("Adaptation Summary Detail Data Ready.")
+
+        return [summary, detail]
+
     def get_summary(self, fsids, location_type, csv=False, limit=100, output_dir=None):
         """Retrieves adaptation summary product data from the First Street Foundation API given a list of FSIDs and
         returns a list of Adaptation Summary objects.
@@ -54,6 +100,9 @@ def get_summary(self, fsids, location_type, csv=False, limit=100, output_dir=Non
             output_dir (str): The output directory to save the generated csvs
         Returns:
             A list of Adaptation Summary
+        Raises:
+            InvalidArgument: The location provided is empty
+            TypeError: The location provided is not a string
         """
 
         if not location_type:

firststreet/api/csv_format.py

@@ -44,6 +44,9 @@ def to_csv(data, product, product_subtype, location_type=None, output_dir=None):
         elif product_subtype == 'summary':
             df = format_adaptation_summary(data)
 
+        elif product_subtype == 'summary_detail':
+            df = format_adaptation_summary_detail(data)
+
         else:
             raise NotImplementedError
 
@@ -83,6 +86,9 @@ def to_csv(data, product, product_subtype, location_type=None, output_dir=None):
             else:
                 df = format_historic_summary(data)
 
+        elif product_subtype == 'summary_event':
+            df = format_historic_summary_event(data)
+
         else:
             raise NotImplementedError
 
@@ -136,8 +142,7 @@ def to_csv(data, product, product_subtype, location_type=None, output_dir=None):
         raise NotImplementedError
 
     # Export CSVs
-    df.fillna(pd.NA, inplace=True)
-    df.to_csv(output_dir + '/' + file_name, index=False)
+    df.fillna(pd.NA).astype(str).to_csv(output_dir + '/' + file_name, index=False)
     logging.info("CSV generated to '{}'.".format(output_dir + '/' + file_name))
 
 
@@ -157,7 +162,7 @@ def format_adaptation_detail(data):
 
 
 def format_adaptation_summary(data):
-    """Reformat the list of data to Adaptation Summary format
+    """Reformat the list of data to Adaptation Summary Detail format
 
     Args:
         data (list): A list of FSF object
@@ -166,9 +171,28 @@ def format_adaptation_summary(data):
     """
     df = pd.DataFrame([vars(o) for o in data]).explode('adaptation').reset_index(drop=True)
     df['fsid'] = df['fsid'].apply(str)
+    df['adaptation'] = df['adaptation'].astype('Int64').apply(str)
     return df[['fsid', 'adaptation', 'properties']]
 
 
+def format_adaptation_summary_detail(data):
+    """Reformat the list of data to Adaptation Summary format
+
+    Args:
+        data (list): A list of FSF object
+    Returns:
+        A pandas formatted DataFrame
+    """
+
+    summary = format_adaptation_summary(data[0])
+    detail = format_adaptation_detail(data[1])
+
+    df = pd.merge(summary, detail, left_on='adaptation', right_on='adaptationId', how='left')\
+        .drop('adaptationId', axis=1)
+
+    return df
+
+
 def format_probability_chance(data):
     """Reformat the list of data to Probability Chance format
 
@@ -245,9 +269,6 @@ def format_probability_count_summary(data):
     Returns:
         A pandas formatted DataFrame
     """
-    # listed_data = [{'fsid': attr.fsid, 'location': [{'state': attr.state}, {'city': attr.city}, {'zcta': attr.zcta},
-    #                                                 {'neighborhood': attr.neighborhood}, {'tract': attr.tract},
-    #                                                 {'county': attr.county}, {'cd': attr.cd}]} for attr in data]
     listed_data = [{'fsid': attr.fsid, 'location': [{'location': 'state', 'data': attr.state},
                                                     {'location': 'city', 'data': attr.city},
                                                     {'location': 'zcta', 'data': attr.zcta},
@@ -462,6 +483,21 @@ def format_historic_summary(data):
     return df[['fsid', 'eventId', 'name', 'type', 'bin', 'count']]
 
 
+def format_historic_summary_event(data):
+    """Reformat the list of data to Historic Summary Event format
+
+    Args:
+        data (list): A list of FSF object
+    Returns:
+        A pandas formatted DataFrame
+    """
+
+    summary = format_historic_summary(data[0])
+    event = format_historic_event(data[1])
+
+    return pd.merge(summary, event, on='eventId', how='left')
+
+
 def format_location_detail_property(data):
     """Reformat the list of data to Location Detail format for property
 

firststreet/api/historic.py

@@ -43,6 +43,52 @@ def get_event(self, fsids, csv=False, limit=100, output_dir=None):
 
         return product
 
+    def get_events_by_location(self, fsids, location_type, csv=False, limit=100, output_dir=None):
+        """Retrieves historic summary product data from the First Street Foundation API given a list of location
+        FSIDs and returns a list of Historic Summary objects.
+
+        Args:
+            fsids (list/file): A First Street Foundation IDs or a file of First Street Foundation IDs
+            location_type (str): The location lookup type
+            csv (bool): To output extracted data to a csv or not
+            limit (int): max number of connections to make
+            output_dir (str): The output directory to save the generated csvs
+        Returns:
+            A list of Historic Event
+        Raises:
+            InvalidArgument: The location provided is empty
+            TypeError: The location provided is not a string
+        """
+
+        if not location_type:
+            raise InvalidArgument(location_type)
+        elif not isinstance(location_type, str):
+            raise TypeError("location is not a string")
+        elif location_type == 'property':
+            raise InvalidArgument("Property is not a valid location type")
+
+        # Get data from api and create objects
+        api_datas = self.call_api(fsids, "historic", "summary", location_type, limit)
+        summary = [HistoricSummary(api_data) for api_data in api_datas]
+
+        fsids = list(set([event.get("eventId") for sum_hist in summary if sum_hist.historic for
+                          event in sum_hist.historic]))
+
+        if fsids:
+            api_datas_event = self.call_api(fsids, "historic", "event", None, limit)
+
+        else:
+            api_datas_event = [{"eventId": None}]
+
+        event = [HistoricEvent(api_data) for api_data in api_datas_event]
+
+        if csv:
+            csv_format.to_csv([summary, event], "historic", "summary_event", location_type, output_dir=output_dir)
+
+        logging.info("Historic Summary Event Data Ready.")
+
+        return [summary, event]
+
     def get_summary(self, fsids, location_type, csv=False, limit=100, output_dir=None):
         """Retrieves historic summary product data from the First Street Foundation API given a list of FSIDs and
         returns a list of Historic Summary objects.
@@ -55,6 +101,9 @@ def get_summary(self, fsids, location_type, csv=False, limit=100, output_dir=Non
             output_dir (str): The output directory to save the generated csvs
         Returns:
             A list of Historic Summary
+        Raises:
+            InvalidArgument: The location provided is empty
+            TypeError: The location provided is not a string
         """
 
         if not location_type:

firststreet/models/geometry.py

@@ -15,7 +15,7 @@ class Geometry:
     def __init__(self, geometry):
 
         if geometry:
-            self.polygon = shape(geometry.get('polygon'))
+            self.polygon = shape(geometry.get('polygon')) if geometry.get('polygon') else None
             self.center = shape(geometry.get('center'))
             if geometry.get('bbox'):
                 self.bbox = shape(geometry.get('bbox'))

tests/api/test_adaptation.py

@@ -105,3 +105,68 @@ def test_mixed_invalid(self):
     def test_mixed_invalid_csv(self, tmpdir):
         adaptation = fs.adaptation.get_summary([2739, 0000], "property", csv=True, output_dir=tmpdir)
         assert len(adaptation) == 2
+
+
+class TestAdaptationSummaryDetail:
+
+    def test_empty(self):
+        with pytest.raises(InvalidArgument):
+            fs.adaptation.get_details_by_location([], "")
+
+    def test_empty_fsid(self):
+        with pytest.raises(InvalidArgument):
+            fs.adaptation.get_details_by_location([], "property")
+
+    def test_empty_type(self):
+        with pytest.raises(InvalidArgument):
+            fs.adaptation.get_details_by_location([1935265], "")
+
+    def test_wrong_fsid_type(self):
+        with pytest.raises(InvalidArgument):
+            fs.adaptation.get_details_by_location(190836953, "city")
+
+    def test_wrong_fsid_number(self):
+        adaptation = fs.adaptation.get_details_by_location([11], "city")
+        assert len(adaptation[0]) == 1
+        assert len(adaptation[1]) == 1
+        assert adaptation[0][0].adaptation is None
+
+    def test_incorrect_lookup_type(self, tmpdir):
+        adaptation = fs.adaptation.get_details_by_location([1935265], "state", csv=True, output_dir=tmpdir)
+        assert len(adaptation[0]) == 1
+        assert len(adaptation[1]) == 1
+        assert adaptation[0][0].adaptation is None
+
+    def test_wrong_adaptation_type(self):
+        with pytest.raises(TypeError):
+            fs.adaptation.get_details_by_location([1935265], 190)
+
+    def test_single(self):
+        adaptation = fs.adaptation.get_details_by_location([1935265], "city")
+        assert len(adaptation[0]) == 1
+        assert len(adaptation[1]) == 2
+
+    def test_multiple(self):
+        adaptation = fs.adaptation.get_details_by_location([1935265, 1714000], "city")
+        assert len(adaptation[0]) == 2
+        assert len(adaptation[1]) == 5
+
+    def test_single_csv(self, tmpdir):
+        adaptation = fs.adaptation.get_details_by_location([1935265], "city", csv=True, output_dir=tmpdir)
+        assert len(adaptation[0]) == 1
+        assert len(adaptation[1]) == 2
+
+    def test_multiple_csv(self, tmpdir):
+        adaptation = fs.adaptation.get_details_by_location([1935265, 1714000], "city", csv=True, output_dir=tmpdir)
+        assert len(adaptation[0]) == 2
+        assert len(adaptation[1]) == 5
+
+    def test_mixed_invalid(self):
+        adaptation = fs.adaptation.get_details_by_location([1935265, 000000000], "city")
+        assert len(adaptation[0]) == 2
+        assert len(adaptation[1]) == 2
+
+    def test_mixed_invalid_csv(self, tmpdir):
+        adaptation = fs.adaptation.get_details_by_location([1935265, 000000000], "city", csv=True, output_dir=tmpdir)
+        assert len(adaptation[0]) == 2
+        assert len(adaptation[1]) == 2

tests/api/test_historic.py

@@ -105,3 +105,68 @@ def test_mixed_invalid(self):
     def test_mixed_invalid_csv(self, tmpdir):
         historic = fs.historic.get_summary([190836953, 000000000], "property", csv=True, output_dir=tmpdir)
         assert len(historic) == 2
+
+
+class TestHistoricSummaryDetail:
+
+    def test_empty(self):
+        with pytest.raises(InvalidArgument):
+            fs.historic.get_events_by_location([], "")
+
+    def test_empty_fsid(self):
+        with pytest.raises(InvalidArgument):
+            fs.historic.get_events_by_location([], "property")
+
+    def test_empty_type(self):
+        with pytest.raises(InvalidArgument):
+            fs.historic.get_events_by_location([190836953], "")
+
+    def test_wrong_fsid_type(self):
+        with pytest.raises(InvalidArgument):
+            fs.historic.get_events_by_location(190836953, "city")
+
+    def test_wrong_fsid_number(self):
+        historic = fs.historic.get_events_by_location([11], "city")
+        assert len(historic[0]) == 1
+        assert len(historic[1]) == 1
+        assert historic[0][0].historic is None
+
+    def test_incorrect_lookup_type(self, tmpdir):
+        historic = fs.historic.get_events_by_location([1982200], "state", csv=True, output_dir=tmpdir)
+        assert len(historic[0]) == 1
+        assert len(historic[1]) == 1
+        assert historic[0][0].historic is None
+
+    def test_wrong_historic_type(self):
+        with pytest.raises(TypeError):
+            fs.historic.get_events_by_location([1982200], 190)
+
+    def test_single(self):
+        historic = fs.historic.get_events_by_location([1982200], "city")
+        assert len(historic[0]) == 1
+        assert len(historic[1]) == 1
+
+    def test_multiple(self):
+        historic = fs.historic.get_events_by_location([1982200, 3905074], "city")
+        assert len(historic[0]) == 2
+        assert len(historic[1]) == 2
+
+    def test_single_csv(self, tmpdir):
+        historic = fs.historic.get_events_by_location([1982200], "city", csv=True, output_dir=tmpdir)
+        assert len(historic[0]) == 1
+        assert len(historic[1]) == 1
+
+    def test_multiple_csv(self, tmpdir):
+        historic = fs.historic.get_events_by_location([1982200, 3905074], "city", csv=True, output_dir=tmpdir)
+        assert len(historic[0]) == 2
+        assert len(historic[1]) == 2
+
+    def test_mixed_invalid(self):
+        historic = fs.historic.get_events_by_location([1982200, 000000000], "city")
+        assert len(historic[0]) == 2
+        assert len(historic[1]) == 1
+
+    def test_mixed_invalid_csv(self, tmpdir):
+        historic = fs.historic.get_events_by_location([1982200, 000000000], "city", csv=True, output_dir=tmpdir)
+        assert len(historic[0]) == 2
+        assert len(historic[1]) == 1