API Reference¶

This page provides a complete reference for the SocialMapper Python API. For census variable details, see the Census Variables Reference.

Installation¶

pip install socialmapper

Quick Start¶

from socialmapper import SocialMapperClient

# Simple analysis
with SocialMapperClient() as client:
    result = client.analyze(
        location="Portland, OR",
        poi_type="amenity",
        poi_name="library",
        travel_time=15,
        census_variables=["total_population", "median_income"]
    )

    if result.is_ok():
        analysis = result.unwrap()
        print(f"Found {analysis.poi_count} libraries")
        print(f"Census units analyzed: {analysis.census_units_analyzed}")

Core Classes¶

SocialMapperClient¶

The main client for interacting with SocialMapper. Implements a context manager for proper resource handling.

from socialmapper import SocialMapperClient, ClientConfig

# Default configuration
with SocialMapperClient() as client:
    # Use the client
    pass

# Custom configuration
config = ClientConfig(
    api_key="your-census-api-key",
    rate_limit=5,      # requests per second
    timeout=600,       # seconds
    cache_strategy=None  # Optional cache implementation
)

with SocialMapperClient(config) as client:
    # Use the client
    pass

Methods¶

analyze()¶

Perform a simple analysis for a location and POI type.

def analyze(
    location: str,
    poi_type: str,
    poi_name: str,
    travel_time: int = 15,
    census_variables: Optional[List[str]] = None,
    travel_mode: str = "drive",
    geographic_level: str = "block-group",
    output_dir: Union[str, Path] = "output",
    **kwargs
) -> Result[AnalysisResult, Error]

Parameters: - location: Location in "City, State" format (e.g., "Portland, OR") - poi_type: OpenStreetMap key (e.g., "amenity", "leisure", "healthcare") - poi_name: OpenStreetMap value (e.g., "library", "park", "hospital") - travel_time: Travel time in minutes (1-120, default: 15) - census_variables: List of census variables to analyze (default: ["total_population"]) - travel_mode: Travel mode - "walk", "bike", or "drive" (default: "drive") - geographic_level: Geographic unit - "block-group" or "zcta" (default: "block-group") - output_dir: Directory for output files (default: "output")

Returns: Result[AnalysisResult, Error] - Success with analysis results or error

Example:

result = client.analyze(
    location="Seattle, WA",
    poi_type="leisure",
    poi_name="park",
    travel_time=10,
    census_variables=["total_population", "median_age", "median_income"]
)

create_analysis()¶

Create a builder for complex analysis configurations.

def create_analysis() -> SocialMapperBuilder

Returns: SocialMapperBuilder - Fluent builder for configuration

Example:

builder = client.create_analysis()
config = (builder
    .with_location("Chicago", "IL")
    .with_osm_pois("amenity", "school")
    .with_travel_time(20)
    .build()
)

run_analysis()¶

Run analysis with a configuration from the builder.

def run_analysis(
    config: Dict[str, Any],
    on_progress: Optional[Callable[[float], None]] = None
) -> Result[AnalysisResult, Error]

Parameters: - config: Configuration dictionary from builder - on_progress: Optional callback function for progress updates (0-100)

Returns: Result[AnalysisResult, Error] - Success with results or error

analyze_addresses()¶

Geocode and analyze a list of addresses.

def analyze_addresses(
    addresses: List[str],
    travel_time: int = 15,
    census_variables: Optional[List[str]] = None,
    **kwargs
) -> Result[AnalysisResult, Error]

Parameters: - addresses: List of address strings to geocode - travel_time: Travel time in minutes - census_variables: Census variables to analyze - **kwargs: Additional options

Returns: Result[AnalysisResult, Error] - Success with results or error

batch()¶

Create a batch context for running multiple analyses efficiently.

def batch() -> BatchContext

Example:

with client.batch() as batch:
    batch.add_analysis(config1)
    batch.add_analysis(config2)
    results = batch.run()

