covid19-etl/etl/com_mobility.py
def parse_file(self, url):
        """
        Converts a CSV file to data we can submit via Sheepdog. Stores the
        records to submit in `self.location_data` and `self.time_series_data`.
        Ignores any records that are already in Sheepdog (relies on unique
        `submitter_id` to check)

        Args:
            url (str): URL at which the CSV file is available
        """

        self.last_submission_date_time = self.metadata_helper.get_last_submission()
        the_lattest_data_datetime = None

        print("Getting data from {}".format(url))

        with closing(self.get(url, stream=True)) as r:
            f = (line.decode("utf-8") for line in r.iter_lines())
            reader = csv.reader(f, delimiter=",", quotechar='"')

            headers = next(reader)

            assert (
                headers[0] != "404: Not Found"
            ), "Unable to get file contents, received {}.".format(headers)

            assert set(self.expected_file_headers).issubset(
                set(headers)
            ), "CSV headers have changed (expected {} is a subset of {}). We may need to update the ETL code".format(
                self.expected_file_headers, headers
            )

            for row in reader:
                # ignore any empty row
                if not row:
                    continue

                row_dict = dict(zip(headers, row))
                if row_dict["country_region_code"] != "US":
                    continue

                if (
                    not self.last_submission_date_time
                    or parse(row_dict["date"]) > self.last_submission_date_time
                ):
                    if (
                        the_lattest_data_datetime is None
                        or the_lattest_data_datetime < parse(row_dict["date"])
                    ):
                        the_lattest_data_datetime = parse(row_dict["date"])

                    summary_location = {}
                    summary_socio_demographic = {}

                    summary_location_submitter_id = format_submitter_id(
                        "summary_location",
                        row_dict["country_region_code"],
                        row_dict["sub_region_1"],
                        row_dict["sub_region_2"],
                        row_dict["metro_area"],
                        row_dict["date"],
                    )

                    summary_socio_demographic_submitter_id = format_submitter_id(
                        "summary_socio_demographic",
                        row_dict["country_region_code"],
                        row_dict["sub_region_1"],
                        row_dict["sub_region_2"],
                        row_dict["metro_area"],
                        row_dict["date"],
                    )

                    summary_location = {
                        "submitter_id": summary_location_submitter_id,
                        "projects": [{"code": self.project_code}],
                    }

                    summary_socio_demographic = {
                        "submitter_id": summary_socio_demographic_submitter_id,
                        "summary_locations": [
                            {"submitter_id": summary_location_submitter_id}
                        ],
                    }

                    for field in [
                        "country_region_code",
                        "country_region",
                        "sub_region_1",
                        "sub_region_2",
                        "metro_area",
                        "iso_3166_2_code",
                        "census_fips_code",
                    ]:
                        gen3_field, func = SPECIAL_MAP_FIELDS[field]
                        summary_location[gen3_field] = func(row_dict[field])

                    for field in [
                        "retail_and_recreation_percent_change_from_baseline",
                        "grocery_and_pharmacy_percent_change_from_baseline",
                        "parks_percent_change_from_baseline",
                        "transit_stations_percent_change_from_baseline",
                        "workplaces_percent_change_from_baseline",
                        "residential_percent_change_from_baseline",
                        "date",
                    ]:
                        gen3_field, func = SPECIAL_MAP_FIELDS[field]
                        summary_socio_demographic[gen3_field] = func(row_dict[field])

                    self.summary_locations.append(summary_location)
                    self.summary_socio_demographics.append(summary_socio_demographic)
        if the_lattest_data_datetime:
            self.last_submission_date_time = the_lattest_data_datetime

        Converts a CSV file to data we can submit via Sheepdog. Stores the
        records to submit in `self.location_data` and `self.time_series_data`.
        Ignores any records that are already in Sheepdog (relies on unique
        `submitter_id` to check)

        Args:
            url (str): URL at which the CSV file is available
        
Converts a CSV file to data we can submit via Sheepdog.
Converts a CSV file to data we can submit via Sheepdog .
def parse_file(self, url):
        self.last_submission_date_time = self.metadata_helper.get_last_submission()
        the_lattest_data_datetime = None
        print("Getting data from {}".format(url))
        with closing(self.get(url, stream=True)) as r:
            f = (line.decode("utf-8") for line in r.iter_lines())
            reader = csv.reader(f, delimiter=",", quotechar='"')
            headers = next(reader)
            assert (
                headers[0] != "404: Not Found"
            ), "Unable to get file contents, received {}.".format(headers)
            assert set(self.expected_file_headers).issubset(
                set(headers)
            ), "CSV headers have changed (expected {} is a subset of {}). We may need to update the ETL code".format(
                self.expected_file_headers, headers
            )
            for row in reader:
                if not row:
                    continue
                row_dict = dict(zip(headers, row))
                if row_dict["country_region_code"] != "US":
                    continue
                if (
                    not self.last_submission_date_time
                    or parse(row_dict["date"]) > self.last_submission_date_time
                ):
                    if (
                        the_lattest_data_datetime is None
                        or the_lattest_data_datetime < parse(row_dict["date"])
                    ):
                        the_lattest_data_datetime = parse(row_dict["date"])
                    summary_location = {}
                    summary_socio_demographic = {}
                    summary_location_submitter_id = format_submitter_id(
                        "summary_location",
                        row_dict["country_region_code"],
                        row_dict["sub_region_1"],
                        row_dict["sub_region_2"],
                        row_dict["metro_area"],
                        row_dict["date"],
                    )
                    summary_socio_demographic_submitter_id = format_submitter_id(
                        "summary_socio_demographic",
                        row_dict["country_region_code"],
                        row_dict["sub_region_1"],
                        row_dict["sub_region_2"],
                        row_dict["metro_area"],
                        row_dict["date"],
                    )
                    summary_location = {
                        "submitter_id": summary_location_submitter_id,
                        "projects": [{"code": self.project_code}],
                    }
                    summary_socio_demographic = {
                        "submitter_id": summary_socio_demographic_submitter_id,
                        "summary_locations": [
                            {"submitter_id": summary_location_submitter_id}
                        ],
                    }
                    for field in [
                        "country_region_code",
                        "country_region",
                        "sub_region_1",
                        "sub_region_2",
                        "metro_area",
                        "iso_3166_2_code",
                        "census_fips_code",
                    ]:
                        gen3_field, func = SPECIAL_MAP_FIELDS[field]
                        summary_location[gen3_field] = func(row_dict[field])
                    for field in [
                        "retail_and_recreation_percent_change_from_baseline",
                        "grocery_and_pharmacy_percent_change_from_baseline",
                        "parks_percent_change_from_baseline",
                        "transit_stations_percent_change_from_baseline",
                        "workplaces_percent_change_from_baseline",
                        "residential_percent_change_from_baseline",
                        "date",
                    ]:
                        gen3_field, func = SPECIAL_MAP_FIELDS[field]
                        summary_socio_demographic[gen3_field] = func(row_dict[field])
                    self.summary_locations.append(summary_location)
                    self.summary_socio_demographics.append(summary_socio_demographic)
        if the_lattest_data_datetime:
            self.last_submission_date_time = the_lattest_data_datetime
def parse_file ( self , url ) : self . last_submission_date_time = self . metadata_helper . get_last_submission ( ) the_lattest_data_datetime = None print ( "Getting data from {}" . format ( url ) ) with closing ( self . get ( url , stream = True ) ) as r : f = ( line . decode ( "utf-8" ) for line in r . iter_lines ( ) ) reader = csv . reader ( f , delimiter = "," , quotechar = '"' ) headers = next ( reader ) assert ( headers [ 0 ] != "404: Not Found" ) , "Unable to get file contents, received {}." . format ( headers ) assert set ( self . expected_file_headers ) . issubset ( set ( headers ) ) , "CSV headers have changed (expected {} is a subset of {}). We may need to update the ETL code" . format ( self . expected_file_headers , headers ) for row in reader : if not row : continue row_dict = dict ( zip ( headers , row ) ) if row_dict [ "country_region_code" ] != "US" : continue if ( not self . last_submission_date_time or parse ( row_dict [ "date" ] ) > self . last_submission_date_time ) : if ( the_lattest_data_datetime is None or the_lattest_data_datetime < parse ( row_dict [ "date" ] ) ) : the_lattest_data_datetime = parse ( row_dict [ "date" ] ) summary_location = { } summary_socio_demographic = { } summary_location_submitter_id = format_submitter_id ( "summary_location" , row_dict [ "country_region_code" ] , row_dict [ "sub_region_1" ] , row_dict [ "sub_region_2" ] , row_dict [ "metro_area" ] , row_dict [ "date" ] , ) summary_socio_demographic_submitter_id = format_submitter_id ( "summary_socio_demographic" , row_dict [ "country_region_code" ] , row_dict [ "sub_region_1" ] , row_dict [ "sub_region_2" ] , row_dict [ "metro_area" ] , row_dict [ "date" ] , ) summary_location = { "submitter_id" : summary_location_submitter_id , "projects" : [ { "code" : self . project_code } ] , } summary_socio_demographic = { "submitter_id" : summary_socio_demographic_submitter_id , "summary_locations" : [ { "submitter_id" : summary_location_submitter_id } ] , } for field in [ "country_region_code" , "country_region" , "sub_region_1" , "sub_region_2" , "metro_area" , "iso_3166_2_code" , "census_fips_code" , ] : gen3_field , func = SPECIAL_MAP_FIELDS [ field ] summary_location [ gen3_field ] = func ( row_dict [ field ] ) for field in [ "retail_and_recreation_percent_change_from_baseline" , "grocery_and_pharmacy_percent_change_from_baseline" , "parks_percent_change_from_baseline" , "transit_stations_percent_change_from_baseline" , "workplaces_percent_change_from_baseline" , "residential_percent_change_from_baseline" , "date" , ] : gen3_field , func = SPECIAL_MAP_FIELDS [ field ] summary_socio_demographic [ gen3_field ] = func ( row_dict [ field ] ) self . summary_locations . append ( summary_location ) self . summary_socio_demographics . append ( summary_socio_demographic ) if the_lattest_data_datetime : self . last_submission_date_time = the_lattest_data_datetime
Converts a CSV file to data we can submit via Sheepdog.
Converts a CSV file to data we can submit via Sheepdog .
"""
        Converts a CSV file to data we can submit via Sheepdog. Stores the
        records to submit in `self.location_data` and `self.time_series_data`.
        Ignores any records that are already in Sheepdog (relies on unique
        `submitter_id` to check)

        Args:
            url (str): URL at which the CSV file is available
        """ # ignore any empty row
self
url
URL at which the CSV file is available
URL at which the CSV file is available