config-sample.yml

@@ -1,12 +1,10 @@
 osm:
   url_template: https://tile.openstreetmap.org/{z}/{x}/{y}.png
   attribution: © OpenStreetMap
-  health_check_url:
 nominatim:
-  api_base_url: https://nominatim.openstreetmap.org
+  api_base_url: 
-  health_check_url: https://nominatim.openstreetmap.org/status
 mastodon:
-  client_id:
+  client_id: 
-  client_secret:
+  client_secret: 
-  access_token:
+  access_token: 
-  api_base_url:
+  api_base_url: 

scl.py

@@ -1,7 +1,7 @@
 import io
 import re
 from datetime import datetime
-from typing import Optional, Tuple
+from typing import Optional
 
 import mastodon
 import requests
@@ -9,7 +9,6 @@ import shapely
 import staticmap
 import yaml
 from mastodon import Mastodon
-from mastodon.return_types import MediaAttachment
 from PIL import Image, ImageDraw, ImageFont
 from shapely import Geometry
 from sqlalchemy import create_engine, select
@@ -19,7 +18,7 @@ from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
 from geospatial import DBGeometry, convert_outage_geometry
 
 REQUESTS_HEADERS = {"User-Agent": "seattlecitylight-mastodon-bot"}
-POST_DATETIME_FORMAT = "%b %e %l:%M %p"
+post_datetime_format = "%b %e %l:%M %p"
 
 scl_events_url = "https://utilisocial.io/datacapable/v2/p/scl/map/events"
 scl_events_response = requests.get(scl_events_url)
@@ -84,6 +83,7 @@ class AttribStaticMap(staticmap.StaticMap, object):
         textSize = fnt.getbbox(self.attribution)
         textPosition = (image.size[0] - textSize[2], image.size[1] - textSize[3])
         offset = 2
+        options = {"fill": (255, 255, 255, 180)}
         d.rectangle(
             [
                 (textPosition[0] - (2 * offset), textPosition[1] - (2 * offset)),
@@ -92,7 +92,7 @@ class AttribStaticMap(staticmap.StaticMap, object):
                     textSize[3] + textPosition[1] + (2 * offset),
                 ),
             ],
-            fill=(255, 255, 255, 180),
+            **options,
         )
 
         # draw text, full opacity
@@ -152,108 +152,6 @@ def get_hashtag_string(event) -> str:
     return hashtag_string
 
 
