energa-outages-api 0.2.0

Check Energa Operator power outages by GPS coordinates or address

energa-outages-api

Python library for checking Energa Operator planned and emergency power outages. Supports lookup by GPS coordinates or postal address.

pip install energa-outages-api

from energa_outages_api import EnergaOutagesClient

---

Features

• Live data from Energa Operator API (northern/central Poland)
• Check outages by GPS coordinates with configurable margin radius and confidence scoring
• Check currently active outages — or planned outages up to N days ahead
• Check outages by address (city, street, building number) with prefix normalization
• Async-first (aiohttp), with a sync wrapper for scripts
• Built-in TTL cache (default 5 min) — safe to call on every HA update cycle
• Zero dependencies beyond aiohttp
• Fully typed, PEP 561 compliant

---

Installation

pip install energa-outages-api

From GitHub:

pip install git+https://github.com/vincentto13/energa-outages-api.git

Requires Python 3.11+.

---

Quick start

Check by GPS coordinates

import asyncio
from energa_outages_api import EnergaOutagesClient

async def main():
    async with EnergaOutagesClient() as client:
        match = await client.check_location(lat=54.352, lon=18.647, margin_m=200)
        if match:
            print(f"Outage detected! Confidence: {match.confidence:.0%}")
            print(f"Region: {match.shutdown.region_name}")
            print(f"Window: {match.shutdown.start_date} → {match.shutdown.end_date}")
        else:
            print("No outage at your location.")

asyncio.run(main())

Check by address

async def main():
    async with EnergaOutagesClient() as client:
        matches = await client.check_address(
            city="Włocławek",
            street="ul. Grodzka",   # "ul." prefix stripped automatically
            number="169",
        )
        for m in matches:
            print(f"{m.matched_address}  [{m.confidence:.0%}]")
            print(f"  {m.shutdown.start_date} → {m.shutdown.end_date}")

Sync usage

from energa_outages_api import EnergaOutagesClientSync

with EnergaOutagesClientSync() as client:
    match = client.check_location(lat=54.352, lon=18.647)
    matches = client.check_address(city="Gdańsk", street="Dworska", number="11")

---

API reference

EnergaOutagesClient / EnergaOutagesClientSync

All async methods have identical sync equivalents on EnergaOutagesClientSync.

client = EnergaOutagesClient(
    session=None,            # optional aiohttp.ClientSession to reuse
    cache_ttl_seconds=300,   # how long to cache the shutdown list (default 5 min)
    timeout_seconds=10,      # HTTP request timeout
)

---

get_shutdowns(force=False) → list[Shutdown]

Returns all currently active shutdowns from the Energa API. Uses the TTL cache unless force=True.

shutdowns = await client.get_shutdowns()
for s in shutdowns:
    print(s.region_name, s.start_date, s.message)

---

check_location(lat, lon, margin_m=200, now=None, force=False) → Optional[OutageMatch]

Returns the highest-confidence match for the given GPS coordinates, or None if no active outage is nearby.

Argument	Type	Description
lat	float	Latitude (WGS84, decimal degrees)
lon	float	Longitude (WGS84, decimal degrees)
margin_m	float	Distance in metres beyond polygon boundary still considered a match (default 200)
now	datetime	Override current time — useful for testing
force	bool	Bypass cache

---

check_location_all(lat, lon, margin_m=200, ...) → list[OutageMatch]

Same as check_location but returns all matches sorted by confidence descending. Useful when a location sits on the boundary between two outage polygons.

---

check_location_planned(lat, lon, margin_m=200, window_hours=168, force=False) → list[OutageMatch]

Returns outages that will affect the location within the next window_hours hours but are not currently active. Results are sorted by shutdown start time ascending (soonest first).

Argument	Type	Description
lat / lon	float	GPS coordinates (WGS84)
margin_m	float	Distance margin in metres beyond polygon boundary
window_hours	float	How far ahead to look (default 168 = 7 days)
force	bool	Bypass cache

async def main():
    async with EnergaOutagesClient() as client:
        # Currently active
        active = await client.check_location(lat=54.352, lon=18.647, margin_m=500)

        # Planned in the next 7 days
        planned = await client.check_location_planned(lat=54.352, lon=18.647, margin_m=500)
        for m in planned:
            next_start = m.shutdown.next_start(datetime.now(timezone.utc))
            print(f"Planned outage starts: {next_start}  ({m.shutdown.message[:80]})")

---

check_address(city, street=None, number=None, force=False) → list[AddressMatch]

Returns all shutdowns whose parsed message matches the given address, sorted by confidence descending.

Argument	Type	Description
city	str	City or village name (required)
street	str | None	Street name. Prefixes like ul., al., os., gen. are stripped automatically
number	str | None	Building number. Matched exactly or against od X do Y ranges

