Skip to content

API documentation

Main object for getting the PECO outage counter data.

BadJSONError

Bases: Exception

Raised when the JSON is invalid.

Source code in src/peco/__init__.py 
258
259
260
261
262
class BadJSONError(Exception):
    """Raised when the JSON is invalid."""

    def __init__(self, message: str = "Bad JSON returned from PECO") -> None:
        super().__init__(message)

HttpError

Bases: Exception

Raised when an error during HTTP request occurs.

Source code in src/peco/__init__.py 
251
252
253
254
255
class HttpError(Exception):
    """Raised when an error during HTTP request occurs."""

    def __init__(self) -> None:
        super().__init__("Bad response from PECO")

IncompatibleMeterError

Bases: MeterError

Raised when the meter is not compatible with the API.

Source code in src/peco/__init__.py 
269
270
271
272
273
class IncompatibleMeterError(MeterError):
    """Raised when the meter is not compatible with the API."""

    def __init__(self) -> None:
        super().__init__("Meter is not compatible with the API")

InvalidCountyError

Bases: ValueError

Raised when the county is invalid.

Source code in src/peco/__init__.py 
244
245
246
247
248
class InvalidCountyError(ValueError):
    """Raised when the county is invalid."""

    def __init__(self, county: str) -> None:
        super().__init__(f"{county} is not a valid county")

MeterError

Bases: Exception

Generic meter error.

Source code in src/peco/__init__.py 
265
266
class MeterError(Exception):
    """Generic meter error."""

PecoOutageApi

Main object for getting the PECO outage counter data.

Source code in src/peco/__init__.py 
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
class PecoOutageApi:
    """Main object for getting the PECO outage counter data."""

    def __init__(self) -> None:
        """Initialize the PECO outage counter object."""

    @staticmethod
    async def get_request(
        url: str,
        websession: aiohttp.ClientSession | None = None,
    ) -> dict[str, Any]:
        """Make a GET request to the API."""
        data: dict[str, Any]
        if websession is not None:
            async with websession.get(url) as r:
                data = await r.json()
        else:
            async with aiohttp.ClientSession() as session, session.get(url) as r:
                data = await r.json()

        if r.status != STATUS_OK:
            raise HttpError

        return data

    @staticmethod
    async def post_request(
        url: str,
        data: dict[str, Any],
        websession: aiohttp.ClientSession | None = None,
    ) -> dict[str, Any]:
        """Make a POST request to the API."""
        if websession is not None:
            async with websession.post(url, json=data) as r:
                data = await r.json(content_type="text/html")
        else:
            async with aiohttp.ClientSession() as session, session.post(
                url,
                json=data,
            ) as r:
                data = await r.json(content_type="text/html")

        if r.status != STATUS_OK:
            raise HttpError

        return data

    async def get_outage_count(
        self: PecoOutageApi,
        county: str,
        websession: aiohttp.ClientSession | None = None,
    ) -> OutageResults:
        """Get the outage count for the given county."""

        if county not in COUNTY_LIST:
            raise InvalidCountyError(county)

        data = await self.get_request(API_URL, websession)

        try:
            id_that_has_the_report: str = data["data"]["interval_generation_data"]
        except KeyError as err:
            raise BadJSONError from err

        report_url = REPORT_URL.format(id_that_has_the_report)
        data = await self.get_request(report_url, websession)

        try:
            areas: list[dict[str, Any]] = data["file_data"]["areas"]
        except KeyError as err:
            raise BadJSONError from err

        outage_result: OutageResults = OutageResults(
            customers_out=0,
            percent_customers_out=0,
            outage_count=0,
            customers_served=0,
        )
        for area in areas:
            if area["name"] == county:
                customers_out = area["cust_a"]["val"]
                percent_customers_out = area["percent_cust_a"]["val"]
                outage_count = area["n_out"]
                customers_served = area["cust_s"]
                outage_result = OutageResults(
                    customers_out=customers_out,
                    percent_customers_out=percent_customers_out,
                    outage_count=outage_count,
                    customers_served=customers_served,
                )
        return outage_result

    async def get_outage_totals(
        self: PecoOutageApi,
        websession: aiohttp.ClientSession | None = None,
    ) -> OutageResults:
        """Get the outage totals for the given county and mode."""
        data = await self.get_request(API_URL, websession)

        try:
            id_that_has_the_report: str = data["data"]["interval_generation_data"]
        except KeyError as err:
            raise BadJSONError from err

        report_url = REPORT_URL.format(id_that_has_the_report)
        data = await self.get_request(report_url, websession)

        try:
            totals = data["file_data"]["totals"]
        except KeyError as err:
            raise BadJSONError from err

        return OutageResults(
            customers_out=totals["cust_a"]["val"],
            percent_customers_out=totals["percent_cust_a"]["val"],
            outage_count=totals["n_out"],
            customers_served=totals["cust_s"],
        )

    async def meter_check(
        self: PecoOutageApi,
        phone_number: str,
        websession: aiohttp.ClientSession | None = None,
    ) -> bool:
        """Check if power is being delivered to the house."""
        if len(phone_number) != PHONE_NUMBER_LENGTH:
            msg = "Phone number must be 10 digits"
            raise ValueError(msg)

        if not phone_number.isdigit():
            msg = "Phone number must be numeric"
            raise ValueError(msg)

        data1 = await self.post_request(QUERY_URL, {"phone": phone_number}, websession)

        if not data1["success"]:
            raise HttpError

        if not data1["data"][0]["smartMeterStatus"]:
            raise IncompatibleMeterError

        auid = data1["data"][0]["auid"]
        acc_number = data1["data"][0]["accountNumber"]

        data2 = await self.post_request(
            PRECHECK_URL,
            {
                "auid": auid,
                "accountNumber": acc_number,
                "phone": phone_number,
            },
            websession,
        )

        if not data2["success"]:
            raise HttpError

        if not data2["data"]["meterPing"]:
            raise UnresponsiveMeterError

        data3 = await self.post_request(
            PING_URL,
            {"auid": auid, "accountNumber": acc_number},
            websession,
        )

        if not data3["success"]:
            raise HttpError

        return bool(data3["data"]["meterInfo"]["pingResult"])

    async def get_map_alerts(
        self: PecoOutageApi,
        websession: aiohttp.ClientSession | None = None,
    ) -> AlertResults:
        """Get the alerts that show on the outage map."""
        data = await self.get_request(API_URL, websession)

        try:
            alert_deployment_id: str = data["controlCenter"]["alertDeploymentId"]
        except KeyError as err:
            raise BadJSONError from err

        if alert_deployment_id is None:
            # No alert
            return AlertResults(alert_content="", alert_title="")

        alerts_url = ALERTS_URL.format(alert_deployment_id)
        data1 = await self.get_request(alerts_url, websession)

        # There is always only one alert.
        # Again, if anyone sees more than one alert, please open an issue.
        try:
            alert = data1["_embedded"]["deployedAlertResourceList"][0]["data"][0]
        except KeyError:
            return AlertResults(
                alert_content="",
                alert_title="",
            )

        parsed_content = TAG_RE.sub("", alert["content"].replace("<br />", "\n\n"))

        return AlertResults(
            alert_content=parsed_content,
            alert_title=alert["bannerTitle"],
        )

