add option to publish specific stations to a certain topic (#720)

.github/workflows/tests-docker.yml

@@ -55,7 +55,7 @@ jobs:
         curl -X POST http://localhost/wis2downloader/subscriptions -H "Content-Type: application/json" -H "Authorization: Bearer github123" -d @test.json
     - name: populate stations from CSV 📡
       run: |
-        python3 wis2box-ctl.py execute wis2box metadata station publish-collection
+        python3 wis2box-ctl.py execute wis2box metadata station publish-collection --path /data/wis2box/metadata/station/station_list.csv
     - name: add Malawi synop data (csv2bufr synop_bufr template) 🇲🇼
       env:
         TOPIC_HIERARCHY: mw-mw_met_centre.data.core.weather.surface-based-observations.synop

docs/source/community/troubleshooting.rst

@@ -99,7 +99,7 @@ To add missing stations, use the station-editor in wis2box-webapp (from wis2box-
 .. code-block:: bash
 
    python3 wis2box-ctl.py login
-   wis2box metadata station publish-collection
+   wis2box metadata station publish-collection /data/wis2box/metadata/station/station_list.csv
 
 
 Error: no such container: wis2box-management

docs/source/reference/quickstart.rst

@@ -62,7 +62,7 @@ Load initial stations:
 
 .. code-block:: bash
 
-    wis2box metadata station publish-collection
+    wis2box metadata station publish-collection /data/wis2box/metadata/station/station_list.csv
 
 Ingest data using the data ingest command to push data to the ``wis2box-incoming`` bucket:
 
@@ -81,7 +81,7 @@ Publish stations:
 
 .. code-block:: bash
 
-    wis2box metadata station publish-collection
+    wis2box metadata station publish-collection /data/wis2box/metadata/station/station_list.csv
 
 Logout of wis2box-management container:
 

docs/source/reference/running/api-publishing.rst

@@ -22,7 +22,7 @@ will generate local station collection GeoJSON for API publication.
 
 .. code-block:: bash
 
-   wis2box metadata station publish-collection
+   wis2box metadata station publish-collection /data/wis2box/metadata/station/station_list.csv
 
 
 .. note:: This command also runs automatically at startup and thereafter every 10 minutes

docs/source/reference/running/station-metadata.rst

@@ -35,7 +35,7 @@ specifying one line per station as follows:
 
 .. literalinclude:: ../../../../examples/config/station_list.csv
 
-Then login in to the wis2box-container and run the command ``wis2box metadata station publish-collection``
+Then login in to the wis2box-container and run the command ``wis2box metadata station publish-collection /data/wis2box/metadata/station/station_list.csv``
 to insert all stations in the station list into the backend.
 
 Within the wis2box-container you can fetch the required station metadata from OSCAR/Surface using the following command:

docs/source/user/setup.rst

@@ -268,7 +268,11 @@ You can also bulk insert a set of stations from a CSV file, by defining the stat
 .. code-block:: bash
 
    python3 wis2box-ctl.py login
-   wis2box metadata station publish-collection
+   wis2box metadata station publish-collection --path /data/wis2box/metadata/stations/station_list.csv --topic-hierarchy mw-mw_met_centre.data.core.weather.surface-based-observations.synop
+
+.. note::
+
+   The ``topic-hierarchy`` option above is only an example; The actual topic hierarchy you should use is based on your centre id and dataset published.
 
 After doing a bulk insert please review the stations in wis2box-webapp and associate each station to the correct topics.
 

wis2box-management/wis2box/metadata/station.py

@@ -39,7 +39,7 @@
 from wis2box.api import (
     delete_collection_item, setup_collection, upsert_collection_item
 )
-from wis2box.env import (DATADIR, API_BACKEND_URL, DOCKER_API_URL,
+from wis2box.env import (API_BACKEND_URL, DOCKER_API_URL,
                          BROKER_HOST, BROKER_USERNAME, BROKER_PASSWORD,
                          BROKER_PORT)
 from wis2box.metadata.base import BaseMetadata
@@ -49,8 +49,6 @@
 
 LOGGER = logging.getLogger(__name__)
 
-STATION_METADATA = DATADIR / 'metadata' / 'station'
-STATIONS = STATION_METADATA / 'station_list.csv'
 WMDR_CODELISTS = Path('/home/wis2box/wmdr-codelists')
 
 
@@ -251,7 +249,8 @@ def check_station_datasets(wigos_station_identifier: str) -> Iterator[dict]:
             yield topic
 
 
-def add_topic_hierarchy(new_topic: str, territory_name: str, wsi: str) -> None:
+def add_topic_hierarchy(new_topic: str, territory_name: str = None,
+                        wsi: str = None) -> None:
     """
     Adds topic hierarchy to topic
 
@@ -264,9 +263,9 @@ def add_topic_hierarchy(new_topic: str, territory_name: str, wsi: str) -> None:
 
     valid_topics = [topic for link, topic in load_datasets()]
     if new_topic not in valid_topics:
-        msg = f'ERROR: Invalid topic: {new_topic}\n'
-        msg += 'Valid topics:\n'
-        msg += '\n'.join(valid_topics)
+        msg = (f'ERROR: Invalid topic: {new_topic}\n'
+               'Valid topics:\n'
+               '\n'.join(valid_topics))
         LOGGER.error(msg)
         return
 
@@ -293,23 +292,26 @@ def add_topic_hierarchy(new_topic: str, territory_name: str, wsi: str) -> None:
         upsert_collection_item('stations', feature)
 
 
-def publish_from_csv() -> None:
+def publish_from_csv(path: Path, topic: str = None) -> None:
     """
     Publishes station collection to API config and backend from csv
 
+    :param path: `Path` to station list CSV (default is `None`)
+    :param topic: `str` of topic hierarchy (default is `None`)
+
     :returns: `None`
     """
 
-    if not STATIONS.exists():
-        msg = f'Please create a station metadata file in {STATION_METADATA}'
+    if not path.exists():
+        msg = f'Station file {path} does not exist'
         LOGGER.error(msg)
         raise RuntimeError(msg)
 
     oscar_baseurl = 'https://oscar.wmo.int/surface/#/search/station/stationReportDetails'  # noqa
 
-    LOGGER.debug(f'Publishing station list from {STATIONS}')
+    LOGGER.debug(f'Publishing station list from {path}')
     station_list = []
-    with STATIONS.open() as fh:
+    with path.open() as fh:
         reader = csv.DictReader(fh)
 
         for row in reader:
@@ -378,6 +380,13 @@ def publish_from_csv() -> None:
                 LOGGER.debug(f'Station does not exist: {err}')
             upsert_collection_item('stations', feature)
 
+            if None not in [path, topic]:
+                msg = f"Adding topic {topic} to station {row['wigos_station_identifier']}"  # noqa
+                LOGGER.debug(msg)
+
+                add_topic_hierarchy(topic, row['territory_name'],
+                                    row['wigos_station_identifier'])
+
     LOGGER.info(f'Updated station list: {station_list}')
     # inform mqtt-metrics-collector
     notify_msg = {
@@ -495,11 +504,14 @@ def setup(ctx, verbosity):
 
 @click.command()
 @click.pass_context
+@cli_helpers.OPTION_PATH
+@cli_helpers.OPTION_TOPIC_HIERARCHY
 @cli_helpers.OPTION_VERBOSITY
-def publish_collection(ctx, verbosity):
+def publish_collection(ctx, path, topic_hierarchy, verbosity):
     """Publish from station_list.csv"""
 
-    publish_from_csv()
+    publish_from_csv(path, topic_hierarchy)
+
     click.echo('Done')