SocialMapperBuilder¶

Fluent builder for creating analysis configurations.

from socialmapper import SocialMapperBuilder, GeographicLevel

config = (SocialMapperBuilder()
    .with_location("Durham", "NC")
    .with_osm_pois("amenity", "library")
    .with_travel_time(20)
    .with_travel_mode("bike")
    .with_census_variables("total_population", "median_income")
    .with_geographic_level(GeographicLevel.ZCTA)
    .enable_isochrone_export()
    .with_output_directory("output/durham")
    .build()
)

Methods¶

with_location()¶

Set the geographic area for analysis.

def with_location(area: str, state: Optional[str] = None) -> Self

Parameters: - area: City, county, or area name - state: Optional state name or abbreviation

Examples:

.with_location("San Francisco", "CA")
.with_location("San Francisco, CA")
.with_location("Cook County", "IL")

with_osm_pois()¶

Configure OpenStreetMap POI search.

def with_osm_pois(
    poi_type: str,
    poi_name: str,
    additional_tags: Optional[Dict[str, str]] = None
) -> Self

Parameters: - poi_type: OSM key (e.g., "amenity", "leisure", "healthcare") - poi_name: OSM value (e.g., "library", "park", "hospital") - additional_tags: Optional additional OSM tags for filtering

Example:

.with_osm_pois("amenity", "school", {"school:type": "elementary"})

with_custom_pois()¶

Use custom POI coordinates from a file.

def with_custom_pois(
    file_path: Union[str, Path],
    name_field: Optional[str] = None,
    type_field: Optional[str] = None
) -> Self

Parameters: - file_path: Path to CSV or JSON file with POI data - name_field: Column name for POI names (auto-detected if not specified) - type_field: Column name for POI types (auto-detected if not specified)

Required columns: latitude, longitude

with_travel_time()¶

Set the travel time for isochrone generation.

def with_travel_time(minutes: int) -> Self

Parameters: - minutes: Travel time in minutes (1-120)

with_travel_mode()¶

Set the travel mode for routing.

def with_travel_mode(mode: Union[str, TravelMode]) -> Self

Parameters: - mode: "walk", "bike", "drive", or TravelMode enum value

with_census_variables()¶

Set census variables to analyze.

def with_census_variables(*variables: str) -> Self

Parameters: - *variables: Variable names or census codes

Example:

.with_census_variables("total_population", "median_income", "B01002_001E")

with_geographic_level()¶

Set the geographic unit for analysis.

def with_geographic_level(level: Union[str, GeographicLevel]) -> Self

Parameters: - level: "block-group", "zcta", or GeographicLevel enum value

enable_isochrone_export()¶

Enable saving of isochrone geometries to GeoParquet files.

def enable_isochrone_export() -> Self

When enabled, isochrone polygons are exported to the output/isochrones/ directory as GeoParquet files. The files are named using the pattern: {base_filename}_{travel_time}min_isochrones.geoparquet.

Example:

config = (SocialMapperBuilder()
    .with_location("Portland", "OR")
    .with_osm_pois("amenity", "library")
    .with_travel_time(15)
    .enable_isochrone_export()  # Exports isochrones to GeoParquet
    .build()
)

# After analysis, the isochrone file will be available at:
# output/isochrones/portland_amenity_library_15min_isochrones.geoparquet

Loading exported isochrones:

import geopandas as gpd

# Load the exported isochrone
isochrones = gpd.read_parquet("output/isochrones/portland_amenity_library_15min_isochrones.geoparquet")

# View the geometry
isochrones.plot()

disable_csv_export()¶

Disable CSV export (enabled by default).

def disable_csv_export() -> Self

with_output_directory()¶

Set the output directory for results.

def with_output_directory(path: Union[str, Path]) -> Self

limit_pois()¶

Limit the number of POIs to analyze.

def limit_pois(max_count: int) -> Self

build()¶

Build and validate the configuration.

def build() -> Dict[str, Any]

