Skip to content
Draft
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 4 additions & 2 deletions .github/workflows/test.yml
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,9 @@ jobs:
python-version: "3.12"

- name: Install dependencies
run: pip install -r requirements.txt
run: |
python -m pip install --upgrade pip
pip install -r requirements-dev.txt

- name: Run tests
run: python -m pytest test_travel.py -v
run: python -m pytest -q
185 changes: 185 additions & 0 deletions app/engine/addressparse.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,185 @@
"""Graceful US address parsing for free-form starting locations.

Accepts comma-separated or run-on addresses and extracts street / city /
state / ZIP so Census geocoding can succeed without forcing users into a
single punctuation style.
"""

from __future__ import annotations

import re
from typing import TypedDict

from app.engine.states import STATE_PATTERN_2LETTER, US_STATE_NAMES

_STREET_SUFFIX = (
r"(?:street|st|avenue|ave|boulevard|blvd|road|rd|drive|dr|lane|ln|"
r"court|ct|circle|cir|way|place|pl|terrace|ter|parkway|pkwy|"
r"highway|hwy|trail|trl|pike)\.?"
)

_ZIP_RE = re.compile(r"\b(\d{5}(?:-\d{4})?)\b")


class AddressParts(TypedDict):
street: str
city: str
state: str
zip: str
formatted: str


def _extract_state(text: str) -> tuple[str, str]:
"""Return (state_code, remainder_without_state)."""
best_len = 0
best_code = ""
text_lower = text.lower()
for name, code in sorted(US_STATE_NAMES.items(), key=lambda x: -len(x[0])):
# Require word boundaries for short names like "dc"
pattern = r"\b" + re.escape(name) + r"\b"
if re.search(pattern, text_lower) and len(name) > best_len:
best_len = len(name)
best_code = code

if best_code:
for full_name, code in US_STATE_NAMES.items():
if code == best_code and len(full_name) == best_len:
remainder = re.sub(
r"\b" + re.escape(full_name) + r"\b",
"",
text,
count=1,
flags=re.IGNORECASE,
)
return best_code, re.sub(r"\s{2,}", " ", remainder).strip(" ,")
return best_code, text

state_match = re.search(STATE_PATTERN_2LETTER, text, flags=re.IGNORECASE)
if state_match:
code = state_match.group(1).upper()
remainder = re.sub(
r"\b" + re.escape(state_match.group(1)) + r"\b",
"",
text,
count=1,
flags=re.IGNORECASE,
)
return code, re.sub(r"\s{2,}", " ", remainder).strip(" ,")
return "", text


def _split_street_city(remainder: str) -> tuple[str, str]:
"""Split leftover text into (street, city) when possible."""
remainder = remainder.strip(" ,")
if not remainder:
return "", ""

# Street number + street name ending in a known suffix, then city
m = re.match(
rf"^(\d+\s+.{{1,80}}?\b{_STREET_SUFFIX})\s+(.+)$",
remainder,
flags=re.IGNORECASE,
)
if m:
return m.group(1).strip(" ,"), m.group(2).strip(" ,")

# Leading street number without recognized suffix: last token(s) = city
if re.match(r"^\d+", remainder):
tokens = remainder.split()
if len(tokens) >= 3:
# Prefer last 1 token as city; if last is short directional, take 2
if len(tokens) >= 4 and tokens[-1].lower() in {"n", "s", "e", "w", "ne", "nw", "se", "sw"}:
return " ".join(tokens[:-2]), " ".join(tokens[-2:])
return " ".join(tokens[:-1]), tokens[-1]
if len(tokens) == 2:
return tokens[0], tokens[1]

# No street number — treat whole remainder as city / place name
return "", remainder


def parse_us_address(raw: str) -> AddressParts:
"""Extract street, city, state, ZIP from free-form or structured input."""
empty: AddressParts = {
"street": "",
"city": "",
"state": "",
"zip": "",
"formatted": (raw or "").strip(),
}
if not raw or not str(raw).strip():
return empty

text = re.sub(r"\s+", " ", str(raw).strip())
text = text.replace(";", ",")

zip_code = ""
zip_m = _ZIP_RE.search(text)
if zip_m:
zip_code = zip_m.group(1)
text = (text[: zip_m.start()] + " " + text[zip_m.end() :]).strip(" ,")

state, remainder = _extract_state(text)
remainder = remainder.strip(" ,")

street = ""
city = ""

if "," in remainder:
parts = [p.strip() for p in remainder.split(",") if p.strip()]
if len(parts) >= 2:
# Heuristic: first part is street if it has a digit; else place,city
if re.search(r"\d", parts[0]):
street = parts[0]
city = parts[-1]
if len(parts) > 2 and not city:
city = parts[1]
else:
city = parts[-1]
street = ", ".join(parts[:-1]) if len(parts) > 1 else ""
if not re.search(r"\d", street):
# "Downtown, Orlando" style — keep last as city, drop street
city = parts[-1]
street = ""
elif len(parts) == 1:
street, city = _split_street_city(parts[0])
else:
street, city = _split_street_city(remainder)

# If state extraction ate everything and we only have zip, bail gracefully
formatted = format_address(street, city, state, zip_code) or str(raw).strip()
return {
"street": street,
"city": city,
"state": state,
"zip": zip_code,
"formatted": formatted,
}


def format_address(street: str, city: str, state: str, zip_code: str) -> str:
"""Build a canonical ``street, city, ST ZIP`` string from parts."""
chunks: list[str] = []
if street:
chunks.append(street.strip())
city_state = " ".join(x for x in [city.strip(), state.strip()] if x).strip()
if zip_code:
city_state = f"{city_state} {zip_code}".strip() if city_state else zip_code
if city_state:
chunks.append(city_state)
return ", ".join(chunks)


