"""
EPA Envirofacts tools for ToolUniverse — US environmental facility data.
The U.S. EPA Envirofacts REST service exposes regulated-facility data, including
the Toxics Release Inventory (TRI) and the Facility Registry Service (FRS). These
tools query the two validated facility tables by state (and optional city).
API: https://data.epa.gov/efservice/{table}/{col}/{val}.../rows/{a}:{b}/JSON
(public, no authentication, US Gov public domain)
"""
from typing import Any, Dict
from urllib.parse import quote
import requests
from .base_tool import BaseTool
from .tool_registry import register_tool
EFSERVICE = "https://data.epa.gov/efservice"
class _EnvirofactsBase(BaseTool):
table = ""
state_col = ""
city_col = "city_name"
def __init__(self, tool_config: Dict[str, Any]):
super().__init__(tool_config)
self.fields = tool_config.get("fields", {}) or {}
self.timeout = self.fields.get("timeout", 30)
def run(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
# A config can opt into a single-column lookup mode (e.g. query the
# tri_reporting_form table by tri_facility_id) by declaring a
# "lookup_column" in its "fields". When absent, fall back to the
# original state/city facility-search behavior.
if self.fields.get("lookup_column"):
return self._run_lookup(arguments)
return self._run_state_search(arguments)
def _run_lookup(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
table = self.fields.get("table") or self.table
column = self.fields["lookup_column"]
param = self.fields.get("lookup_param", column)
value = (arguments.get(param) or "").strip()
if not value:
return {
"status": "error",
"error": f"'{param}' is required",
}
try:
limit = max(1, min(int(arguments.get("limit") or 10), 100))
except (TypeError, ValueError):
limit = 10
try:
offset = max(0, int(arguments.get("offset") or 0))
except (TypeError, ValueError):
offset = 0
filter_path = f"{EFSERVICE}/{table}/{column}/{quote(value)}"
path = f"{filter_path}/rows/{offset}:{offset + limit - 1}/JSON"
total = self._fetch_count(filter_path)
rows = self._fetch(path)
if isinstance(rows, dict) and rows.get("status") == "error":
return rows
next_offset = offset + len(rows)
total_is_exact = total is not None
if total is None:
total = next_offset
has_more = next_offset < total if total_is_exact else len(rows) == limit
return {
"status": "success",
"data": [self._summarize(r) for r in rows if isinstance(r, dict)],
"metadata": {
"total_results": total,
"total_is_exact": total_is_exact,
"returned": len(rows),
"offset": offset,
"has_more": has_more,
"next_offset": next_offset if has_more else None,
"table": table,
"query": {param: value},
"source": "EPA Envirofacts",
},
}
def _run_state_search(self, arguments: Dict[str, Any]) -> Dict[str, Any]:
state = (arguments.get("state") or "").strip().upper()
if not state:
return {
"status": "error",
"error": "'state' (2-letter code, e.g. 'CA') is required",
}
# Fix-R15C-2: a full state name (e.g. "Louisiana") doesn't match
# any row in Envirofacts' 2-letter state_code column, so it
# silently returns an empty result set indistinguishable from "this
# state genuinely has zero facilities" -- confirmed live the same
# state has hundreds of real rows under its actual 2-letter code
# ("LA"). Reject non-2-letter input instead of querying it.
if len(state) != 2 or not state.isalpha():
return {
"status": "error",
"error": (
f"'state' must be a 2-letter US state code (e.g. 'CA', 'TX'), "
f"got {arguments.get('state')!r}"
),
}
city = (arguments.get("city") or "").strip()
try:
limit = max(1, min(int(arguments.get("limit") or 10), 100))
except (TypeError, ValueError):
limit = 10
try:
offset = max(0, int(arguments.get("offset") or 0))
except (TypeError, ValueError):
offset = 0
filter_path = f"{EFSERVICE}/{self.table}/{self.state_col}/{quote(state)}"
if city:
filter_path += f"/{self.city_col}/{quote(city.upper())}"
path = f"{filter_path}/rows/{offset}:{offset + limit - 1}/JSON"
total = self._fetch_count(filter_path)
rows = self._fetch(path)
if isinstance(rows, dict) and rows.get("status") == "error":
return rows
next_offset = offset + len(rows)
total_is_exact = total is not None
if total is None:
total = next_offset
has_more = next_offset < total if total_is_exact else len(rows) == limit
return {
"status": "success",
"data": [self._summarize(r) for r in rows if isinstance(r, dict)],
"metadata": {
"total_results": total,
"total_is_exact": total_is_exact,
"returned": len(rows),
"offset": offset,
"has_more": has_more,
"next_offset": next_offset if has_more else None,
"table": self.table,
"query": {"state": state, "city": city or None},
"source": "EPA Envirofacts",
},
}
def _fetch(self, path: str) -> Any:
try:
resp = requests.get(
path, headers={"Accept": "application/json"}, timeout=self.timeout
)
resp.raise_for_status()
rows = resp.json()
except requests.exceptions.Timeout:
return {
"status": "error",
"error": f"EPA Envirofacts request timed out after {self.timeout}s",
}
except requests.exceptions.RequestException as e:
return {"status": "error", "error": f"EPA Envirofacts request failed: {e}"}
except ValueError:
return {
"status": "error",
"error": "EPA Envirofacts returned a non-JSON response",
}
if not isinstance(rows, list):
return []
return rows
def _fetch_count(self, filter_path: str) -> int | None:
"""Return Envirofacts' exact count without failing an otherwise valid page."""
count_rows = self._fetch(f"{filter_path}/count/JSON")
if (
isinstance(count_rows, list)
and count_rows
and isinstance(count_rows[0], dict)
):
raw_count = count_rows[0].get("TOTALQUERYRESULTS")
try:
return int(raw_count)
except (TypeError, ValueError):
pass
return None
@staticmethod
def _summarize(
r: Dict[str, Any],
) -> Dict[str, Any]: # pragma: no cover - overridden
return r