-def generate_post_map_image(
-    event, event_class, outage_geometries: shapely.MultiPolygon
-) -> Tuple[MediaAttachment, str]:
-    # Fallback location from the SCL API in case one couldn't be reverse geocoded
-    area_text = event["city"]
-    map = AttribStaticMap(
-        1024,
-        1024,
-        url_template=osm_url_template,
-        tile_size=512,
-    )
-    assert event["polygons"]["type"] == "polygon"
-    for ring in event["polygons"]["rings"]:
-        polygon = staticmap.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)
-
-    try:
-        outage_center: shapely.Point = outage_geometries.centroid
-
-        assert outage_center.geom_type == "Point"
-        # Check to make sure the calculated lat and lon are sane enough
-        # NW Corner
-        assert outage_center.y < 48 and outage_center.x > -122.6
-        # SE Corner
-        assert outage_center.y > 47.2 and outage_center.x < -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=outage_center.y,
-            lon=outage_center.x,
-        )
-        geocode_response = requests.get(geocode_url, headers=REQUESTS_HEADERS)
-        try:
-            geocode = geocode_response.json()
-        except requests.JSONDecodeError:
-            print("JSON could not be loaded from nominatim API")
-            raise
-
-        city = geocode["features"][0]["properties"]["geocoding"]["city"]
-        street = geocode["features"][0]["properties"]["geocoding"]["name"]
-        event["geoloc_city"] = city
-
-        if city != "Seattle":
-            city_not_seattle_text = " of {}".format(city)
-        else:
-            city_not_seattle_text = ""
-
-        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)
-            event["neighborhood"] = locality
-        elif "district" in geocode["features"][0]["properties"]["geocoding"]:
-            district = geocode["features"][0]["properties"]["geocoding"]["district"]
-            alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
-                street,
-                district,
-                city_not_seattle_text,
-            )
-            area_text = "the {} area{}".format(
-                district,
-                city_not_seattle_text,
-            )
-            event["neighborhood"] = district
-        else:
-            alt_text = "A map showing the location of the outage, centered around {} in {}.".format(
-                street,
-                city,
-            )
-            area_text = city
-    except Exception:
-        alt_text = "A map showing the location of the outage."
-
-    map_image = map.render()
-
-    with io.BytesIO() as map_image_file:
-        map_image.save(map_image_file, format="WebP", method=6)
-        map_media_post = mastodon_client.media_post(
-            map_image_file,
-            mime_type="image/webp",
-            description=alt_text,
-        )
-    return (map_media_post, area_text)
-
-
 def do_initial_post(
     event,
     event_class,
@@ -262,25 +160,118 @@ def do_initial_post(
     outage_geometries: shapely.MultiPolygon,
 ) -> dict[str, str | None]:
     post_id = None
-    map_media_post = None
+    map_media_post_id = None
-    area_text = str()
+    # Fallback location from the SCL API in case one couldn't be reverse geocoded
-
+    area_text = event["city"]
     try:
-        map_media_post, area_text = generate_post_map_image(
+        map = AttribStaticMap(
-            event, event_class, outage_geometries
+            1024,
+            1024,
+            url_template=osm_url_template,
+            tile_size=512,
         )
+        assert event["polygons"]["type"] == "polygon"
+        for ring in event["polygons"]["rings"]:
+            polygon = staticmap.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)
+
+        try:
+            outage_center: shapely.Point = outage_geometries.centroid
+
+            assert outage_center.geom_type == "Point"
+            # Check to make sure the calculated lat and lon are sane enough
+            # NW Corner
+            assert outage_center.y < 48 and outage_center.x > -122.6
+            # SE Corner
+            assert outage_center.y > 47.2 and outage_center.x < -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=outage_center.y,
+                lon=outage_center.x,
+            )
+            geocode_response = requests.get(geocode_url, headers=REQUESTS_HEADERS)
+            try:
+                geocode = geocode_response.json()
+            except requests.JSONDecodeError:
+                print("JSON could not be loaded from nominatim API")
+                raise
+
+            city = geocode["features"][0]["properties"]["geocoding"]["city"]
+            street = geocode["features"][0]["properties"]["geocoding"]["name"]
+            event["geoloc_city"] = city
+
+            if city != "Seattle":
+                city_not_seattle_text = " of {}".format(city)
+            else:
+                city_not_seattle_text = ""
+
+            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)
+                event["neighborhood"] = locality
+            elif "district" in geocode["features"][0]["properties"]["geocoding"]:
+                district = geocode["features"][0]["properties"]["geocoding"]["district"]
+                alt_text = "A map showing the location of the outage, centered around {} in the {} area{}.".format(
+                    street,
+                    district,
+                    city_not_seattle_text,
+                )
+                area_text = "the {} area{}".format(
+                    district,
+                    city_not_seattle_text,
+                )
+                event["neighborhood"] = district
+            else:
+                alt_text = "A map showing the location of the outage, centered around {} in {}.".format(
+                    street,
+                    city,
+                )
+                area_text = city
+        except Exception:
+            alt_text = "A map showing the location of the outage."
+
+        map_image = map.render()
+
+        with io.BytesIO() as map_image_file:
+            map_image.save(map_image_file, format="WebP", method=6)
+            map_media_post = mastodon_client.media_post(
+                map_image_file.getvalue(),
+                mime_type="image/webp",
+                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)
 
     est_restoration_post_text = str()
     if estimated_restoration_time > datetime.now():
         est_restoration_post_text = "\nEst. Restoration: {}\n".format(
-            estimated_restoration_time.strftime(POST_DATETIME_FORMAT)
+            estimated_restoration_time.strftime(post_datetime_format)
         )
 
     post_text = """Seattle City Light is reporting a {} outage in {}.
@@ -291,7 +282,7 @@ Cause: {}
 {}""".format(
         event_class["size"].lower(),
         area_text,
-        start_time.strftime(POST_DATETIME_FORMAT),
+        start_time.strftime(post_datetime_format),
         est_restoration_post_text,
         event["cause"],
         hashtag_string,
@@ -373,7 +364,6 @@ with Session(engine) as session:
             existing_record = lookup_result.one()
             updated_properties = []
             updated_entries = []
-            map_media_post = None
 
             est_restoration_diff_mins = (
                 abs(
@@ -392,7 +382,7 @@ with Session(engine) as session:
                     updated_properties.append("estimated restoration")
                     updated_entries.append(
                         "Est. Restoration: {}".format(
-                            estimated_restoration_time.strftime(POST_DATETIME_FORMAT)
+                            estimated_restoration_time.strftime(post_datetime_format)
                         )
                     )
             if event["cause"] != existing_record.cause:
@@ -416,12 +406,8 @@ with Session(engine) as session:
             max_event_class = classify_event_size(existing_record.max_num_people)
             if existing_record.outage_geometries != outage_geometries:
                 print("Geometries modified")
-                updated_properties.append("area")
                 existing_record.outage_geometries = outage_geometries
                 existing_record.geometries_modified = True
-                map_media_post, _ = generate_post_map_image(
-                    event, event_class, outage_geometries
-                )
 
             if updated_properties:
                 updated_properties.sort()
@@ -446,7 +432,6 @@ with Session(engine) as session:
                         post_result = mastodon_client.status_post(
                             status="\n".join(updated_entries),
                             in_reply_to_id=existing_record.most_recent_post_id,
-                            media_ids=map_media_post,
                             visibility="public",
                             language="en",
                         )