def normalize_us_address(raw: str) -> str:
"""Return a comma-structured address when components can be inferred.

Unparseable input is returned stripped unchanged (never raises).
"""
if not raw or not str(raw).strip():
return raw
parts = parse_us_address(raw)
# Only rewrite when we confidently found state or zip — otherwise keep raw
# so hotel/landmark names are not mangled.
if parts["state"] or parts["zip"]:
return parts["formatted"]
return str(raw).strip()
35 changes: 15 additions & 20 deletions app/engine/calendar.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,6 +9,7 @@
from __future__ import annotations

import logging
import re
from datetime import datetime, timedelta
from typing import Any

Expand Down Expand Up @@ -221,32 +222,26 @@ def _pick_event_date(plan: dict) -> datetime:


def _parse_to_datetime(time_str: str, base_date: datetime) -> datetime | None:
"""Parse a time string like '07:30 AM' or '08:00 AM +1' into a datetime."""
import re
"""Parse a free-form time string into a datetime on *base_date*."""
from app.engine.timeparse import normalize_time, parse_time

time_str = time_str.strip()
time_str = (time_str or "").strip()
if not time_str:
return None

# Handle "+1" suffix for next-day events
is_next_day = "+1" in time_str
time_str = time_str.replace("+1", "").strip()

m = re.match(r"(\d{1,2}):(\d{2})\s*(AM|PM)", time_str, re.IGNORECASE)
if not m:
# Handle "+1" / "+N" suffix for next-day events (also preserved by normalize)
is_next_day = bool(re.search(r"\+\d+", time_str))
normalized = normalize_time(time_str)
hour, minute = parse_time(normalized, default=(-1, -1))
if hour < 0:
return None

hour = int(m.group(1))
minute = int(m.group(2))
meridiem = m.group(3).upper()

if meridiem == "PM" and hour < 12:
hour += 12
if meridiem == "AM" and hour == 12:
hour = 0

result = base_date.replace(hour=hour, minute=minute, second=0, microsecond=0)
if is_next_day:
result = base_date.replace(hour=hour % 24, minute=minute, second=0, microsecond=0)
# Day offsets beyond +1 from normalize_time's fmt (+N) — apply full offset
day_m = re.search(r"\+(\d+)", normalized)
if day_m:
result += timedelta(days=int(day_m.group(1)))
elif is_next_day:
result += timedelta(days=1)
return result

Expand Down
50 changes: 24 additions & 26 deletions app/engine/osrm.py
Original file line number Diff line number Diff line change
Expand Up @@ -427,41 +427,31 @@ def _geocode_census(addr: str) -> tuple[float, float] | None:
Handles structured US addresses (street, city, state, zip) that
Nominatim often fails on. No API key, no rate limit.

Free-form input without commas (e.g. ``9801 International Dr Orlando FL
32819``) is normalized via :func:`parse_us_address` before the Census
request.

Returns (lat, lon) or None.
"""
import json
import urllib.request
import urllib.parse

# Parse address components from the input string.
parts = [p.strip() for p in addr.split(",")]
if len(parts) < 2:
return None
from app.engine.addressparse import parse_us_address

street = parts[0]
zip_match = re.search(r'\b(\d{5}(?:-\d{4})?)\b', parts[-1])
zip_code = zip_match.group(1) if zip_match else ""
has_street_number = bool(re.search(r'\d', parts[0]))
if not has_street_number and not zip_code:
return None
state, _remainder_state = _extract_state(parts[-1])
if len(parts) >= 3:
city = parts[-2].strip()
elif len(parts) == 2 and state:
if re.search(r'\d', parts[0]):
remainder = re.sub(r'\b\d{5}(?:-\d{4})?\b', '', parts[1])
remainder = re.sub(r'\b[A-Za-z]{2}\b', '', remainder)
city = remainder.strip().rstrip(",").strip()
else:
city = parts[0].strip()
street = ""
else:
city = ""
parts = parse_us_address(addr)
street = parts["street"]
city = parts["city"]
state = parts["state"]
zip_code = parts["zip"]

if not city or not state:
return None
has_street = bool(re.search(r'\d', street))
if has_street and not zip_code:
# Census structured lookup works best with a street; city+state+zip is ok
# for place-level matches when street is absent.
if street and not zip_code:
# Keep prior behavior: street-level queries without ZIP are unreliable
# on Census — fall through to Nominatim.
return None

params = urllib.parse.urlencode({
Expand All @@ -479,7 +469,10 @@ def _geocode_census(addr: str) -> tuple[float, float] | None:
data = json.loads(resp.read().decode())
if data.get("result", {}).get("addressMatches"):
coords = data["result"]["addressMatches"][0]["coordinates"]
logger.info("Census geocoded: %r → (%f, %f)", addr[:60], coords["y"], coords["x"])
logger.info(
"Census geocoded: %r → (%f, %f) [street=%r city=%r state=%r zip=%r]",
addr[:60], coords["y"], coords["x"], street, city, state, zip_code,
)
return (coords["y"], coords["x"])
return None
except Exception:
Expand Down Expand Up @@ -533,6 +526,11 @@ def _osrm_transit(origin: str, destination: str) -> dict[str, str] | None:
import json
import time

from app.engine.addressparse import normalize_us_address

origin = normalize_us_address(origin)
destination = normalize_us_address(destination)

# ── Step 1a: Geocode origin to determine country context ────────────
origin_coord = _geocode_census(origin)
if not origin_coord:
Expand Down
Loading
Loading