Returns: Configuration dictionary for use with run_analysis()

Raises: ValueError if configuration is invalid

Result Types¶

SocialMapper uses the Result pattern for explicit error handling without exceptions.

Result[T, E]¶

A generic result type that can contain either a success value (Ok) or an error (Err).

from socialmapper import Result, Ok, Err

# Pattern matching (Python 3.10+)
match result:
    case Ok(value):
        print(f"Success: {value}")
    case Err(error):
        print(f"Error: {error}")

# Traditional approach
if result.is_ok():
    value = result.unwrap()
else:
    error = result.unwrap_err()

Methods¶

is_ok() -> bool: Check if result is successful
is_err() -> bool: Check if result is an error
unwrap() -> T: Get success value (raises if error)
unwrap_err() -> E: Get error value (raises if success)
unwrap_or(default: T) -> T: Get value or default
map(func: Callable[[T], U]) -> Result[U, E]: Transform success value
map_err(func: Callable[[E], F]) -> Result[T, F]: Transform error value

AnalysisResult¶

Results from a successful analysis.

@dataclass
class AnalysisResult:
    poi_count: int                    # Number of POIs found
    isochrone_count: int             # Number of isochrones generated
    census_units_analyzed: int       # Number of census units analyzed
    files_generated: Dict[str, Path] # Paths to generated files
    metadata: Dict[str, Any]         # Additional metadata

    def is_complete(self) -> bool:
        """Check if analysis completed successfully."""
        return self.poi_count > 0 and self.isochrone_count > 0

Files Generated:

The files_generated dictionary may contain: - census_data: Path to the CSV file with census demographics - map: Path to the generated map image (if maps enabled) - isochrone_data: Path to the GeoParquet file with isochrone geometries (if isochrone export enabled)

Example:

if result.is_ok():
    analysis = result.unwrap()

    # Access generated files
    if 'census_data' in analysis.files_generated:
        print(f"Census data: {analysis.files_generated['census_data']}")

    if 'isochrone_data' in analysis.files_generated:
        print(f"Isochrone file: {analysis.files_generated['isochrone_data']}")

        # Load and use the isochrone
        import geopandas as gpd
        isochrones = gpd.read_parquet(analysis.files_generated['isochrone_data'])

Error¶

Structured error information.

@dataclass
class Error:
    type: ErrorType              # Error category
    message: str                 # Human-readable message
    context: Dict[str, Any]      # Additional context
    cause: Optional[Exception]   # Original exception if any
    traceback: Optional[str]     # Stack trace

ErrorType¶

Error categories for better error handling.

class ErrorType(Enum):
    VALIDATION = auto()         # Invalid input parameters
    NETWORK = auto()           # Network connectivity issues
    FILE_NOT_FOUND = auto()    # File or path not found
    PERMISSION_DENIED = auto() # Insufficient permissions
    RATE_LIMIT = auto()        # API rate limit exceeded
    CENSUS_API = auto()        # Census API errors
    OSM_API = auto()           # OpenStreetMap API errors
    GEOCODING = auto()         # Address geocoding failures
    PROCESSING = auto()        # General processing errors
    UNKNOWN = auto()           # Unclassified errors

Configuration Classes¶

ClientConfig¶

Configuration for the SocialMapper client.

@dataclass
class ClientConfig:
    api_key: Optional[str] = None       # Census API key
    rate_limit: float = 1.0             # Requests per second
    timeout: int = 600                  # Request timeout in seconds
    cache_strategy: Optional[CacheStrategy] = None  # Cache implementation
    max_retries: int = 3                # Maximum retry attempts
    log_level: str = "INFO"             # Logging level

GeographicLevel¶

Geographic units for census data.

class GeographicLevel(Enum):
    BLOCK_GROUP = "block-group"  # Census block groups (~600-3000 people)
    ZCTA = "zcta"               # ZIP Code Tabulation Areas

TravelMode¶

