scl.py

@@ -75,7 +75,7 @@ class AttribStaticMap(StaticMap, object):
         image.paste(txt, (0, 0), txt)
 
 
-def classify_event_size(num_people: int) -> dict[str, str, bool]:
+def classify_event_size(num_people: int) -> dict[str, str | bool]:
     if num_people < 250:
         return {
             "size": "Small",
@@ -96,6 +96,161 @@ def classify_event_size(num_people: int) -> dict[str, str, bool]:
         }
 
 
+def get_hashtag_string(event) -> str:
+    hashtag_string = "#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format(
+        event["identifier"]
+    )
+    return hashtag_string
+
+
+def do_initial_post(
+    event, event_class, start_time: datetime, estimated_restoration_time: datetime
+) -> dict[str, str | None]:
+    post_id = None
+    map_media_post_id = None
+    # Fallback location from the SCL API in case one couldn't be reverse geocoded
+    area_text = event["city"]
+    try:
+        map = AttribStaticMap(
+            512,
+            512,
+            url_template="https://tiles.stadiamaps.com/tiles/outdoors/{z}/{x}/{y}.png?api_key="
+            + stadiamaps_api_key,
+        )
+        assert event["polygons"]["type"] == "polygon"
+        for ring in event["polygons"]["rings"]:
+            polygon = Polygon(
+                ring,
+                # Appending 7F to the fill_color makes it 50% transparent
+                fill_color="{}7F".format(event_class["outage_color"]),
+                outline_color=event_class["outage_color"],
+                simplify=True,
+            )
+            map.add_polygon(polygon)
+        map_image = map.render()
+
+        try:
+
+            def num2deg(xtile, ytile, zoom):
+                n = 1 << zoom
+                lon_deg = xtile / n * 360.0 - 180.0
+                lat_rad = math.atan(math.sinh(math.pi * (1 - 2 * ytile / n)))
+                lat_deg = math.degrees(lat_rad)
+                return lat_deg, lon_deg
+
+            center_lat_lon = num2deg(map.x_center, map.y_center, map.zoom)
+
+            # Check to make sure the calculated lat and lon are sane enough
+            # NW Corner
+            assert center_lat_lon[0] < 48 and center_lat_lon[1] > -122.6
+            # SE Corner
+            assert center_lat_lon[0] > 47.2 and center_lat_lon[1] < -122
+
+            # Zoom level 17 ensures that we won't get any building/POI names, just street names
+            geocode_url = "{nominatim_url}/reverse?lat={lat}&lon={lon}&format=geocodejson&zoom=17".format(
+                nominatim_url=nominatim_url,
+                lat=center_lat_lon[0],
+                lon=center_lat_lon[1],
+            )
+            geocode_headers = {"User-Agent": "seattlecitylight-mastodon-bot"}
+            geocode_response = requests.get(geocode_url, headers=geocode_headers)
+            try:
+                geocode = geocode_response.json()
+            except requests.JSONDecodeError:
+                print("JSON could not be loaded from nominatim API")
+                raise
+
+            if geocode["features"][0]["properties"]["geocoding"]["city"] != "Seattle":
+                city_not_seattle_text = " of {}".format(
+                    geocode["features"][0]["properties"]["geocoding"]["city"]
+                )
+            else:
+                city_not_seattle_text = ""
+
+            street = geocode["features"][0]["properties"]["geocoding"]["name"]
+
+            if (
+                "locality" in geocode["features"][0]["properties"]["geocoding"]
+                and event_class["size"] != "Large"
+            ):
+                locality = geocode["features"][0]["properties"]["geocoding"]["locality"]
+                if locality == "Uptown":
+                    locality = "Lower Queen Anne"
+
+                alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
+                    street,
+                    locality,
+                    city_not_seattle_text,
+                )
+                area_text = "the {} area{}".format(locality, city_not_seattle_text)
+            elif "district" in geocode["features"][0]["properties"]["geocoding"]:
+                alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
+                    street,
+                    geocode["features"][0]["properties"]["geocoding"]["district"],
+                    city_not_seattle_text,
+                )
+                area_text = "the {} area{}".format(
+                    geocode["features"][0]["properties"]["geocoding"]["district"],
+                    city_not_seattle_text,
+                )
+            else:
+                alt_text = "A map showing the location of the outage, centered around {} in {}.".format(
+                    street,
+                    geocode["features"][0]["properties"]["geocoding"]["city"],
+                )
+                area_text = geocode["features"][0]["properties"]["geocoding"]["city"]
+        except Exception:
+            alt_text = "A map showing the location of the outage."
+
+        with io.BytesIO() as map_image_file:
+            map_image.save(map_image_file, format="PNG", optimize=True)
+            map_media_post = mastodon_client.media_post(
+                map_image_file.getvalue(),
+                mime_type="image/png",
+                description=alt_text,
+            )
+        map_media_post_id = map_media_post["id"]
+
+    except Exception as e:
+        print(e)
+        print(
+            "Ran into an issue with generating/uploading the map. Will post without it."
+        )
+        map_media_post = None
+    hashtag_string = get_hashtag_string(event)
+
+    post_text = """Seattle City Light is reporting a {} outage in {}.
+
+Start Date: {}
+Est. Restoration: {}
+Cause: {}
+
+{}""".format(
+        event_class["size"].lower(),
+        area_text,
+        start_time.strftime(post_datetime_format),
+        estimated_restoration_time.strftime(post_datetime_format),
+        event["cause"],
+        hashtag_string,
+    )
+
+    print(
+        "Posting the following to Mastodon, with a post length of {}:\n{}".format(
+            len(post_text), post_text
+        )
+    )
+
+    post_result = mastodon_client.status_post(
+        status=post_text,
+        media_ids=map_media_post,
+        visibility="public",
+        language="en",
+    )
+    post_id = post_result["id"]
+
+    return {"post_id": post_id, "map_media_post_id": map_media_post_id}
+
+
 class Base(DeclarativeBase):
     pass
 
