feeds: add asian hornet sightings in Bern

resspublica/bernAsianHornet.py

@@ -0,0 +1,189 @@
+import pandas as pd
+import copy
+import geopandas as gpd
+from shapely import wkb
+import matplotlib.pyplot as plt
+import datetime
+from datetime import date, time
+from bs4 import BeautifulSoup
+from tinydb import TinyDB, Query
+from time import sleep
+from zoneinfo import ZoneInfo
+from pathlib import Path
+
+
+import logging
+logger = logging.getLogger("resspublica")
+
+from .translations import *
+from .utils import *
+
+
+
+# The data set doesn't show directly the day/month only the year. But it does link to a webpage where we can fetch the date
+# We cache the date by id to avoid making too much requests
+def extract_announcement_date(observationId, observationUrl):
+    if db.contains(q.id == observationId):
+        logger.debug(f"observation {str(observationId)} was already in db.")
+        return date.fromisoformat(db.get(q.id == observationId)["date"])
+    else:
+        logger.debug(f"observation {str(observationId)} is a new observation.")
+        html = fetchUrlToHtml(observationUrl)
+        soup = BeautifulSoup(html, "html.parser")
+    
+        for row in soup.select("table tbody tr"):
+            cells = row.find_all("td")
+            if cells:
+                raw_date = cells[0].get_text(strip=True)
+                dt = date.strptime(raw_date, "%d.%m.%Y")
+                db.upsert({"id": observationId, "date": dt.isoformat()}, q.id == observationId)
+                return dt
+
+def generateBernAsianHornetFeed(ASSETS, CACHE):
+
+    arbitraryStartDate = date.fromisoformat("2026-01-01") # start date for weekly image generation
+
+    logger.info("Generating Asian Hornets sightings in Bern feed...")
+    ASIAN_HORNETS_DB_PATH = CACHE / "bernAsianHornet.json"
+    global db
+    global q
+    db = TinyDB(ASIAN_HORNETS_DB_PATH)
+    q = Query()
+
+
+    # 1. Load sightings
+    url = "https://geofiles.be.ch/geoportal/pub/download/ASHORNIS/ashornis_sichtnet.parquet"
+    df = pd.read_parquet(url)
+
+    # 2. Convert WKB geometry safely
+    def safe_load(x):
+        try:
+            return wkb.loads(x)
+        except Exception:
+            return None
+
+    df["geometry"] = df["geometry"].apply(safe_load)
+    df = df.dropna(subset=["geometry"])
+
+    gdf = gpd.GeoDataFrame(df, geometry="geometry", crs="EPSG:2056")
+
+    # We create a weekly image
+    # We just need to check if we haven't already created it
+    for start, end in weeklyRangesFrom(arbitraryStartDate): # arbitrary start date
+        if Path( CACHE / f"bernASianHornets-fr-{start.isoformat()}-{end.isoformat()}.png").exists(): # we only check french, but if one language exists the other ones should also
+            logger.debug(f"Week {start.isoformat()}-{end.isoformat()} was already cached. Skipping...")
+            continue
+        logger.debug(f"Generating week {start.isoformat()}-{end.isoformat()} ...")
+    
+        weeklySubsetIdList = [] # we can only select a subset by giving a list of id
+    
+        logger.debug(f"{str(len(gdf))} observations found")
+        for observations in gdf.itertuples():
+            # each observation is a tuple similar to 
+            #Pandas(Index=0, objectid=6178, meldjahr=2026, anzsichd=nan, anzsichy=1, urlinf_de='https://www.inforama.ch/images/global/beratung/PflanzenbauTierhaltung/Bienen/Asiatische-Hornisse/Sichtungen-von-Asiatischen-Hornissen.pdf', urlinf_fr='https://www.inforama.ch/images/global/beratung/PflanzenbauTierhaltung/Bienen/Asiatische-Hornisse/Observations-de-frelons-asiatiques.pdf', urlah_de='https://geofiles.be.ch/geoportal/pub/zusatzdaten/ASHORNIS/ASHORNIS_22_25_DE.html', urlah_fr='https://geofiles.be.ch/geoportal/pub/zusatzdaten/ASHORNIS/ASHORNIS_22_25_FR.html', katanzsid=0, katanzsiy=1, geometry=<POLYGON ((2594000 1180000, 2594000 1182000, 2596000 1182000, 2596000 118000...>, bbox={'min_x': 2594000.0, 'min_y': 1180000.0, 'max_x': 2596000.0, 'max_y': 1182000.0})
+
+            if start < extract_announcement_date(observations[1],observations[8]) < end:
+                weeklySubsetIdList.append(observations[1])
+            else:
+                logger.debug(f"observation {str(observations[1])} {extract_announcement_date(observations[1], observations[8]).isoformat()} is not in the range ({start.isoformat()} - {end.isoformat()})")
+        weeklySubset = gdf[gdf["objectid"].isin(weeklySubsetIdList)]
+
+        # as the canton borders do not change (frequently), we just download once the data
+        gdbPathToCantonBoundariesDirectory = ASSETS / "swissBOUNDARIES3D_1_5_LV95_LN02.gdb"
+
+        cantonsBoundariesData = gpd.read_file(
+            gdbPathToCantonBoundariesDirectory,
+            layer="TLM_KANTONSGEBIET"
+        )
+    
+        bernBoundaries = cantonsBoundariesData[cantonsBoundariesData["KANTONSNUMMER"] == 2].to_crs(gdf.crs)
+    
+        fig, ax = plt.subplots(figsize=(10, 10))
+    
+        bernBoundaries.plot(ax=ax, facecolor="none", edgecolor="black", linewidth=2)
+    
+        weeklySubset.plot(
+            ax=ax,
+            markersize=5,
+            color="red",
+            alpha=0.6
+        )
+        for lang in ["fr", "de", "rm", "it", "en"]:
+            plt.title(f"{translatedBernAsianHornetSightings[lang]} {start.isoformat()} - {end.isoformat()}")
+            plt.axis("off")
+            #plt.show()
+            plt.savefig(CACHE / f"bernASianHornets-{lang}-{start.isoformat()}-{end.isoformat()}.png", dpi=120, bbox_inches="tight")
+        logger.info("Finished creating images for Bern Asian Hornet Feeds. Preparing feed...")
+    feeds = {
+        "fr": [],
+        "it": [],
+        "de": [],
+        "rm": [],
+        "en": []
+    }
+    for start, end in weeklyRangesFrom(arbitraryStartDate): # arbitrary start date
+        weeklyEntry = {}
+
+        weeklyEntry["id"] = datetime.combine(start, time(12, 0)).timestamp() # we use epoch time as the id
+        weeklyEntry["creationDate"] = end.isoformat()
+        weeklyEntry["date"] = end.isoformat() # there is no point of updating it later, as (I think) data isn't retroactively put
+        weeklyEntry["source"] = "opendata.swiss"
+
+        for lang in ["fr", "it", "de", "rm", "en"]:
+            if lang == "rm":
+                weeklyEntry["url"] = "https://opendata.swiss/en/dataset/asiatische-hornisse" # there is no Romansh translation
+            else:
+                weeklyEntry["url"] = f"https://opendata.swiss/{lang}/dataset/asiatische-hornisse"
+            weeklyEntry["title"] = translatedBernAsianHornetSightings[lang] + f" {start.isoformat()}-{end.isoformat()}"
+            weeklyEntry["text"] = f"<img src=\"https://raw.githubusercontent.com/tomasriveral/ReSSPublica/refs/heads/main/.cache/bernASianHornets-{lang}-{start.isoformat()}-{end.isoformat()}.png\" alt=\"{translatedBernAsianHornetSightings[lang]} {start.isoformat()}-{end.isoformat()}\">" # yes there is an error in filename. the s in asian is capitalized. I don't really want to regenerate all the images...
+            feeds[lang].append(copy.deepcopy(weeklyEntry))
+
+    generateFeed(
+        "Asian Hornet sightings in Bern",
+        "RSS feed of Asian Hornet sightings in Bern",
+        "asianHornetSightingsInBern",
+        "en",
+        ["rss", "atom"],
+        datetime.now().replace(tzinfo=ZoneInfo("Europe/Zurich")),
+        feeds["en"]
+    )
+    
+    generateFeed(
+        "Observations de frelons asiatiques à Berne",
+        "Flux RSS des observations de frelons asiatiques à Berne",
+        "observationsFrelonsAsiatiquesBerne",
+        "fr",
+        ["rss", "atom"],
+        datetime.now().replace(tzinfo=ZoneInfo("Europe/Zurich")),
+        feeds["fr"]
+    )
+    
+    generateFeed(
+        "Sichtungen von Asiatischen Hornissen in Bern",
+        "RSS-Feed der Sichtungen von Asiatischen Hornissen in Bern",
+        "sichtungenAsiatischerHornissenBern",
+        "de",
+        ["rss", "atom"],
+        datetime.now().replace(tzinfo=ZoneInfo("Europe/Zurich")),
+        feeds["de"]
+    )
+    
+    generateFeed(
+        "Avvistamenti di calabroni asiatici a Berna",
+        "Feed RSS degli avvistamenti di calabroni asiatici a Berna",
+        "avvistamentiCalabroniAsiaticiBerna",
+        "it",
+        ["rss", "atom"],
+        datetime.now().replace(tzinfo=ZoneInfo("Europe/Zurich")),
+        feeds["it"]
+    )
+    
+    generateFeed(
+        "Observaziuns da vespras asiaticas a Berna",
+        "Feed RSS da las observaziuns da vespras asiaticas a Berna",
+        "observaziunsVesprasAsiaticasBerna",
+        "rm",
+        ["rss", "atom"],
+        datetime.now().replace(tzinfo=ZoneInfo("Europe/Zurich")),
+        feeds["rm"]
+    )