Available travel modes for routing.

class TravelMode(Enum):
    WALK = "walk"   # Walking (5 km/h)
    BIKE = "bike"   # Cycling (15 km/h)
    DRIVE = "drive" # Driving (uses road network)

Convenience Functions¶

High-level functions for common use cases.

quick_analysis()¶

Perform a quick analysis with minimal configuration.

def quick_analysis(
    location: str,
    poi_search: str,
    travel_time: int = 15,
    census_variables: Optional[List[str]] = None,
    output_dir: Union[str, Path] = "output"
) -> Result[AnalysisResult, Error]

Parameters: - location: Location in "City, State" format - poi_search: POI search in "type:name" format (e.g., "amenity:library") - travel_time: Travel time in minutes - census_variables: Census variables to analyze - output_dir: Output directory

Example:

from socialmapper import quick_analysis

result = quick_analysis(
    "Portland, OR",
    "amenity:school",
    travel_time=10,
    census_variables=["total_population", "median_income"]
)

analyze_location()¶

Analyze a specific location with custom POIs.

def analyze_location(
    city: str,
    state: str,
    poi_type: str = "amenity",
    poi_name: str = "library",
    **options
) -> Result[AnalysisResult, Error]

Parameters: - city: City name - state: State name or abbreviation - poi_type: OpenStreetMap POI type - poi_name: OpenStreetMap POI name - **options: Additional options (travel_time, census_variables, etc.)

analyze_custom_pois()¶

Analyze custom POIs from a file.

def analyze_custom_pois(
    poi_file: Union[str, Path],
    travel_time: int = 15,
    census_variables: Optional[List[str]] = None,
    name_field: Optional[str] = None,
    type_field: Optional[str] = None,
    **options
) -> Result[AnalysisResult, Error]

Parameters: - poi_file: Path to CSV or JSON file with POI coordinates - travel_time: Travel time in minutes - census_variables: Census variables to analyze - name_field: Field name for POI names - type_field: Field name for POI types - **options: Additional options

analyze_dataframe()¶

Analyze POIs from a pandas DataFrame.

def analyze_dataframe(
    df: pd.DataFrame,
    lat_col: str = "latitude",
    lon_col: str = "longitude",
    name_col: Optional[str] = "name",
    type_col: Optional[str] = "type",
    **options
) -> Result[AnalysisResult, Error]

Parameters: - df: DataFrame with POI data - lat_col: Column name for latitude - lon_col: Column name for longitude - name_col: Column name for POI names - type_col: Column name for POI types - **options: Additional analysis options

Async Support¶

SocialMapper provides async support for concurrent operations.

AsyncSocialMapper¶

Asynchronous client for streaming and concurrent operations.

from socialmapper import AsyncSocialMapper
import asyncio

async def main():
    async with AsyncSocialMapper(config) as mapper:
        # Stream POIs as they're found
        async for poi in mapper.stream_pois():
            print(f"Found: {poi.name}")

        # Generate isochrones with progress
        async for progress in mapper.generate_isochrones_with_progress():
            print(f"Progress: {progress.completed}/{progress.total}")

asyncio.run(main())

Error Handling¶

Basic Error Handling¶

result = client.analyze("Portland, OR", "amenity", "library")

if result.is_ok():
    analysis = result.unwrap()
    print(f"Success! Found {analysis.poi_count} POIs")
else:
    error = result.unwrap_err()
    print(f"Error: {error.message}")

    # Check error type
    if error.type == ErrorType.CENSUS_API:
        print("Census API error - check your API key")

Pattern Matching (Python 3.10+)¶

match result:
    case Ok(analysis):
        print(f"Found {analysis.poi_count} POIs")
    case Err(Error(type=ErrorType.RATE_LIMIT)):
        print("Rate limited - try again later")
    case Err(error):
        print(f"Other error: {error}")

Collecting Multiple Results¶

from socialmapper import collect_results

