Merge branch '146-check-if-we-can-access-station-s-operator-from-stationxml' into 'main'

Resolve "Check if we can access station's operator from stationxml"

Closes #146

See merge request !142

alembic/versions/2024_01_09_1706-583315eadd30_add_operator_table.py

+"""add operator table
+
+Revision ID: 583315eadd30
+Revises: d5df53b6a17a
+Create Date: 2024-01-09 17:06:07.703796
+
+"""
+import sqlalchemy as sa
+
+from alembic import op
+
+# revision identifiers, used by Alembic.
+revision = "583315eadd30"
+down_revision = "d5df53b6a17a"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table(
+        "operator",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("agency", sa.String(), nullable=False),
+        sa.Column("website", sa.String(), nullable=True),
+        sa.PrimaryKeyConstraint("id", name=op.f("pk_operator")),
+        sa.UniqueConstraint("agency", name=op.f("uq_operator_agency")),
+    )
+    op.create_table(
+        "operated_by",
+        sa.Column("station_id", sa.Integer(), nullable=False),
+        sa.Column("operator_id", sa.Integer(), nullable=False),
+        sa.ForeignKeyConstraint(
+            ["operator_id"],
+            ["operator.id"],
+            name=op.f("fk_operated_by_operator_id_operator"),
+        ),
+        sa.ForeignKeyConstraint(
+            ["station_id"],
+            ["station.id"],
+            name=op.f("fk_operated_by_station_id_station"),
+        ),
+        sa.PrimaryKeyConstraint(
+            "station_id", "operator_id", name=op.f("pk_operated_by")
+        ),
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade() -> None:
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table("operated_by")
+    op.drop_table("operator")
+    # ### end Alembic commands ###

app.ini

@@ -8,6 +8,7 @@ listen = localhost:6543
 [pshell]
 # models
 Network = waveqc.models.Network
+Operator = waveqc.models.Operator
 Station = waveqc.models.Station
 Channel = waveqc.models.Channel
 Check = waveqc.models.Check

waveqc/models.py

@@ -20,8 +20,10 @@ from sqlalchemy import (
     Text,
     UniqueConstraint,
     and_,
+    delete,
     extract,
     func,
+    not_,
     or_,
     select,
     update,
@@ -92,6 +94,30 @@ def checks_scope(dbsession: scoped_session[Session]) -> Subquery:
     return select(func.generate_series(start, end, delta).label("day")).subquery()
 
 
+class OperatedBy(Base):
+    __tablename__ = "operated_by"
+
+    station_id: Mapped[int] = mapped_column(ForeignKey("station.id"), primary_key=True)
+    operator_id: Mapped[int] = mapped_column(
+        ForeignKey("operator.id"), primary_key=True
+    )
+
+
+class Operator(Base):
+    __tablename__ = "operator"
+
+    id: Mapped[int] = mapped_column(primary_key=True)  # noqa: A003
+    agency: Mapped[str] = mapped_column(unique=True)
+    website: Mapped[str | None]
+
+    stations: Mapped[list["Station"]] = relationship(
+        secondary="operated_by", back_populates="operators"
+    )
+
+    def __repr__(self) -> str:
+        return f"Operator({self.agency})"
+
+
 class Network(Base):
     __tablename__ = "network"
 
@@ -149,6 +175,57 @@ class Network(Base):
             )
         )
 
+    def link_operators_to_stations(
+        self: "Network",
+        dbsession: scoped_session[Session],
+        stations: list[ObspyStation],
+    ) -> None:
+        for item in stations:
+            station = dbsession.scalars(
+                select(Station).where(
+                    Station.network_id == self.id, Station.code == item.code
+                )
+            ).one()
+            # Remove old operators from stations
+            for operator in station.operators:
+                agencies = [obspy_operator.agency for obspy_operator in item.operators]
+                if operator.agency not in agencies:
+                    station.operators.remove(operator)
+            # Add new operators to stations
+            for operator in item.operators:
+                agency = dbsession.scalars(
+                    select(Operator).where(Operator.agency == operator.agency)
+                ).one()
+                if agency not in station.operators:
+                    station.operators.append(agency)
+
+    @staticmethod
+    def purge_orphaned_operators(dbsession: scoped_session[Session]) -> None:
+        dbsession.execute(delete(Operator).where(not_(Operator.stations.any())))
+
+    def populate_operators(
+        self: "Network",
+        dbsession: scoped_session[Session],
+        stations: list[ObspyStation],
+    ) -> None:
+        operators = [
+            dict(item)
+            for item in {
+                tuple({"agency": operator.agency, "website": None}.items())
+                for station in stations
+                for operator in station.operators
+            }
+        ]
+        insert_stmt = insert(Operator).values(operators)
+        update_stmt = {
+            code.name: code for code in insert_stmt.excluded if code.name not in ["id"]
+        }
+        dbsession.execute(
+            insert_stmt.on_conflict_do_update(
+                constraint="uq_operator_agency", set_=update_stmt
+            )
+        )
+
     def populate_channels(
         self: "Network",
         dbsession: scoped_session[Session],
@@ -219,6 +296,9 @@ class Station(Base):
         back_populates="station",
         cascade="all, delete",
     )
+    operators: Mapped[list["Operator"]] = relationship(
+        secondary="operated_by", back_populates="stations"
+    )
 
     def __repr__(self: "Station") -> str:
         return f"Station(code={self.code})"

waveqc/tasks.py

@@ -108,7 +108,10 @@ def update_inventory(date: str) -> None:
     # We need to update all stations first to get an end_date if available
     for inventory_network in inventory.networks:
         network = Network().populate(dbsession, inventory_network)
+        network.populate_operators(dbsession, inventory_network.stations)
         network.populate_stations(dbsession, inventory_network.stations)
+        network.link_operators_to_stations(dbsession, inventory_network.stations)
+        network.purge_orphaned_operators(dbsession)
     target = UTCDateTime(date)
     inventory = client.get_stations(
         network=",".join(settings.MONITORED_NETWORKS),