resspublica/main.py

@@ -2,19 +2,22 @@ from pathlib import Path
 import os
 import argparse
 import logging
-
-from .translations import *
-
 logger = logging.getLogger("resspublica")
+from datetime import date
 
 from .utils import *
 from .federalInitiativesFeeds import *
 from .translations import *
-
+from .bernAsianHornet import *
 
 def main():
     parser = argparse.ArgumentParser()
     parser.add_argument("--verbose", action="store_true", help="Enable debug logging")
+    parser.add_argument("--gen_federalInitiatives", action="store_true", help="Generate federal popular initiatives feed")
+    parser.add_argument("--gen_bernAsianHornets", action="store_true", help="Generate Asian hornets sightings in Bern feed (only Mondays)")
+    parser.add_argument("--force_gen_bernAsianHornets", action="store_true", help="Generate Asian hornets sightings in Bern feed even when not Monday")
+
+
     args = parser.parse_args()
 
     logging.basicConfig(
@@ -24,7 +27,12 @@ def main():
 
     CACHE = Path(os.environ.get("RESSPUBLICA_CACHE", ".cache"))
     CACHE.mkdir(parents=True, exist_ok=True)
+    ASSETS = Path(os.environ.get("RESSPUBLICA_ASSETS"))
+    ASSETS.mkdir(parents=True, exist_ok=True)
 
     logging.info("Starting feed generation")
-    generateFederalFeed(CACHE)
+    if args.gen_federalInitiatives:
+        generateFederalFeed(CACHE)
+    if (date.today().weekday() == 0 and args.gen_bernAsianHornets) or args.force_gen_bernAsianHornets: 
+        generateBernAsianHornetFeed(ASSETS, CACHE)
     logging.info("Done")

resspublica/translations.py

@@ -69,3 +69,10 @@ translatedSignatureDates = {
     "de": ("<p>Die Unterschriftensammlung begann am", "und endet ungefähr im</p>"),
     "it": ("<p>La raccolta delle firme è iniziata il", "e terminerà circa nel</p>"),
 }
+translatedBernAsianHornetSightings = {
+    "en": ("Asian Hornet sightings in Bern"),
+    "fr": ("Observations de frelons asiatiques à Berne"),
+    "de": ("Sichtungen von Asiatischen Hornissen in Bern"),
+    "it": ("Avvistamenti di calabroni asiatici a Berna"),
+    "rm": ("Observaziuns da vespras asiaticas a Berna"),
+}

resspublica/utils.py

@@ -1,12 +1,22 @@
 from feedgen.feed import FeedGenerator
-from datetime import datetime
+from datetime import datetime, date, timedelta
 from dateutil.relativedelta import relativedelta
-from .translations import *
 from zoneinfo import ZoneInfo
+from io import BytesIO
+import pycurl
+
+from .translations import *
+
 import logging
 logger = logging.getLogger("resspublica")
 
 def generateFeed(title, description, fileName, language, standards, lastUpdateTime, entries):
+    # entries must be a dictionary with the following keys:
+    # title, id, date, creationDate, url, source, text (html format)
+    # other keys will be ignored
+
+    # Note : if you want to only show an image, but the html code for an image in text
+
     logger.info(f"Generating feed {title}")
     fg = FeedGenerator()
     fg.title(title)
@@ -70,3 +80,39 @@ def getSignatureInfo(start_date, lang="fr"):
 
 def getValue(row, key):
     return row.get(key, {}).get("value") or None
+
+def fetchUrlToHtml(url: str) -> str:
+    buffer = BytesIO()
+
+    c = pycurl.Curl()
+    c.setopt(c.URL, url)
+    c.setopt(c.WRITEDATA, buffer)
+    c.setopt(c.FOLLOWLOCATION, True)
+    c.setopt(c.TIMEOUT, 20)
+    c.perform()
+    c.close()
+    
+    sleep(1) # avoid making requests too quickly. This shouldn't slow down much as we should try to cache as much possible
+
+    return buffer.getvalue().decode("utf-8", errors="replace")
+
+def weeklyRangesFrom(start_date: date):
+    today = date.today()
+
+    # ISO week start = Monday
+    def week_start(d):
+        return d - timedelta(days=d.weekday())
+
+    current_week_start = week_start(today)
+
+    # align first week start
+    start = week_start(start_date)
+
+    weeks = []
+
+    while start < current_week_start:
+        end = start + timedelta(days=6)
+        weeks.append((start, end))
+        start += timedelta(days=7)
+
+    return weeks