Prefix stripping means all of these are equivalent:

await client.check_address(city="Gdańsk", street="Dworska")
await client.check_address(city="Gdańsk", street="ul. Dworska")
await client.check_address(city="Gdańsk", street="ulica Dworska")

---

Return types

OutageMatch

Returned by check_location / check_location_all.

Attribute	Type	Description
confidence	float	0.0–1.0. 1.0 = inside polygon, linear decay to 0.0 at margin_m
is_inside	bool	True if the point is inside the outage polygon
distance_to_boundary_m	float	Metres to the nearest polygon edge. 0.0 if inside
shutdown	Shutdown	Full shutdown details
guid	str	Shortcut to shutdown.guid
to_dict()	dict	Flat dict, suitable for Home Assistant extra_state_attributes

AddressMatch

Returned by check_address.

Attribute	Type	Description
confidence	float	1.0 = all fields matched; 0.7 = street matched but number not confirmed
matched_address	ParsedAddress	The specific address entry that matched
shutdown	Shutdown	Full shutdown details
to_dict()	dict	Flat dict

Shutdown

Attribute	Type	Description
guid	str	Unique shutdown ID
region_name	str	Region, e.g. "Gdańsk"
dept_name	str	Department, e.g. "Gdańsk"
areas	list[str]	Administrative area names
message	str	Free-text affected addresses
start_date	datetime	Outage start (UTC)
end_date	datetime	Outage end (UTC)
hours	list[HourWindow]	Sub-windows within the day
shutdown_type	int	1 = planned, 2 = emergency
shutdown_type_label	str	"planned" or "emergency"
ring	list[[lon, lat]]	GeoJSON polygon boundary
centroid_lon/lat	float	Polygon centre point
is_active_at(now)	bool	True if now falls inside any active hours window
is_active_within(from_dt, to_dt)	bool	True if any hours window overlaps the given interval
next_start(after)	datetime | None	Earliest hours window start that begins after after
parsed_addresses()	list[ParsedAddress]	Parses message into structured addresses

ParsedAddress

Structured result of parsing a shutdown message.

Attribute	Type	Description
place	str	City or village name
street	str | None	Street name, None for rural addresses
numbers	list[str]	Building numbers, e.g. ["11", "13/A", "od 20 do 22"]
str(addr)	str	Human-readable, e.g. "Gdańsk ul. Dworska 11, 13/A"

---

Confidence scoring

GPS-based (check_location)

Point inside polygon          → confidence = 1.0
Point outside, within margin  → confidence = 1.0 − (distance / margin_m)
Point beyond margin           → no match returned

Address-based (check_address)

City + street + number all match (exact or within od/do range)  → 1.0
City + street match, number provided but not found in list       → 0.7
City + street match, no number asked                             → 1.0
City only match                                                   → 1.0

A confidence of 0.7 means the street segment is definitely affected, but your specific building number was not listed — it may still be impacted.

---

Home Assistant integration

EnergaOutagesClient is designed to be used in a Home Assistant custom integration:

• Async-first — drop into a DataUpdateCoordinator with no threading concerns
• Shared session — pass HA's aiohttp_client.async_get_clientsession(hass) to avoid creating extra connections
• to_dict() — both OutageMatch and AddressMatch return a flat dict ready for extra_state_attributes
• force=True — for manual refresh triggered by a service call

from homeassistant.helpers.aiohttp_client import async_get_clientsession
from energa_outages_api import EnergaOutagesClient

session = async_get_clientsession(hass)
client = EnergaOutagesClient(session=session, cache_ttl_seconds=300)

# In coordinator _async_update_data:
match = await client.check_location(lat=hass.config.latitude, lon=hass.config.longitude)

---

Exceptions

Exception	When raised
EnergaOutagesError	Base class for all library exceptions
EnergaOutagesAPIError	HTTP or network failure fetching outage data
EnergaOutagesParseError	Unexpected structure in the API response

from energa_outages_api import EnergaOutagesAPIError, EnergaOutagesParseError

try:
    match = await client.check_location(lat=54.352, lon=18.647)
except EnergaOutagesAPIError as e:
    print(f"Could not reach Energa API: {e}")
except EnergaOutagesParseError as e:
    print(f"Unexpected API response: {e}")

---

Coverage

Energa Operator serves northern and central Poland. Regions covered include Gdańsk, Gdynia, Sopot, Toruń, Bydgoszcz, Olsztyn, Płock, Włocławek, Kalisz, Konin, and surrounding areas. Wrocław and southern Poland are served by Tauron, not Energa — queries for those areas will return no results.

---

License

MIT