Shelter Capacity & Resource Tracking Schemas
Problem Framing
Two hours after a levee overtops, the emergency operations center is standing up nine shelters across three counties, and the capacity picture is already wrong. One high-school gym reports “space available” on a whiteboard photo texted at 19:10 while a spreadsheet from the same site, emailed at 19:25, shows it at 140 of 120 cots — over capacity — because two agencies counted the same intake line. A partner county’s shelter has run its accessible beds to zero but still shows green on the map because the feed carries one headcount and no breakdown. Dispatch, reading the map, routes a bus of forty evacuees including three wheelchair users to a site that cannot take any of them. None of the source systems is broken; there is simply no schema that makes capacity, occupancy, and resource inventory mean the same thing across every agency writing to the map. This page specifies that schema and the deterministic Python that computes state from it, implementing the Core Emergency GIS Architecture & Data Standards contract for shelter and resource features that feed a shared Common Operating Picture (COP) under National Incident Management System (NIMS) resource typing and Federal Emergency Management Agency (FEMA) National Shelter System reporting.
Prerequisites
This workflow assumes a senior engineer’s fluency with the Python geospatial and validation stack and the following preconditions before the first shelter record is published:
- Packages:
pydantic >= 2.0for the schema contract,shapely >= 2.0for geometry construction, andgeopandas >= 0.12only where a shelter set is materialized to a layer. No heavyweight routing or raster dependency is required — capacity math is deliberately pure Python so it runs identically on a cloud node and a disconnected field tablet. - A geocoded, normalized shelter location. Every shelter is a point feature in a single canonical reference system. Address-to-point normalization is owned upstream by the Real-Time Geocoding & Location Normalization workflow; this stage assumes each record already carries a valid
[lon, lat]pair in EPSG:4326 and never re-geocodes. - A declared unit convention. Resource quantities must agree on units before they are summed: potable water in litres, meals as ready-to-eat equivalents, cots as integer beds. A mixed-unit inventory is a data defect, not a rounding problem, so the unit is part of the schema, not a comment.
- A lineage sink. Every published shelter feature emits provenance — reporting agency, source system, observation time, and the schema version that validated it — under the Emergency Metadata Standards contract, so a post-incident audit can reconstruct who reported which capacity value and when.
Schema and Capacity-State Model
A shelter feature is not one number on a map; it is a small graph of typed entities whose state must be computed the same way on every node. The governing entity is the shelter itself — a point geometry carrying several bounded capacity fields and an operational status. Hanging off it is a list of typed resource items (cots, potable water, meals, medical supplies, mobility equipment), each tagged with a NIMS kind and type code so mutual-aid partners count them in the same categories. The capacity status is never stored as an editable flag; it is derived from the ratio of current occupancy to maximum capacity through fixed thresholds, so two agencies holding the same record can never disagree about whether the shelter is full. The diagram below shows the data model on the left and the deterministic state ladder it drives on the right.
Step-by-Step Implementation
Step 1 — Define the shelter and resource schema contract
The contract is the boundary where an impossible capacity value is stopped. Model the shelter and each resource item as pydantic models with bounded integer fields and cross-field invariants — occupancy cannot be negative, an accessible-bed count cannot exceed the total, and committed resource quantity cannot exceed what is on hand. Carry the NIMS kind and type code on every resource so cots and potable water are counted in categories mutual-aid partners already recognize. Validate before any record reaches the datastore, so the map can never render a shelter holding negative people or 300 of 120 beds because of a typo in an intake spreadsheet.
import logging
from enum import Enum
from typing import List, Optional
from pydantic import BaseModel, Field, model_validator
logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s")
logger = logging.getLogger("incidentgis.shelter")
class ShelterStatus(str, Enum):
OPEN = "open"
NEAR_CAPACITY = "near_capacity"
FULL = "full"
OVER_CAPACITY = "over_capacity"
CLOSED = "closed"
class ShelterRejected(RuntimeError):
"""Raised when a shelter or resource record violates the schema contract."""
class ResourceItem(BaseModel):
resource_type: str # human label, e.g. "cot", "potable_water", "meal_rte"
nims_type: Optional[str] = None # NIMS resource kind/type code where the item is typed
quantity_on_hand: int = Field(ge=0)
quantity_committed: int = Field(ge=0)
reorder_threshold: int = Field(ge=0)
unit: str # litres, beds, rte_meals — no mixed units per type
@model_validator(mode="after")
def committed_within_hand(self) -> "ResourceItem":
# A committed count above what is physically present is a data defect, not a shortage.
if self.quantity_committed > self.quantity_on_hand:
raise ValueError(
f"{self.resource_type}: committed {self.quantity_committed} "
f"exceeds on-hand {self.quantity_on_hand}"
)
return self
class Shelter(BaseModel):
shelter_id: str
name: str
lon: float = Field(ge=-180.0, le=180.0)
lat: float = Field(ge=-90.0, le=90.0)
maximum_capacity: int = Field(ge=0)
current_occupancy: int = Field(ge=0)
ada_capacity: int = Field(ge=0, default=0)
medical_capacity: int = Field(ge=0, default=0)
pet_capacity: int = Field(ge=0, default=0)
is_open: bool = True
resources: List[ResourceItem] = Field(default_factory=list)
reporting_agency: str
observed_utc: str # ISO 8601, normalized upstream
@model_validator(mode="after")
def sub_capacities_within_total(self) -> "Shelter":
# Accessible and medical beds are carved from the governing total, never added on top.
for label, value in (("ada", self.ada_capacity), ("medical", self.medical_capacity)):
if value > self.maximum_capacity:
raise ValueError(
f"{label}_capacity {value} exceeds maximum_capacity {self.maximum_capacity}"
)
return self
def parse_shelter(record: dict) -> Shelter:
"""Validate a raw shelter record against the contract, failing closed on any violation."""
try:
shelter = Shelter(**record)
except ValueError as exc:
logger.error("Shelter rejected: %s", exc)
raise ShelterRejected(str(exc)) from exc
logger.info("Accepted shelter %s (%s)", shelter.shelter_id, shelter.name)
return shelter
Step 2 — Compute a deterministic occupancy state
Status must be a pure function of the committed integer fields, never a flag any node can toggle by hand. Compute the capacity state from the occupancy-to-maximum ratio using fixed thresholds and integer arithmetic on the boundary, so a shelter at exactly its maximum reads FULL rather than tipping between FULL and NEAR_CAPACITY on a floating-point rounding error. The closed operational flag is an override that always wins: a closed shelter is CLOSED regardless of how many cots remain, because dispatch must not route to it. Because the same inputs yield the same status on every node, this is the property that lets two agencies edit the same feature without disagreeing about whether it is full — the same determinism the Conflict Resolution in Multi-Agency Edits workflow depends on when it reconciles concurrent updates.
import logging
logger = logging.getLogger("incidentgis.shelter")
NEAR_CAPACITY_RATIO = 0.90 # fraction of maximum at which a shelter is flagged near capacity
def derive_status(shelter: "Shelter") -> ShelterStatus:
"""Deterministically map occupancy and the open flag to a capacity state.
Integer comparisons on the boundary avoid float rounding: FULL is exact
equality with the maximum, over capacity is a strict integer excess.
"""
if not shelter.is_open:
return ShelterStatus.CLOSED
maximum = shelter.maximum_capacity
occupied = shelter.current_occupancy
# A zero-capacity open shelter is degenerate; treat any occupant as over capacity.
if maximum <= 0:
logger.warning("Shelter %s open with zero maximum_capacity", shelter.shelter_id)
return ShelterStatus.OVER_CAPACITY if occupied > 0 else ShelterStatus.FULL
if occupied > maximum:
return ShelterStatus.OVER_CAPACITY
if occupied == maximum:
return ShelterStatus.FULL
# near_capacity boundary computed in integers: occupied * 100 >= ratio * maximum * 100
near_threshold = (int(NEAR_CAPACITY_RATIO * 100) * maximum)
if occupied * 100 >= near_threshold:
return ShelterStatus.NEAR_CAPACITY
return ShelterStatus.OPEN
def accessible_beds_remaining(shelter: "Shelter", ada_occupied: int) -> int:
"""Accessible beds are a separate constraint; a shelter can be OPEN yet ADA-full."""
if ada_occupied < 0:
raise ValueError("ada_occupied cannot be negative")
return max(0, shelter.ada_capacity - ada_occupied)
Step 3 — Raise over-capacity and resupply alerts
State on a map is passive; an alert is what actually moves a bus. Evaluate two independent rules per shelter and emit a structured event for each breach. The first is over-capacity: any shelter whose derived status is OVER_CAPACITY raises an alert carrying the overflow count so the operations center can open relief space. The second is resource sufficiency: for every typed resource, the available quantity is what is on hand minus what is already committed, and when that drops below the reorder threshold the shelter needs resupply before the next operational period. Both events are logged and returned as data, never printed, so they can be routed to the incident messaging bus and the audit trail alike.
import logging
from dataclasses import dataclass, field
from datetime import datetime, timezone
from typing import List
logger = logging.getLogger("incidentgis.shelter")
@dataclass(frozen=True)
class ShelterAlert:
shelter_id: str
reason_code: str # "over_capacity" | "resource_low"
detail: str
severity: str # "warning" | "critical"
raised_utc: str = field(default_factory=lambda: datetime.now(timezone.utc).isoformat())
def evaluate_alerts(shelter: "Shelter") -> List[ShelterAlert]:
"""Return over-capacity and per-resource resupply alerts for one shelter."""
alerts: List[ShelterAlert] = []
try:
status = derive_status(shelter)
except ValueError as exc:
# A record that cannot even be scored is itself a critical alert, not a silent drop.
logger.error("Cannot score shelter %s: %s", shelter.shelter_id, exc)
return [ShelterAlert(shelter.shelter_id, "scoring_error", str(exc), "critical")]
if status is ShelterStatus.OVER_CAPACITY:
overflow = shelter.current_occupancy - shelter.maximum_capacity
detail = f"{overflow} over the {shelter.maximum_capacity}-person maximum"
alerts.append(ShelterAlert(shelter.shelter_id, "over_capacity", detail, "critical"))
logger.warning("OVER CAPACITY %s: %s", shelter.shelter_id, detail)
for item in shelter.resources:
available = item.quantity_on_hand - item.quantity_committed
if available < item.reorder_threshold:
detail = (
f"{item.resource_type}: {available} {item.unit} available "
f"(< reorder {item.reorder_threshold})"
)
alerts.append(ShelterAlert(shelter.shelter_id, "resource_low", detail, "warning"))
logger.info("Resupply needed %s: %s", shelter.shelter_id, detail)
return alerts
Step 4 — Publish shelter features to the Common Operating Picture
The final stage serializes each validated, scored shelter into a GeoJSON feature the COP can render directly. The geometry is the normalized point; the properties carry the derived status, the raw capacity fields, and — critically — the lineage stamp so a reviewer can trace every rendered value back to the agency and observation time that produced it. Serialization never recomputes capacity from a display string; it reads the same integer fields Step 2 scored, so the published status and the alert stream can never diverge. This is the payload the ingestion boundary in Geospatial Data Ingestion Pipelines accepts and de-duplicates before it reaches the operational map.
import logging
from typing import Any, Dict
logger = logging.getLogger("incidentgis.shelter")
SCHEMA_VERSION = "shelter-schema/1.2.0"
def to_geojson_feature(shelter: "Shelter") -> Dict[str, Any]:
"""Serialize a shelter to a COP-ready GeoJSON feature with derived status and lineage."""
try:
status = derive_status(shelter)
except ValueError as exc:
logger.error("Refusing to publish unscored shelter %s: %s", shelter.shelter_id, exc)
raise ShelterRejected(f"cannot derive status for {shelter.shelter_id}") from exc
resources = [
{
"resource_type": r.resource_type,
"nims_type": r.nims_type,
"available": r.quantity_on_hand - r.quantity_committed,
"on_hand": r.quantity_on_hand,
"unit": r.unit,
}
for r in shelter.resources
]
feature = {
"type": "Feature",
"geometry": {"type": "Point", "coordinates": [shelter.lon, shelter.lat]},
"properties": {
"shelter_id": shelter.shelter_id,
"name": shelter.name,
"status": status.value,
"maximum_capacity": shelter.maximum_capacity,
"current_occupancy": shelter.current_occupancy,
"ada_capacity": shelter.ada_capacity,
"medical_capacity": shelter.medical_capacity,
"pet_capacity": shelter.pet_capacity,
"resources": resources,
# lineage — provenance for post-incident audit
"reporting_agency": shelter.reporting_agency,
"observed_utc": shelter.observed_utc,
"schema_version": SCHEMA_VERSION,
},
}
logger.info("Published shelter %s as %s", shelter.shelter_id, status.value)
return feature
Configuration Reference
Tune these parameters per deployment; a steady-state cloud node and a field-provisioned laptop will diverge, but every node in one incident must share the same threshold set or their derived statuses will disagree.
| Parameter | Env var | Default | Notes |
|---|---|---|---|
| Near-capacity ratio | SHELTER_NEAR_RATIO |
0.90 |
Fraction of maximum that flags NEAR_CAPACITY; lower it for slow-intake sites where lead time matters. |
| Schema version | SHELTER_SCHEMA_VERSION |
shelter-schema/1.2.0 |
Stamped into every feature’s lineage; bump on any field change so audits pin the contract. |
| Strict units | SHELTER_STRICT_UNITS |
true |
When true, a resource whose unit is unknown for its type is rejected, not coerced. |
| Zero-capacity policy | SHELTER_ZERO_CAP_OVER |
true |
Treat an open shelter with zero maximum and any occupant as over capacity rather than valid. |
| Resupply lead periods | SHELTER_RESUPPLY_PERIODS |
1 |
Operational periods of buffer the reorder threshold should cover before a shortage. |
| Alert sink | SHELTER_ALERT_SINK |
stdout-json |
Structured event target; point at the incident messaging bus in production, never a bare log file. |
| Canonical CRS | SHELTER_TARGET_CRS |
EPSG:4326 |
Geometry interchange CRS; shelters publish in WGS 84 for cross-agency COP rendering. |
Verification & Smoke Test
Run these assertions on a staging node before promoting a schema or threshold change. They confirm the contract rejects impossible records, the state function is deterministic on the boundary, and over-capacity and resupply alerts fire exactly when they should.
def smoke_test() -> None:
base = {
"shelter_id": "SHL-001",
"name": "Riverside High Gym",
"lon": -95.37,
"lat": 29.76,
"maximum_capacity": 120,
"current_occupancy": 108,
"ada_capacity": 12,
"reporting_agency": "County EOC",
"observed_utc": "2026-07-13T19:25:00+00:00",
"resources": [
{"resource_type": "cot", "quantity_on_hand": 120, "quantity_committed": 108,
"reorder_threshold": 20, "unit": "beds"},
{"resource_type": "potable_water", "quantity_on_hand": 400, "quantity_committed": 380,
"reorder_threshold": 50, "unit": "litres"},
],
}
# 1. Boundary determinism: 108/120 = 0.90 → exactly NEAR_CAPACITY, not OPEN.
shelter = parse_shelter(base)
assert derive_status(shelter) is ShelterStatus.NEAR_CAPACITY
# 2. Over-capacity is a strict integer excess and raises a critical alert.
over = parse_shelter({**base, "current_occupancy": 140})
assert derive_status(over) is ShelterStatus.OVER_CAPACITY
codes = {a.reason_code for a in evaluate_alerts(over)}
assert "over_capacity" in codes
# 3. Resource sufficiency: water available 400-380=20 < threshold 50 → resupply flag.
water_codes = [a for a in evaluate_alerts(shelter) if a.reason_code == "resource_low"]
assert any("potable_water" in a.detail for a in water_codes)
# 4. The contract fails closed on an impossible sub-capacity.
try:
parse_shelter({**base, "ada_capacity": 999})
raise AssertionError("expected ShelterRejected for ada_capacity > maximum")
except ShelterRejected:
pass
# 5. A closed shelter is CLOSED regardless of remaining beds.
closed = parse_shelter({**base, "is_open": False, "current_occupancy": 0})
assert derive_status(closed) is ShelterStatus.CLOSED
logger.info("shelter smoke test passed")
smoke_test()
A one-line dependency check for continuous integration confirms the stack is importable before the suite runs:
python -c "import pydantic, shapely, geopandas; print('shelter stack ok')"
python -m emergency_shelter.smoke # exits non-zero on any failed assertion
Integration With Adjacent Workflows
Shelter tracking sits downstream of location and upstream of the map. Each shelter’s point geometry arrives already normalized from the Real-Time Geocoding & Location Normalization workflow, so this stage never re-geocodes an address; it trusts a validated [lon, lat] and rejects anything outside geographic bounds at the schema gate. Every published feature emits provenance — reporting agency, observation time, and schema version — under the Emergency Metadata Standards contract, which is what makes a contested capacity number defensible after the incident. When two agencies report the same physical shelter, the deterministic status function is what lets Conflict Resolution in Multi-Agency Edits reconcile them without a human arbitrating whether the site is full. And because capacity math is pure Python with no network dependency, the same scoring runs against a locally cached shelter set under the Offline GIS Data Caching Strategies pattern when a field node loses connectivity mid-operation.
Troubleshooting
Symptom: a shelter flips between FULL and NEAR_CAPACITY on consecutive refreshes with no occupancy change. The state is being computed with a floating-point ratio and a >= comparison that rounds differently across builds. Use the integer boundary math in Step 2 — exact equality for FULL, a strict integer excess for over capacity — so a shelter at its maximum resolves to one state deterministically on every node.
Symptom: the map shows green while wheelchair users are being turned away. Only the total headcount is being scored; accessible beds are a separate constraint. Carry ada_capacity and its occupancy independently, expose accessible_beds_remaining, and render an ADA-full badge even when the governing status is OPEN, because a shelter can have general space and zero accessible space at once.
Symptom: resupply alerts fire late, after a resource is already exhausted. The rule is comparing raw quantity_on_hand to the threshold and ignoring commitments. Compute available as on-hand minus committed (Step 3); a cot count of 120 with 118 committed has an available of 2, and a threshold of 20 must trip on the available figure, not the shelf figure.
Symptom: occupancy occasionally posts as a negative number and corrupts the ratio. A raw feed is bypassing the schema contract and writing straight to the datastore. Route every write through parse_shelter; the Field(ge=0) bounds and the cross-field validators reject a negative occupancy or an over-total sub-capacity before it can be scored or rendered.
Symptom: a post-incident audit cannot tell which agency reported a disputed capacity value. The lineage fields are missing from the published feature. Ensure to_geojson_feature always stamps reporting_agency, observed_utc, and schema_version; a feature without provenance should never reach the COP, and the serializer should raise rather than publish one.
Frequently Asked Questions
Should shelter capacity be a single number or several typed capacities?
Several. A single headcount hides the constraints that actually turn people away: accessible (ADA) beds, pet spaces, isolation or medical beds, and total sleeping capacity each fill at different rates. Model maximum_capacity as the governing total but carry ada_capacity, medical_capacity, and pet_capacity as separate bounded fields, because a shelter can be full for wheelchair users while still open for ambulatory occupants, and dispatch has to see both.
How do you keep occupancy state consistent across many field nodes? Make the state a pure function of committed integer fields, not a stored flag that any node can edit. Compute status from the occupancy-to-maximum ratio with fixed thresholds and integer arithmetic, avoid floating-point comparisons on the boundary, and treat the closed operational flag as an override that always wins. Because the same inputs produce the same status everywhere, two nodes that hold the same record never disagree about whether a shelter is full.
How does this schema align with NIMS resource typing and the FEMA National Shelter System? Carry the NIMS kind and type code on every resource item so cots, potable water, and medical staff are counted in categories that mutual-aid partners already recognize, and expose the facility fields the FEMA National Shelter System expects — capacity, current population, and open or closed status — as first-class attributes. The schema is a superset: it satisfies external reporting while carrying the extra spatial and lineage fields the operational Common Operating Picture needs.
Related
- Emergency Metadata Standards — the lineage contract every published shelter feature stamps for audit.
- Real-Time Geocoding & Location Normalization — normalizes the shelter address into the point geometry this schema trusts.
- Geospatial Data Ingestion Pipelines — validates and de-duplicates the shelter features before they reach the operational map.
- Conflict Resolution in Multi-Agency Edits — reconciles concurrent reports of the same shelter using the deterministic status.
Up: Core Emergency GIS Architecture & Data Standards