results = [
    client.analyze("Portland, OR", "amenity", "library"),
    client.analyze("Seattle, WA", "amenity", "school"),
    client.analyze("Eugene, OR", "leisure", "park")
]

match collect_results(results):
    case Ok(analyses):
        for analysis in analyses:
            print(f"Analysis complete: {analysis.poi_count} POIs")
    case Err(errors):
        for error in errors:
            print(f"Error: {error}")

Environment Variables¶

SocialMapper uses the following environment variables:

CENSUS_API_KEY: Your Census Bureau API key (required for census data)
SOCIALMAPPER_OUTPUT_DIR: Default output directory (default: "output")
SOCIALMAPPER_LOG_LEVEL: Logging level (default: "INFO")
SOCIALMAPPER_CACHE_DIR: Cache directory (default: "cache")

You can set these in a .env file:

CENSUS_API_KEY=your-api-key-here
SOCIALMAPPER_OUTPUT_DIR=/path/to/output
SOCIALMAPPER_LOG_LEVEL=DEBUG

Caching¶

SocialMapper supports pluggable caching strategies.

Implementing a Cache Strategy¶

from socialmapper import CacheStrategy
from typing import Any, Optional

class RedisCache(CacheStrategy):
    def get(self, key: str) -> Optional[Any]:
        # Implement Redis get
        pass

    def set(self, key: str, value: Any, ttl: Optional[int] = None):
        # Implement Redis set with TTL
        pass

# Use with client
config = ClientConfig(cache_strategy=RedisCache())

Rate Limiting¶

The client includes built-in rate limiting to respect API limits.

# Configure rate limit
config = ClientConfig(
    rate_limit=5.0  # 5 requests per second
)

# Rate limits are applied automatically
with SocialMapperClient(config) as client:
    # All API calls are rate limited
    result = client.analyze(...)

Logging¶

SocialMapper uses Python's standard logging module.

import logging

# Configure logging
logging.basicConfig(level=logging.DEBUG)

# Or set via environment variable
# SOCIALMAPPER_LOG_LEVEL=DEBUG

Examples¶

Complete Analysis Pipeline¶

from socialmapper import SocialMapperClient, SocialMapperBuilder

# Configure the analysis
config = (SocialMapperBuilder()
    .with_location("Austin", "TX")
    .with_osm_pois("amenity", "school")
    .with_travel_time(15)
    .with_travel_mode("walk")
    .with_census_variables(
        "total_population",
        "median_age",
        "median_income",
        "percent_poverty"
    )
    .with_geographic_level("block-group")
    .enable_isochrone_export()
    .with_output_directory("output/austin-schools")
    .build()
)

# Run the analysis
with SocialMapperClient() as client:
    result = client.run_analysis(
        config,
        on_progress=lambda p: print(f"Progress: {p:.1f}%")
    )

    match result:
        case Ok(analysis):
            print(f"Analysis complete!")
            print(f"Found {analysis.poi_count} schools")
            print(f"Analyzed {analysis.census_units_analyzed} census units")
            print(f"Files saved to: {analysis.files_generated}")
        case Err(error):
            print(f"Analysis failed: {error}")

Batch Processing¶

from socialmapper import SocialMapperClient

locations = [
    ("Portland", "OR"),
    ("Seattle", "WA"),
    ("San Francisco", "CA")
]

with SocialMapperClient() as client:
    with client.batch() as batch:
        for city, state in locations:
            config = (client.create_analysis()
                .with_location(city, state)
                .with_osm_pois("amenity", "hospital")
                .with_travel_time(20)
                .build()
            )
            batch.add_analysis(config)

        results = batch.run()

        for (city, state), result in zip(locations, results):
            match result:
                case Ok(analysis):
                    print(f"{city}: {analysis.poi_count} hospitals")
                case Err(error):
                    print(f"{city}: Failed - {error}")

Custom POI Analysis¶

from socialmapper import analyze_custom_pois