__init__() -> None

Initialize the PECO outage counter object.

Source code in src/peco/__init__.py 
27
28
def __init__(self) -> None:
    """Initialize the PECO outage counter object."""

get_map_alerts(websession: aiohttp.ClientSession | None = None) -> AlertResults async

Get the alerts that show on the outage map.

Source code in src/peco/__init__.py 
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
async def get_map_alerts(
    self: PecoOutageApi,
    websession: aiohttp.ClientSession | None = None,
) -> AlertResults:
    """Get the alerts that show on the outage map."""
    data = await self.get_request(API_URL, websession)

    try:
        alert_deployment_id: str = data["controlCenter"]["alertDeploymentId"]
    except KeyError as err:
        raise BadJSONError from err

    if alert_deployment_id is None:
        # No alert
        return AlertResults(alert_content="", alert_title="")

    alerts_url = ALERTS_URL.format(alert_deployment_id)
    data1 = await self.get_request(alerts_url, websession)

    # There is always only one alert.
    # Again, if anyone sees more than one alert, please open an issue.
    try:
        alert = data1["_embedded"]["deployedAlertResourceList"][0]["data"][0]
    except KeyError:
        return AlertResults(
            alert_content="",
            alert_title="",
        )

    parsed_content = TAG_RE.sub("", alert["content"].replace("<br />", "\n\n"))

    return AlertResults(
        alert_content=parsed_content,
        alert_title=alert["bannerTitle"],
    )

get_outage_count(county: str, websession: aiohttp.ClientSession | None = None) -> OutageResults async

Get the outage count for the given county.

Source code in src/peco/__init__.py 
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
async def get_outage_count(
    self: PecoOutageApi,
    county: str,
    websession: aiohttp.ClientSession | None = None,
) -> OutageResults:
    """Get the outage count for the given county."""

    if county not in COUNTY_LIST:
        raise InvalidCountyError(county)

    data = await self.get_request(API_URL, websession)

    try:
        id_that_has_the_report: str = data["data"]["interval_generation_data"]
    except KeyError as err:
        raise BadJSONError from err

    report_url = REPORT_URL.format(id_that_has_the_report)
    data = await self.get_request(report_url, websession)

    try:
        areas: list[dict[str, Any]] = data["file_data"]["areas"]
    except KeyError as err:
        raise BadJSONError from err

    outage_result: OutageResults = OutageResults(
        customers_out=0,
        percent_customers_out=0,
        outage_count=0,
        customers_served=0,
    )
    for area in areas:
        if area["name"] == county:
            customers_out = area["cust_a"]["val"]
            percent_customers_out = area["percent_cust_a"]["val"]
            outage_count = area["n_out"]
            customers_served = area["cust_s"]
            outage_result = OutageResults(
                customers_out=customers_out,
                percent_customers_out=percent_customers_out,
                outage_count=outage_count,
                customers_served=customers_served,
            )
    return outage_result

