13938815212

Committed 19 Mar 2025 04:20AM UTC coverage: 94.635% (-0.6%) from 95.228%

Build # 13938815212

Build Type

Pull #9

github

Committed by

mcarans

Commit Message

Fix missing market

Pull Request Pull Request #9: HDXDSYS-1916 Fix missing markets

Run Details

4 of 7 new or added lines in 1 file covered. (57.14%)

441 of 466 relevant lines covered (94.64%)

0.95 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

89.02

/src/hdx/scraper/wfp/foodprices/wfp_food.py

import logging
from typing import Dict, List, Optional, Tuple

from .source_processing import process_source
from hdx.api.configuration import Configuration
from hdx.location.currency import Currency, CurrencyError
from hdx.location.wfp_api import WFPAPI
from hdx.utilities.dateparse import iso_string_from_datetime, parse_date
from hdx.utilities.dictandlist import dict_of_lists_add

logger = logging.getLogger(__name__)


class WFPFood:
    def __init__(
        self,
        countryiso3: str,
        configuration: Configuration,
        showcase_url: Optional[str],
        source: Optional[str],
        commodity_to_category: Dict[str, str],
    ):
        self._countryiso3 = countryiso3
        self._configuration = configuration
        self._showcase_url = showcase_url
        self._source = source
        self._commodity_to_category = commodity_to_category
        self._prices_data = []
        self._market_to_adm = {}

    def get_price_markets(self, wfp_api: WFPAPI) -> List[Dict]:
        dbmarkets = []
        prices_data = wfp_api.get_items(
            "MarketPrices/PriceMonthly", self._countryiso3
        )
        if not prices_data:
            logger.info(f"{self._countryiso3} has no prices data!")
            return dbmarkets
        self._prices_data = prices_data
        for market in wfp_api.get_items("Markets/List", self._countryiso3):
            market_id = market["marketId"]
            market_name = market["marketName"]
            admin1 = market["admin1Name"]
            admin2 = market["admin2Name"]
            latitude = market["marketLatitude"]
            longitude = market["marketLongitude"]
            self._market_to_adm[market_id] = (
                admin1,
                admin2,
                latitude,
                longitude,
            )
            dbmarkets.append(
                {
                    "market_id": market_id,
                    "market": market_name,
                    "countryiso3": self._countryiso3,
                    "admin1": admin1,
                    "admin2": admin2,
                    "latitude": latitude,
                    "longitude": longitude,
                }
            )
        logger.info(f"{len(prices_data)} prices rows")
        return dbmarkets

    def generate_rows(self, dbmarkets: List[Dict]) -> Tuple[Dict, Dict, Dict]:
        rows = {}
        markets = {}
        sources = {}
        for price_data in self._prices_data:
            priceflag = price_data["commodityPriceFlag"]
            if not all(
                x in ("actual", "aggregate") for x in priceflag.split(",")
            ):
                continue
            commodity_id = price_data["commodityID"]
            category = self._commodity_to_category[commodity_id]
            market_id = price_data["marketID"]
            market_name = price_data["marketName"]
            if market_name == "National Average":
                adm1 = adm2 = lat = lon = ""
            else:
                result = self._market_to_adm.get(market_id)
                if result:
                    adm1, adm2, lat, lon = result
                else:
                    adm1 = adm2 = lat = lon = ""
                    self._market_to_adm[market_id] = adm1, adm2, lat, lon
                    dbmarkets.append(
                        {
                            "market_id": market_id,
                            "market": market_name,
                            "countryiso3": self._countryiso3,
                        }
                    )

            process_source(sources, price_data["commodityPriceSourceName"])
            date_str = price_data["commodityPriceDate"]
            date = parse_date(date_str)
            date_str = iso_string_from_datetime(date)
            commodity = price_data["commodityName"]
            unit = price_data["commodityUnitName"]
            pricetype = price_data["priceTypeName"]
            price = price_data["commodityPrice"]
            currency = price_data["currencyName"]
            currency = self._configuration["currency_mappings"].get(
                currency, currency
            )
            try:
                usdprice = Currency.get_historic_value_in_usd(
                    price, currency, date
                )
            except (CurrencyError, ZeroDivisionError):
                usdprice = None
            key = (
                priceflag,
                date,
                adm1,
                adm2,
                market_name,
                category,
                commodity,
                unit,
                pricetype,
            )
            if key not in rows:
                rows[key] = (
                    date_str,
                    market_id,
                    lat,
                    lon,
                    commodity_id,
                    currency,
                    price,
                    usdprice,
                )
            if adm1 and adm2 and category and usdprice:
                adm1adm2market = adm1, adm2, market_name
                commodities = markets.get(adm1adm2market, {})
                dict_of_lists_add(
                    commodities,
                    (commodity, unit, pricetype, currency),
                    (date_str, usdprice),
                )
                markets[adm1adm2market] = commodities
        if rows:
            logger.info(
                f"{len(rows)} unique prices rows of price type actual or aggregate"
            )
        else:
            logger.info(f"{self._countryiso3} has no prices!")
        return rows, markets, sources