# Analyze custom locations from CSV
result = analyze_custom_pois(
    "my_locations.csv",
    travel_time=15,
    census_variables=["total_population", "median_age"],
    name_field="location_name",
    type_field="location_type"
)

if result.is_ok():
    analysis = result.unwrap()
    print(f"Analyzed {analysis.poi_count} custom locations")

Isochrone Export and Visualization¶

from socialmapper import SocialMapperClient, SocialMapperBuilder
import geopandas as gpd
import matplotlib.pyplot as plt

# Configure analysis with isochrone export
with SocialMapperClient() as client:
    config = (SocialMapperBuilder()
        .with_location("Portland", "OR")
        .with_osm_pois("amenity", "library")
        .with_travel_time(15)
        .with_travel_mode("walk")
        .with_census_variables("total_population", "median_income")
        .enable_isochrone_export()  # Enable isochrone export
        .build()
    )

    result = client.run_analysis(config)

# Process results
if result.is_ok():
    analysis = result.unwrap()
    print(f"Found {analysis.poi_count} libraries")
    print(f"Population within 15-min walk: {analysis.metadata.get('total_population_sum', 'N/A')}")

    # Load and visualize the exported isochrones
    if 'isochrone_data' in analysis.files_generated:
        isochrones = gpd.read_parquet(analysis.files_generated['isochrone_data'])

        # Create a visualization
        fig, ax = plt.subplots(figsize=(10, 10))
        isochrones.plot(ax=ax, alpha=0.5, color='blue', edgecolor='black')
        ax.set_title("15-minute walking isochrones from libraries in Portland, OR")
        plt.show()

        # Export to other formats for GIS software
        isochrones.to_file("portland_library_isochrones.geojson", driver="GeoJSON")
        print("Isochrones exported to GeoJSON for use in GIS software")
else:
    print(f"Analysis failed: {result.unwrap_err()}")

Version Information¶

import socialmapper

# Get version
print(socialmapper.__version__)

# Check API version
from socialmapper.api import __version__ as api_version
print(f"API version: {api_version}")

Migration from Legacy API¶

If you're using the old run_socialmapper function, migrate to the new client-based API:

# Old API (deprecated)
from socialmapper import run_socialmapper
results = run_socialmapper(state="CA", county="Los Angeles", ...)

# New API (recommended)
from socialmapper import SocialMapperClient
with SocialMapperClient() as client:
    result = client.analyze("Los Angeles, CA", "amenity", "library")
    if result.is_ok():
        analysis = result.unwrap()
        # Use analysis results

API Reference¶

Installation¶

Quick Start¶

Core Classes¶

SocialMapperClient¶

Methods¶

analyze()¶

create_analysis()¶

run_analysis()¶

analyze_addresses()¶

batch()¶

SocialMapperBuilder¶

Methods¶

with_location()¶

with_osm_pois()¶

with_custom_pois()¶

with_travel_time()¶

with_travel_mode()¶

with_census_variables()¶

with_geographic_level()¶

enable_isochrone_export()¶

disable_csv_export()¶

with_output_directory()¶

limit_pois()¶

build()¶

Result Types¶

Result[T, E]¶

Methods¶

AnalysisResult¶

Error¶

ErrorType¶

Configuration Classes¶

ClientConfig¶

GeographicLevel¶

TravelMode¶

Convenience Functions¶

quick_analysis()¶

analyze_location()¶

analyze_custom_pois()¶

analyze_dataframe()¶

Async Support¶

AsyncSocialMapper¶

Error Handling¶

Basic Error Handling¶

Pattern Matching (Python 3.10+)¶

Collecting Multiple Results¶

Environment Variables¶

Caching¶

Implementing a Cache Strategy¶

Rate Limiting¶

Logging¶

Examples¶

Complete Analysis Pipeline¶

Batch Processing¶

Custom POI Analysis¶

Isochrone Export and Visualization¶

Version Information¶

Migration from Legacy API¶

See Also¶