get_outage_totals(websession: aiohttp.ClientSession | None = None) -> OutageResults async

Get the outage totals for the given county and mode.

Source code in src/peco/__init__.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
async def get_outage_totals(
    self: PecoOutageApi,
    websession: aiohttp.ClientSession | None = None,
) -> OutageResults:
    """Get the outage totals for the given county and mode."""
    data = await self.get_request(API_URL, websession)

    try:
        id_that_has_the_report: str = data["data"]["interval_generation_data"]
    except KeyError as err:
        raise BadJSONError from err

    report_url = REPORT_URL.format(id_that_has_the_report)
    data = await self.get_request(report_url, websession)

    try:
        totals = data["file_data"]["totals"]
    except KeyError as err:
        raise BadJSONError from err

    return OutageResults(
        customers_out=totals["cust_a"]["val"],
        percent_customers_out=totals["percent_cust_a"]["val"],
        outage_count=totals["n_out"],
        customers_served=totals["cust_s"],
    )

get_request(url: str, websession: aiohttp.ClientSession | None = None) -> dict[str, Any] async staticmethod

Make a GET request to the API.

Source code in src/peco/__init__.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@staticmethod
async def get_request(
    url: str,
    websession: aiohttp.ClientSession | None = None,
) -> dict[str, Any]:
    """Make a GET request to the API."""
    data: dict[str, Any]
    if websession is not None:
        async with websession.get(url) as r:
            data = await r.json()
    else:
        async with aiohttp.ClientSession() as session, session.get(url) as r:
            data = await r.json()

    if r.status != STATUS_OK:
        raise HttpError

    return data

meter_check(phone_number: str, websession: aiohttp.ClientSession | None = None) -> bool async

Check if power is being delivered to the house.

Source code in src/peco/__init__.py 
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
async def meter_check(
    self: PecoOutageApi,
    phone_number: str,
    websession: aiohttp.ClientSession | None = None,
) -> bool:
    """Check if power is being delivered to the house."""
    if len(phone_number) != PHONE_NUMBER_LENGTH:
        msg = "Phone number must be 10 digits"
        raise ValueError(msg)

    if not phone_number.isdigit():
        msg = "Phone number must be numeric"
        raise ValueError(msg)

    data1 = await self.post_request(QUERY_URL, {"phone": phone_number}, websession)

    if not data1["success"]:
        raise HttpError

    if not data1["data"][0]["smartMeterStatus"]:
        raise IncompatibleMeterError

    auid = data1["data"][0]["auid"]
    acc_number = data1["data"][0]["accountNumber"]

    data2 = await self.post_request(
        PRECHECK_URL,
        {
            "auid": auid,
            "accountNumber": acc_number,
            "phone": phone_number,
        },
        websession,
    )

    if not data2["success"]:
        raise HttpError

    if not data2["data"]["meterPing"]:
        raise UnresponsiveMeterError

    data3 = await self.post_request(
        PING_URL,
        {"auid": auid, "accountNumber": acc_number},
        websession,
    )

    if not data3["success"]:
        raise HttpError

    return bool(data3["data"]["meterInfo"]["pingResult"])

post_request(url: str, data: dict[str, Any], websession: aiohttp.ClientSession | None = None) -> dict[str, Any] async staticmethod

Make a POST request to the API.

Source code in src/peco/__init__.py 
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
@staticmethod
async def post_request(
    url: str,
    data: dict[str, Any],
    websession: aiohttp.ClientSession | None = None,
) -> dict[str, Any]:
    """Make a POST request to the API."""
    if websession is not None:
        async with websession.post(url, json=data) as r:
            data = await r.json(content_type="text/html")
    else:
        async with aiohttp.ClientSession() as session, session.post(
            url,
            json=data,
        ) as r:
            data = await r.json(content_type="text/html")

    if r.status != STATUS_OK:
        raise HttpError

    return data

UnresponsiveMeterError

Bases: MeterError

Raised when the meter is not responding.

Source code in src/peco/__init__.py 
276
277
278
279
280
class UnresponsiveMeterError(MeterError):
    """Raised when the meter is not responding."""

    def __init__(self, message: str = "Meter is not responding") -> None:
        super().__init__(message)