+# -*- coding: utf-8 -*-
+"""ELM Ordinance full processing logic"""
+import os
+import time
+import json
+import asyncio
+import logging
+from datetime import datetime, timedelta
+from pathlib import Path
+from functools import partial
+
+import openai
+import pandas as pd
+from langchain.text_splitter import RecursiveCharacterTextSplitter
+
+from elm import ApiBase
+from elm.ords.download import download_county_ordinance
+from elm.ords.extraction import (
+ extract_ordinance_text_with_ngram_validation,
+ extract_ordinance_values,
+)
+from elm.ords.services.usage import UsageTracker
+from elm.ords.services.openai import OpenAIService, usage_from_response
+from elm.ords.services.provider import RunningAsyncServices
+from elm.ords.services.threaded import (
+ TempFileCache,
+ FileMover,
+ CleanedFileWriter,
+ OrdDBFileWriter,
+ UsageUpdater,
+)
+from elm.ords.services.cpu import PDFLoader, read_pdf_doc, read_pdf_doc_ocr
+from elm.ords.utilities import (
+ RTS_SEPARATORS,
+ load_all_county_info,
+ load_counties_from_fp,
+)
+from elm.ords.utilities.location import County
+from elm.ords.utilities.queued_logging import (
+ LocationFileLog,
+ LogListener,
+ NoLocationFilter,
+)
+
+logger = logging.getLogger(__name__)
+
+
+OUT_COLS = [
+ "county",
+ "state",
+ "FIPS",
+ "feature",
+ "fixed_value",
+ "mult_value",
+ "mult_type",
+ "adder",
+ "min_dist",
+ "max_dist",
+ "value",
+ "units",
+ "ord_year",
+ "last_updated",
+ "section",
+ "source",
+ "comment",
+]
+
+CHECK_COLS = [
+ "fixed_value",
+ "mult_value",
+ "adder",
+ "min_dist",
+ "max_dist",
+ "value",
+]
+
+
+
+
[docs]
+
async def process_counties_with_openai(
+
out_dir,
+
county_fp=None,
+
model="gpt-4",
+
azure_api_key=None,
+
azure_version=None,
+
azure_endpoint=None,
+
llm_call_kwargs=None,
+
llm_service_rate_limit=4000,
+
text_splitter_chunk_size=3000,
+
text_splitter_chunk_overlap=300,
+
num_urls_to_check_per_county=5,
+
max_num_concurrent_browsers=10,
+
file_loader_kwargs=None,
+
pytesseract_exe_fp=None,
+
td_kwargs=None,
+
tpe_kwargs=None,
+
ppe_kwargs=None,
+
log_dir=None,
+
clean_dir=None,
+
county_ords_dir=None,
+
county_dbs_dir=None,
+
log_level="INFO",
+
):
+
"""Download and extract ordinances for a list of counties.
+
+
Parameters
+
----------
+
out_dir : path-like
+
Path to output directory. This directory will be created if it
+
does not exist. This directory will contain the structured
+
ordinance output CSV as well as all of the scraped ordinance
+
documents (PDFs and HTML text files). Usage information and
+
default options for log/clean directories will also be stored
+
here.
+
county_fp : path-like, optional
+
Path to CSV file containing a list of counties to extract
+
ordinance information for. This CSV should have "County" and
+
"State" columns that contains the county and state names.
+
By default, ``None``, which runs the extraction for all known
+
counties (this is untested and not currently recommended).
+
model : str, optional
+
Name of LLM model to perform scraping. By default, ``"gpt-4"``.
+
azure_api_key : str, optional
+
Azure OpenAI API key. By default, ``None``, which pulls the key
+
from the environment variable ``AZURE_OPENAI_API_KEY`` instead.
+
azure_version : str, optional
+
Azure OpenAI API version. By default, ``None``, which pulls the
+
version from the environment variable ``AZURE_OPENAI_VERSION``
+
instead.
+
azure_endpoint : str, optional
+
Azure OpenAI API endpoint. By default, ``None``, which pulls the
+
endpoint from the environment variable ``AZURE_OPENAI_ENDPOINT``
+
instead.
+
llm_call_kwargs : dict, optional
+
Keyword-value pairs used to initialize an
+
`elm.ords.llm.LLMCaller` instance. By default, ``None``.
+
llm_service_rate_limit : int, optional
+
Token rate limit of LLm service being used (OpenAI).
+
By default, ``4000``.
+
text_splitter_chunk_size : int, optional
+
Chunk size input to
+
`langchain.text_splitter.RecursiveCharacterTextSplitter`.
+
By default, ``3000``.
+
text_splitter_chunk_overlap : int, optional
+
Chunk overlap input to
+
`langchain.text_splitter.RecursiveCharacterTextSplitter`.
+
By default, ``300``.
+
num_urls_to_check_per_county : int, optional
+
Number of unique Google search result URL's to check for
+
ordinance document. By default, ``5``.
+
max_num_concurrent_browsers : int, optional
+
Number of unique concurrent browser instances to open when
+
performing Google search. Setting this number too high on a
+
machine with limited processing can lead to increased timeouts
+
and therefore decreased quality of Google search results.
+
By default, ``10``.
+
pytesseract_exe_fp : path-like, optional
+
Path to pytesseract executable. If this option is specified, OCR
+
parsing for PDf files will be enabled via pytesseract.
+
By default, ``None``.
+
td_kwargs : dict, optional
+
Keyword-value argument pairs to pass to
+
:class:`tempfile.TemporaryDirectory`. The temporary directory is
+
used to store files downloaded from the web that are still being
+
parsed for ordinance information. By default, ``None``.
+
tpe_kwargs : dict, optional
+
Keyword-value argument pairs to pass to
+
:class:`concurrent.futures.ThreadPoolExecutor`. The thread pool
+
executor is used to run I/O intensive tasks like writing to a
+
log file. By default, ``None``.
+
ppe_kwargs : dict, optional
+
Keyword-value argument pairs to pass to
+
:class:`concurrent.futures.ProcessPoolExecutor`. The process
+
pool executor is used to run CPU intensive tasks like loading
+
a PDF file. By default, ``None``.
+
log_dir : path-like, optional
+
Path to directory for log files. This directory will be created
+
if it does not exist. By default, ``None``, which
+
creates a ``logs`` folder in the output directory for the
+
county-specific log files.
+
clean_dir : path-like, optional
+
Path to directory for cleaned ordinance text output. This
+
directory will be created if it does not exist. By default,
+
``None``, which creates a ``clean`` folder in the output
+
directory for the cleaned ordinance text files.
+
county_ords_dir : path-like, optional
+
Path to directory for individual county ordinance file outputs.
+
This directory will be created if it does not exist.
+
By default, ``None``, which creates a ``county_ord_files``
+
folder in the output directory.
+
county_dbs_dir : path-like, optional
+
Path to directory for individual county ordinance database
+
outputs. This directory will be created if it does not exist.
+
By default, ``None``, which creates a ``county_dbs`` folder in
+
the output directory.
+
log_level : str, optional
+
Log level to set for county retrieval and parsing loggers.
+
By default, ``"INFO"``.
+
+
Returns
+
-------
+
pd.DataFrame
+
DataFrame of parsed ordinance information. This file will also
+
be stored in the output directory under "wind_db.csv".
+
"""
+
start_time = time.time()
+
log_listener = LogListener(["elm"], level=log_level)
+
dirs = _setup_folders(
+
out_dir,
+
log_dir=log_dir,
+
clean_dir=clean_dir,
+
cod=county_ords_dir,
+
cdd=county_dbs_dir,
+
)
+
out_dir, log_dir, clean_dir, county_ords_dir, county_dbs_dir = dirs
+
async with log_listener as ll:
+
_setup_main_logging(log_dir, log_level, ll)
+
db = await _process_with_logs(
+
out_dir,
+
log_dir,
+
clean_dir,
+
county_ords_dir,
+
county_dbs_dir,
+
ll,
+
county_fp=county_fp,
+
model=model,
+
azure_api_key=azure_api_key,
+
azure_version=azure_version,
+
azure_endpoint=azure_endpoint,
+
llm_call_kwargs=llm_call_kwargs,
+
llm_service_rate_limit=llm_service_rate_limit,
+
text_splitter_chunk_size=text_splitter_chunk_size,
+
text_splitter_chunk_overlap=text_splitter_chunk_overlap,
+
num_urls_to_check_per_county=num_urls_to_check_per_county,
+
max_num_concurrent_browsers=max_num_concurrent_browsers,
+
file_loader_kwargs=file_loader_kwargs,
+
pytesseract_exe_fp=pytesseract_exe_fp,
+
td_kwargs=td_kwargs,
+
tpe_kwargs=tpe_kwargs,
+
ppe_kwargs=ppe_kwargs,
+
log_level=log_level,
+
)
+
_record_total_time(out_dir / "usage.json", time.time() - start_time)
+
return db
+
[docs]
+
async def download_docs_for_county_with_logging(
+
listener,
+
log_dir,
+
county,
+
text_splitter,
+
num_urls=5,
+
file_loader_kwargs=None,
+
browser_semaphore=None,
+
level="INFO",
+
**kwargs,
+
):
+
"""Retrieve ordinance document for a single county with async logs.
+
+
Parameters
+
----------
+
listener : elm.ords.utilities.queued_logging.LogListener
+
Active ``LogListener`` instance that can be passed to
+
:class:`elm.ords.utilities.queued_logging.LocationFileLog`.
+
log_dir : path-like
+
Path to output directory to contain log file.
+
county : elm.ords.utilities.location.Location
+
County to retrieve ordinance document for.
+
text_splitter : obj, optional
+
Instance of an object that implements a `split_text` method.
+
The method should take text as input (str) and return a list
+
of text chunks. Langchain's text splitters should work for this
+
input.
+
num_urls : int, optional
+
Number of unique Google search result URL's to check for
+
ordinance document. By default, ``5``.
+
file_loader_kwargs : dict, optional
+
Dictionary of keyword-argument pairs to initialize
+
:class:`elm.web.file_loader.AsyncFileLoader` with. The
+
"pw_launch_kwargs" key in these will also be used to initialize
+
the :class:`elm.web.google_search.PlaywrightGoogleLinkSearch`
+
used for the google URL search. By default, ``None``.
+
browser_semaphore : asyncio.Semaphore, optional
+
Semaphore instance that can be used to limit the number of
+
playwright browsers open concurrently. If ``None``, no limits
+
are applied. By default, ``None``.
+
level : str, optional
+
Log level to set for retrieval logger. By default, ``"INFO"``.
+
**kwargs
+
Keyword-value pairs used to initialize an
+
`elm.ords.llm.LLMCaller` instance.
+
+
Returns
+
-------
+
elm.web.document.BaseDocument | None
+
Document instance for the ordinance document, or ``None`` if no
+
document was found. Extracted ordinance information is stored in
+
the document's ``metadata`` attribute.
+
"""
+
with LocationFileLog(
+
listener, log_dir, location=county.full_name, level=level
+
):
+
task = asyncio.create_task(
+
download_doc_for_county(
+
county,
+
text_splitter,
+
num_urls=num_urls,
+
file_loader_kwargs=file_loader_kwargs,
+
browser_semaphore=browser_semaphore,
+
**kwargs,
+
),
+
name=county.full_name,
+
)
+
try:
+
doc, *__ = await asyncio.gather(task)
+
except KeyboardInterrupt:
+
raise
+
except Exception as e:
+
logger.error(
+
"Encountered error while processing %s:", county.full_name
+
)
+
logger.exception(e)
+
doc = None
+
+
return doc
+
[docs]
+
async def download_doc_for_county(
+
county,
+
text_splitter,
+
num_urls=5,
+
file_loader_kwargs=None,
+
browser_semaphore=None,
+
**kwargs,
+
):
+
"""Download and parse ordinance document for a single county.
+
+
Parameters
+
----------
+
county : elm.ords.utilities.location.Location
+
County to retrieve ordinance document for.
+
text_splitter : obj, optional
+
Instance of an object that implements a `split_text` method.
+
The method should take text as input (str) and return a list
+
of text chunks. Langchain's text splitters should work for this
+
input.
+
num_urls : int, optional
+
Number of unique Google search result URL's to check for
+
ordinance document. By default, ``5``.
+
file_loader_kwargs : dict, optional
+
Dictionary of keyword-argument pairs to initialize
+
:class:`elm.web.file_loader.AsyncFileLoader` with. The
+
"pw_launch_kwargs" key in these will also be used to initialize
+
the :class:`elm.web.google_search.PlaywrightGoogleLinkSearch`
+
used for the google URL search. By default, ``None``.
+
browser_semaphore : asyncio.Semaphore, optional
+
Semaphore instance that can be used to limit the number of
+
playwright browsers open concurrently. If ``None``, no limits
+
are applied. By default, ``None``.
+
**kwargs
+
Keyword-value pairs used to initialize an
+
`elm.ords.llm.LLMCaller` instance.
+
+
Returns
+
-------
+
elm.web.document.BaseDocument | None
+
Document instance for the ordinance document, or ``None`` if no
+
document was found. Extracted ordinance information is stored in
+
the document's ``metadata`` attribute.
+
"""
+
start_time = time.time()
+
doc = await download_county_ordinance(
+
county,
+
text_splitter,
+
num_urls=num_urls,
+
file_loader_kwargs=file_loader_kwargs,
+
browser_semaphore=browser_semaphore,
+
**kwargs,
+
)
+
if doc is None:
+
await _record_time_and_usage(start_time, **kwargs)
+
return None
+
+
doc.metadata["location"] = county
+
doc.metadata["location_name"] = county.full_name
+
await _record_usage(**kwargs)
+
+
doc = await extract_ordinance_text_with_ngram_validation(
+
doc, text_splitter, **kwargs
+
)
+
await _record_usage(**kwargs)
+
+
doc = await _write_cleaned_text(doc)
+
doc = await extract_ordinance_values(doc, **kwargs)
+
+
ord_count = _num_ords_in_doc(doc)
+
if ord_count > 0:
+
doc = await _move_file_to_out_dir(doc)
+
doc = await _write_ord_db(doc)
+
logger.info(
+
"%d ordinance value(s) found for %s. Outputs are here: '%s'",
+
ord_count,
+
county.full_name,
+
doc.metadata["ord_db_fp"],
+
)
+
else:
+
logger.info("No ordinances found for %s.", county.full_name)
+
+
await _record_time_and_usage(start_time, **kwargs)
+
return doc