1	import logging	1✔
2	from typing import Dict, List, Optional, Tuple	1✔
3
4	from .source_processing import process_source	1✔
5	from hdx.api.configuration import Configuration	1✔
6	from hdx.location.currency import Currency, CurrencyError	1✔
7	from hdx.location.wfp_api import WFPAPI	1✔
8	from hdx.utilities.dateparse import iso_string_from_datetime, parse_date	1✔
9	from hdx.utilities.dictandlist import dict_of_lists_add	1✔
10
11	logger = logging.getLogger(__name__)	1✔
12
13
14	class WFPFood:	1✔
15	def __init__(	1✔
16	self,
17	countryiso3: str,
18	configuration: Configuration,
19	showcase_url: Optional[str],
20	source: Optional[str],
21	commodity_to_category: Dict[str, str],
22	):
23	self._countryiso3 = countryiso3	1✔
24	self._configuration = configuration	1✔
25	self._showcase_url = showcase_url	1✔
26	self._source = source	1✔
27	self._commodity_to_category = commodity_to_category	1✔
28	self._prices_data = []	1✔
29	self._market_to_adm = {}	1✔
30
31	def get_price_markets(self, wfp_api: WFPAPI) -> List[Dict]:	1✔
32	dbmarkets = []	1✔
33	prices_data = wfp_api.get_items(	1✔
34	"MarketPrices/PriceMonthly", self._countryiso3
35	)
36	if not prices_data:	1✔
37	logger.info(f"{self._countryiso3} has no prices data!")	×
38	return dbmarkets	×
39	self._prices_data = prices_data	1✔
40	for market in wfp_api.get_items("Markets/List", self._countryiso3):	1✔
41	market_id = market["marketId"]	1✔
42	market_name = market["marketName"]	1✔
43	admin1 = market["admin1Name"]	1✔
44	admin2 = market["admin2Name"]	1✔
45	latitude = market["marketLatitude"]	1✔
46	longitude = market["marketLongitude"]	1✔
47	self._market_to_adm[market_id] = (	1✔
48	admin1,
49	admin2,
50	latitude,
51	longitude,
52	)
53	dbmarkets.append(	1✔
54	{
55	"market_id": market_id,
56	"market": market_name,
57	"countryiso3": self._countryiso3,
58	"admin1": admin1,
59	"admin2": admin2,
60	"latitude": latitude,
61	"longitude": longitude,
62	}
63	)
64	logger.info(f"{len(prices_data)} prices rows")	1✔
65	return dbmarkets	1✔
66
67	def generate_rows(self, dbmarkets: List[Dict]) -> Tuple[Dict, Dict, Dict]:	1✔
68	rows = {}	1✔
69	markets = {}	1✔
70	sources = {}	1✔
71	for price_data in self._prices_data:	1✔
72	priceflag = price_data["commodityPriceFlag"]	1✔
73	if not all(	1✔
74	x in ("actual", "aggregate") for x in priceflag.split(",")
75	):
76	continue	1✔
77	commodity_id = price_data["commodityID"]	1✔
78	category = self._commodity_to_category[commodity_id]	1✔
79	market_id = price_data["marketID"]	1✔
80	market_name = price_data["marketName"]	1✔
81	if market_name == "National Average":	1✔
82	adm1 = adm2 = lat = lon = ""	×
83	else:
84	result = self._market_to_adm.get(market_id)	1✔
85	if result:	1✔
86	adm1, adm2, lat, lon = result	1✔
87	else:
NEW 88	adm1 = adm2 = lat = lon = ""	×
NEW 89	self._market_to_adm[market_id] = adm1, adm2, lat, lon	×
NEW 90	dbmarkets.append(	×
91	{
92	"market_id": market_id,
93	"market": market_name,
94	"countryiso3": self._countryiso3,
95	}
96	)
97
98	process_source(sources, price_data["commodityPriceSourceName"])	1✔
99	date_str = price_data["commodityPriceDate"]	1✔
100	date = parse_date(date_str)	1✔
101	date_str = iso_string_from_datetime(date)	1✔
102	commodity = price_data["commodityName"]	1✔
103	unit = price_data["commodityUnitName"]	1✔
104	pricetype = price_data["priceTypeName"]	1✔
105	price = price_data["commodityPrice"]	1✔
106	currency = price_data["currencyName"]	1✔
107	currency = self._configuration["currency_mappings"].get(	1✔
108	currency, currency
109	)
110	try:	1✔
111	usdprice = Currency.get_historic_value_in_usd(	1✔
112	price, currency, date
113	)
114	except (CurrencyError, ZeroDivisionError):	×
115	usdprice = None	×
116	key = (	1✔
117	priceflag,
118	date,
119	adm1,
120	adm2,
121	market_name,
122	category,
123	commodity,
124	unit,
125	pricetype,
126	)
127	if key not in rows:	1✔
128	rows[key] = (	1✔
129	date_str,
130	market_id,
131	lat,
132	lon,
133	commodity_id,
134	currency,
135	price,
136	usdprice,
137	)
138	if adm1 and adm2 and category and usdprice:	1✔
139	adm1adm2market = adm1, adm2, market_name	1✔
140	commodities = markets.get(adm1adm2market, {})	1✔
141	dict_of_lists_add(	1✔
142	commodities,
143	(commodity, unit, pricetype, currency),
144	(date_str, usdprice),
145	)
146	markets[adm1adm2market] = commodities	1✔
147	if rows:	1✔
148	logger.info(	1✔
149	f"{len(rows)} unique prices rows of price type actual or aggregate"
150	)
151	else:
152	logger.info(f"{self._countryiso3} has no prices!")	×
153	return rows, markets, sources	1✔

OCHA-DAP / hdx-scraper-wfp-foodprices / 13938815212

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous