Add economic NFIP product

Author: Lyetenth

firststreet/__init__.py

@@ -13,8 +13,7 @@
 from firststreet.api.location import Location
 from firststreet.api.probability import Probability
 from firststreet.api.tile import Tile
-from firststreet.api.aal import AAL
-from firststreet.api.avm import AVM
+from firststreet.api.economic import AAL, AVM, Economic
 from firststreet.errors import MissingAPIKeyError
 from firststreet.http_util import Http
 
@@ -61,3 +60,4 @@ def __init__(self, api_key=None, connection_limit=100, rate_limit=20000, rate_pe
         self.tile = Tile(self.http)
         self.aal = AAL(self.http)
         self.avm = AVM(self.http)
+        self.economic = Economic(self.http)

firststreet/__main__.py

@@ -66,7 +66,8 @@
                                         'tile.get_historic_event',
                                         'aal.get_summary',
                                         'avm.get_avm',
-                                        'avm.get_provider']:
+                                        'avm.get_provider',
+                                        'economic.get_property_nfip']:
                 logging.error("Product not found. Please check that the argument"
                               " provided is correct: {}".format(argument.product))
                 input("Press Enter to continue...")
@@ -387,6 +388,12 @@
                                         output_dir=argument.output_dir,
                                         extra_param=formatted_params)
 
+                elif argument.product == 'economic.get_property_nfip':
+                    fs.economic.get_property_nfip(search_items,
+                                                  csv=True,
+                                                  output_dir=argument.output_dir,
+                                                  extra_param=formatted_params)
+
                 else:
                     logging.error("Product not found. Please check that the argument"
                                   " provided is correct: {}".format(argument.product))

firststreet/api/aal.py

@@ -171,6 +171,15 @@ def to_csv(data, product, product_subtype, location_type=None, output_dir=None):
         else:
             raise NotImplementedError
 
+    elif product == 'economic':
+
+        if product_subtype == 'nfip':
+
+            df = format_economic_nfip_premium(data)
+
+        else:
+            raise NotImplementedError
+
     else:
         raise NotImplementedError
 
@@ -1280,3 +1289,29 @@ def format_avm_provider(data):
     df['provider_id'] = df['provider_id'].astype('Int64').apply(str)
 
     return df[['provider_id', 'valid_id', 'provider_name', "provider_logo", 'error']]
+
+
+def format_economic_nfip_premium(data):
+    """Reformat the list of data to AVM Provider format
+
+    Args:
+        data (list): A list of FSF object
+    Returns:
+        A pandas formatted DataFrame
+    """
+    df = pd.json_normalize([vars(o) for o in data]).explode("data")
+
+    if not df[['data']].isna().values.all():
+        df = pd.concat([df.drop(['data'], axis=1), df['data'].apply(pd.Series)], axis=1)
+    else:
+        df.drop(['data'], axis=1, inplace=True)
+        df['estimate'] = pd.NA
+        df['building'] = pd.NA
+        df['contents'] = pd.NA
+
+    df['fsid'] = df['fsid'].apply(str)
+    df['estimate'] = df['estimate'].astype('Int64').apply(str)
+    df['building'] = df['building'].astype('Int64').apply(str)
+    df['contents'] = df['contents'].astype('Int64').apply(str)
+
+    return df[['fsid', 'valid_id', 'estimate', "building", 'contents', 'error']]

firststreet/api/avm.py

@@ -0,0 +1,160 @@
+# Author: Kelvin Lai <kelvin@firststreet.org>
+# Copyright: This module is owned by First Street Foundation
+
+# Standard Imports
+import logging
+
+# Internal Imports
+from firststreet.api import csv_format
+from firststreet.api.api import Api
+from firststreet.errors import InvalidArgument
+from firststreet.models.economic import AVMProperty, AVMProvider, AALSummaryProperty, AALSummaryOther, NFIPPremium
+
+
+class AAL(Api):
+    """This class receives a list of search_items and handles the creation of an aal product from the request.
+
+        Methods:
+            get_summary: Retrieves a list of AAL Summary for the given list of IDs
+        """
+
+    def get_summary(self, search_items, location_type, csv=False, output_dir=None, extra_param=None):
+        """Retrieves AAL summary product data from the First Street Foundation API given a list of search_items and
+        returns a list of AAL Summary objects.
+
+        Args:
+            search_items (list/file): A First Street Foundation IDs, lat/lng pair, address, 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
+            output_dir (str): The output directory to save the generated csvs
+            extra_param (dict): Extra parameter to be added to the url
+
+        Returns:
+            A list of AAL Summary
+        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")
+
+        # Get data from api and create objects
+        if extra_param and "depths" in extra_param:
+            extra_param["depths"] = ','.join(map(str, extra_param["depths"]))
+
+        api_datas = self.call_api(search_items, "economic/aal", "summary", location_type, extra_param=extra_param)
+
+        product = []
+        for api_data, fsid in api_datas:
+            api_data["fsid"] = fsid
+
+            if location_type == "property":
+                product.append(AALSummaryProperty(api_data))
+            else:
+                product.append(AALSummaryOther(api_data))
+
+        if csv:
+            csv_format.to_csv(product, "economic_aal", "summary", location_type, output_dir=output_dir)
+
+        logging.info("AAL Summary Data Ready.")
+
+        return product
+
+
+class AVM(Api):
+    """This class receives a list of search_items and handles the creation of an AVM product from the request.
+
+        Methods:
+            get_avm: Retrieves a list of AVM for the given list of IDs
+            get_provider: Retrieves a list of AVM providers for the given list of IDs
+        """
+
+    def get_avm(self, search_items, csv=False, output_dir=None, extra_param=None):
+        """Retrieves AVM product data from the First Street Foundation API given a list of search_items and
+        returns a list of AVM objects.
+
+        Args:
+            search_items (list/file): A First Street Foundation IDs, lat/lng pair, address, or a
+                file of First Street Foundation IDs
+            csv (bool): To output extracted data to a csv or not
+            output_dir (str): The output directory to save the generated csvs
+            extra_param (dict): Extra parameter to be added to the url
+
+        Returns:
+            A list of AVM
+        """
+        # Get data from api and create objects
+        api_datas = self.call_api(search_items, "economic", "avm", "property", extra_param=extra_param)
+
+        product = [AVMProperty(api_data) for api_data in api_datas]
+
+        if csv:
+            csv_format.to_csv(product, "economic_avm", "avm", "property", output_dir=output_dir)
+
+        logging.info("AVM Data Ready.")
+
+        return product
+
+    def get_provider(self, search_items, csv=False, output_dir=None, extra_param=None):
+        """Retrieves AVM provider product data from the First Street Foundation API given a list of search_items and
+        returns a list of AVM provider objects.
+
+        Args:
+            search_items (list/file): A First Street Foundation IDs, lat/lng pair, address, or a
+                file of First Street Foundation IDs
+            csv (bool): To output extracted data to a csv or not
+            output_dir (str): The output directory to save the generated csvs
+            extra_param (dict): Extra parameter to be added to the url
+
+        Returns:
+            A list of AVM Provider
+        """
+        # Get data from api and create objects
+        api_datas = self.call_api(search_items, "economic/avm", "provider", extra_param=extra_param)
+
+        product = [AVMProvider(api_data) for api_data in api_datas]
+
+        if csv:
+            csv_format.to_csv(product, "economic_avm", "provider", output_dir=output_dir)
+
+        logging.info("AVM Provider Data Ready.")
+
+        return product
+
+
+class Economic(Api):
+    """This class receives a list of search_items and handles the creation of an economic product from the request.
+
+        Methods:
+            get_property_nfip: Retrieves a list of property nfip premiums for the given list of IDs
+        """
+
+    def get_property_nfip(self, search_items, csv=False, output_dir=None, extra_param=None):
+        """Retrieves AVM product data from the First Street Foundation API given a list of search_items and
+        returns a list of AVM objects.
+
+        Args:
+            search_items (list/file): A First Street Foundation IDs, lat/lng pair, address, or a
+                file of First Street Foundation IDs
+            csv (bool): To output extracted data to a csv or not
+            output_dir (str): The output directory to save the generated csvs
+            extra_param (dict): Extra parameter to be added to the url
+
+        Returns:
+            A list of property NFIP premiums
+        """
+        # Get data from api and create objects
+        api_datas = self.call_api(search_items, "economic", "nfip", "property", extra_param=extra_param)
+
+        product = [NFIPPremium(api_data) for api_data in api_datas]
+
+        if csv:
+            csv_format.to_csv(product, "economic", "nfip", "property", output_dir=output_dir)
+
+        logging.info("NFIP Premium Data Ready.")
+
+        return product