@@ -142,11 +297,8 @@ with Session(engine) as session:
         else:
             status = None
 
-        hashtag_string = "#SeattleCityLightOutage #SCLOutage #SCLOutage{}".format(
-            event["identifier"]
-        )
-
         try:
+            hashtag_string = get_hashtag_string(event)
             existing_record = lookup_result.one()
             updated_properties = []
             updated_entries = []
@@ -208,9 +360,16 @@ with Session(engine) as session:
                     existing_record.most_recent_post_id = post_result["id"]
                 elif max_event_class["is_postable"]:
                     print(
-                        "This event would have been able to be posted, but someone didn't write the logic to do that for events that scaled up."
+                        "Posting an event that grew above the threshold required to post"
                     )
-
+                    initial_post_result = do_initial_post(
+                        event, event_class, start_time, estimated_restoration_time
+                    )
+                    existing_record.initial_post_id = initial_post_result["post_id"]
+                    existing_record.most_recent_post_id = initial_post_result["post_id"]
+                    existing_record.map_media_post_id = initial_post_result[
+                        "map_media_post_id"
+                    ]
             session.commit()
 
         except NoResultFound:
@@ -224,174 +383,11 @@ with Session(engine) as session:
                     )
                 )
             else:
-                # Fallback location from the SCL API in case one couldn't be reverse geocoded
+                initial_post_result = do_initial_post(
-                area_text = event["city"]
+                    event, event_class, start_time, estimated_restoration_time
-
-                try:
-                    map = AttribStaticMap(
-                        512,
-                        512,
-                        url_template="https://tiles.stadiamaps.com/tiles/outdoors/{z}/{x}/{y}.png?api_key="
-                        + stadiamaps_api_key,
                 )
-                    assert event["polygons"]["type"] == "polygon"
+                post_id = initial_post_result["post_id"]
-                    for ring in event["polygons"]["rings"]:
+                map_media_post_id = initial_post_result["map_media_post_id"]
-                        polygon = Polygon(
-                            ring,
-                            # Appending 7F to the fill_color makes it 50% transparent
-                            fill_color="{}7F".format(event_class["outage_color"]),
-                            outline_color=event_class["outage_color"],
-                            simplify=True,
-                        )
-                        map.add_polygon(polygon)
-                    map_image = map.render()
-
-                    try:
-
-                        def num2deg(xtile, ytile, zoom):
-                            n = 1 << zoom
-                            lon_deg = xtile / n * 360.0 - 180.0
-                            lat_rad = math.atan(
-                                math.sinh(math.pi * (1 - 2 * ytile / n))
-                            )
-                            lat_deg = math.degrees(lat_rad)
-                            return lat_deg, lon_deg
-
-                        center_lat_lon = num2deg(map.x_center, map.y_center, map.zoom)
-
-                        # Check to make sure the calculated lat and lon are sane enough
-                        # NW Corner
-                        assert center_lat_lon[0] < 48 and center_lat_lon[1] > -122.6
-                        # SE Corner
-                        assert center_lat_lon[0] > 47.2 and center_lat_lon[1] < -122
-
-                        # Zoom level 17 ensures that we won't get any building/POI names, just street names
-                        geocode_url = "{nominatim_url}/reverse?lat={lat}&lon={lon}&format=geocodejson&zoom=17".format(
-                            nominatim_url=nominatim_url,
-                            lat=center_lat_lon[0],
-                            lon=center_lat_lon[1],
-                        )
-                        geocode_headers = {
-                            "User-Agent": "seattlecitylight-mastodon-bot"
-                        }
-                        geocode_response = requests.get(
-                            geocode_url, headers=geocode_headers
-                        )
-                        try:
-                            geocode = geocode_response.json()
-                        except requests.JSONDecodeError:
-                            print("JSON could not be loaded from nominatim API")
-                            raise
-
-                        if (
-                            geocode["features"][0]["properties"]["geocoding"]["city"]
-                            != "Seattle"
-                        ):
-                            city_not_seattle_text = " of {}".format(
-                                geocode["features"][0]["properties"]["geocoding"][
-                                    "city"
-                                ]
-                            )
-                        else:
-                            city_not_seattle_text = ""
-
-                        street = geocode["features"][0]["properties"]["geocoding"][
-                            "name"
-                        ]
-
-                        if (
-                            "locality"
-                            in geocode["features"][0]["properties"]["geocoding"]
-                            and event_class["size"] != "Large"
-                        ):
-                            locality = geocode["features"][0]["properties"][
-                                "geocoding"
-                            ]["locality"]
-                            if locality == "Uptown":
-                                locality = "Lower Queen Anne"
-
-                            alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
-                                street,
-                                locality,
-                                city_not_seattle_text,
-                            )
-                            area_text = "the {} area{}".format(
-                                locality, city_not_seattle_text
-                            )
-                        elif (
-                            "district"
-                            in geocode["features"][0]["properties"]["geocoding"]
-                        ):
-                            alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
-                                street,
-                                geocode["features"][0]["properties"]["geocoding"][
-                                    "district"
-                                ],
-                                city_not_seattle_text,
-                            )
-                            area_text = "the {} area{}".format(
-                                geocode["features"][0]["properties"]["geocoding"][
-                                    "district"
-                                ],
-                                city_not_seattle_text,
-                            )
-                        else:
-                            alt_text = "A map showing the location of the outage, centered around {} in {}.".format(
-                                street,
-                                geocode["features"][0]["properties"]["geocoding"][
-                                    "city"
-                                ],
-                            )
-                            area_text = geocode["features"][0]["properties"][
-                                "geocoding"
-                            ]["city"]
-                    except Exception:
-                        alt_text = "A map showing the location of the outage."
-
-                    with io.BytesIO() as map_image_file:
-                        map_image.save(map_image_file, format="PNG", optimize=True)
-                        map_media_post = mastodon_client.media_post(
-                            map_image_file.getvalue(),
-                            mime_type="image/png",
-                            description=alt_text,
-                        )
-                    map_media_post_id = map_media_post["id"]
-
-                except Exception as e:
-                    print(e)
-                    print(
-                        "Ran into an issue with generating/uploading the map. Will post without it."
-                    )
-                    map_media_post = None
-
-                post_text = """Seattle City Light is reporting a {} outage in {}.
-
-    Start Date: {}
-    Est. Restoration: {}
-    Cause: {}
-
-    {}""".format(
-                    event_class["size"].lower(),
-                    area_text,
-                    start_time.strftime(post_datetime_format),
-                    estimated_restoration_time.strftime(post_datetime_format),
-                    event["cause"],
-                    hashtag_string,
-                )
-
-                print(
-                    "Posting the following to Mastodon, with a post length of {}:\n{}".format(
-                        len(post_text), post_text
-                    )
-                )
-
-                post_result = mastodon_client.status_post(
-                    status=post_text,
-                    media_ids=map_media_post,
-                    visibility="public",
-                    language="en",
-                )
-                post_id = post_result["id"]
 
             new_outage_record = SclOutage(
                 scl_outage_id=event["id"],