758 adding wiskis query support

docs/source/Explanation/CommandLineGuide.rst

@@ -2501,6 +2501,9 @@ of colon separation in files, the default in Sarracenia is WHATFN
 
 The possible keywords are :
 
+**None**
+ - no Sundew-style filename processing. Leave the filename alone.
+   (Not the same as NONE, described below.)
 
 **WHATFN**
  - the first part of the Sundew filename (string before first :)

docs/source/Reference/sr3_options.7.rst

@@ -749,6 +749,9 @@ of colon separation in files, the default in Sarracenia is WHATFN
 
 The possible keywords are :
 
+**None**
+ - the filename is not modified at all. (different from NONE!) 
+   turn off any Sundew compatibility filename processing.
 
 **WHATFN**
  - the first part of the Sundew filename (string before first :)

docs/source/fr/Explication/GuideLigneDeCommande.rst

@@ -2491,6 +2491,10 @@ de la séparation par des deux-points dans les fichiers, le défaut dans Sarrace
 
 Les mots-clés possibles sont :
 
+**None**
+ - Aucune modification du nom de fichier (enlever toute interprétation de style Sundew)
+   N.B. différent de NONE décrit plus loin.
+
 **WHATFN**
  - la première partie du nom de fichier Sundew (chaîne de caractères avant le premier : )
 

docs/source/fr/Reference/sr3_options.7.rst

@@ -738,6 +738,10 @@ de la séparation par des deux-points dans les fichiers, le défaut dans Sarrace
 
 Les mots-clés possibles sont :
 
+**None**
+ - Aucune modification du nom de fichier (enlever toute interprétation de style Sundew)
+   N.B. différent de NONE décrit plus loin.
+
 **WHATFN**
  - la première partie du nom de fichier Sundew (chaîne de caractères avant le premier : )
 

sarracenia/examples/flow/opg.conf

@@ -0,0 +1,30 @@
+#
+# Ontario Power Generation
+# a scheduled flow to bring some observations from their WISKI server.
+# 
+#wiski_ts_length 24h
+#wiski_ts_name caw_Cmd
+#wiski_ts_parameterTypeName Air Temperature
+
+
+callback scheduled.wiski
+
+callback post.message
+
+wiski_ts_length 24h
+wiski_ts_name caw_Cmd
+
+scheduled_hour 0,6,12,18
+scheduled_minute 17
+
+post_broker amqp://tfeed@localhost
+post_exchange xs_tfeed_opg
+
+pollUrl https://kiwis.opg.com
+
+# to ignore colons in the filenames generated.
+filename None
+
+post_baseUrl file:/
+post_baseDir /tmp/wiski
+directory /tmp/wiski

sarracenia/flowcb/clamav.py

@@ -18,13 +18,14 @@
 import time
 
 from sarracenia import nowflt
-from sarracenia.featuredetection import features
 import sarracenia
 from sarracenia.flowcb import FlowCB
 
 #
 # Support for features inventory mechanism.
 #
+from sarracenia.featuredetection import features
+
 features['clamd'] = { 'modules_needed': [ 'pyclamd' ], 'Needed': True,
         'lament' : 'cannot use clamd to av scan files transferred',
         'rejoice' : 'can use clamd to av scan files transferred' }

sarracenia/flowcb/scheduled/__init__.py

@@ -65,7 +65,7 @@ def update_appointments(self,when):
         logger.info( f"for {when}: {json.dumps(list(map( lambda x: str(x), self.appointments))) } ")
 
 
-    def __init__(self,options):
+    def __init__(self,options,logger=logger):
         super().__init__(options,logger)
         self.o.add_option( 'scheduled_interval', 'duration', 0 )
         self.o.add_option( 'scheduled_hour', 'list', [] )
@@ -166,7 +166,7 @@ def wait_until( self, appointment ):
     def wait_until_next( self ):
 
         if self.o.scheduled_interval > 0:
-            self.wait_seconds(self.o.scheduled_interval)
+            self.wait_seconds(datetime.timedelta(seconds=self.o.scheduled_interval))
             return
 
         if ( len(self.o.scheduled_hour) > 0 ) or ( len(self.o.scheduled_minute) > 0 ):

sarracenia/flowcb/scheduled/wiski.py

@@ -0,0 +1,215 @@
+import logging
+import requests
+import base64
+
+import datetime
+import os
+import sys
+import time
+
+from datetime import date
+
+import sarracenia
+from sarracenia.flowcb.scheduled import Scheduled
+
+#
+# Support for features inventory mechanism.
+#
+from sarracenia.featuredetection import features
+
+features['wiski'] = { 'modules_needed': [ 'kiwis_pie' ], 'Needed': True,
+        'lament' : 'cannot poll sites that provide data using WISKI API' ,
+        'rejoice' : 'can poll sites that provide data with WISKI API' }
+
+try:
+    from kiwis_pie import KIWIS
+    sarracenia.features['wiski']['present'] = True
+except:
+    sarracenia.features['wiski']['present'] = False
+
+
+
+logger = logging.getLogger(__name__)
+
+
+
+
+class Wiski(Scheduled):
+    """
+
+       Plugin to Poll a WISKIS server that uses  ( https://www.kisters.net/wiski )
+       Uses the kiwis_pie ( https://github.com/amacd31/kiwis_pie ) to do that.
+
+       In the credentials.conf file, need an authentication  entry like:
+
+       https://user:password@theserver.com
+
+       and in the config file:
+
+       pollUrl https://theserver.com
+
+       wiski polling parametrization (these are defined by and passed to kiwis_pie):
+
+       wiski_ts_length -- how long a timeseries to request (default is 24 hours)
+       wiski_ts_name   -- name of the timeseries (not used currently.)
+       wiski_ts_parameterTypeName -- Air Temperature.
+
+       inherits the following settings from Scheduled (interval overrides the other two.)
+
+       scheduled_hour    -- a list of hours (0-59) to poll.
+       scheduled_minute -- a list of minutes (0-59) to poll. 
+       scheduled_interval  -- just a time interval between polls.
+
+       2023/09 plugin by Peter Silva, based on original tidbits by Mohamed Rehouma.
+    """
+
+    def __init__(self,options):
+        super().__init__(options,logger)
+
+        # use self.o.PollUrl for credentials.
+        self.details = None
+        if self.o.pollUrl is not None:
+            ok, self.details = sarracenia.config.Config.credentials.get(
+                self.o.pollUrl)
+
+        if self.o.pollUrl is None or self.details == None:
+            logger.error("pollUrl option incorrect or missing\n")
+            sys.exit(1)
+
+        # assert: now self.details.url.username/password are set.
+        self.basic_auth = "Basic " + base64.b64encode(f"{self.details.url.username}:{self.details.url.password}".encode()).decode()
+
+        if self.details.url.port:
+            self.main_url = self.details.url.scheme + "://" + self.details.url.hostname + ":" + self.details.url.port + "/KiWIS/KiWIS"
+        else:
+            self.main_url = self.details.url.scheme + "://" + self.details.url.hostname + "/KiWIS/KiWIS"
+
+        self.host = self.details.url.hostname
+        self.token = None
+        #self.token = self.submit_tokenization_request()
+
+        # meteorological parameter settings.
+        self.o.add_option( 'wiski_ts_length', 'duration', '24h' )
+        self.o.add_option( 'wiski_ts_name', 'str', 'caw_Cmd' )
+        self.o.add_option( 'wiski_ts_parameterTypeName', 'str', 'Air Temperature' )
+
+        self.ts_length = datetime.timedelta( seconds=self.o.wiski_ts_length )
+
+        logger.info( f"main_url: {self.main_url} timeseries_length={self.ts_length}, writing to {self.o.directory}" )
+
+
+
+    def submit_tokenization_request(self):
+        # ECCC User Definition
+
+        logger.info("requesting a new token")
+
+        body_string = "grant_type=client_credentials&scope=empty"
+
+        # Request Token using HTTP POST request
+        token_url = "https://" + self.host + "/KiWebPortal/rest/auth/oidcServer/token"
+        headers = {
+            "Content-Type": "application/x-www-form-urlencoded",
+            "Authorization": self.basic_auth,
+            "Host": self.host,
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+        }
+
+        response = requests.post(token_url, data=body_string, headers=headers)
+
+        str_response_status = response.status_code
+        str_response_text = response.text
+
+        # Process response
+        token_array = str_response_text.split('"')
+        if str_response_status == 201:
+            submit_tokenization_request = token_array[3]
+        else:
+            logger.error( f"tokenization_request status: {str_response_text}" )
+            submit_tokenization_request = "Error"
+
+        return submit_tokenization_request
+
+    def gather(self): # placeholder
+
+        messages=[]
+
+        self.wait_until_next()
+
+        while (1):
+            if self.stop_requested:
+                return messages
+
+            self.token = self.submit_tokenization_request()
+            authenticated_url = self.main_url
+            headers = {
+                "Authorization" : f"Bearer {self.token}",
+                "Content-Type" : "application/json"
+            }
+
+            response = requests.get(authenticated_url,headers=headers)
+
+            logger.info(response)
+
+            if response.status_code == 200:
+                break
+            else:
+                logger.info( f"request failed. Status code: {response.status_code}: {response.text}" )
+
+        k = KIWIS(self.main_url, headers=headers )
+
+        now = datetime.datetime.fromtimestamp(time.time(),datetime.timezone.utc)
+        then = now - self.ts_length
+
+        logger.info( f"stations: {k.get_station_list().station_id} " )
+
+        for station_id in k.get_station_list().station_id:
+
+            if self.stop_requested:
+                return messages
+
+            timeseries = k.get_timeseries_list(station_id = station_id ).ts_id
+            #logger.info( f"looping over the timeseries: {timeseries}" )
+
+            #timeseries = k.get_timeseries_list(station_id = station_id, ts_name =self.o.wiski_ts_name, parametertype_name = self.o.wiski_ts_parameterTypeName ).ts_id
+            for ts_id in timeseries:
+                # writing files on windows is quite painful, so many illegal characters.
+                if sys.platform.startswith( "win" ):
+                    fname = f"{self.o.directory}{os.sep}ts_{ts_id}_{station_id}__{str(now).replace(' ','T')}.csv" 
+                    fname = fname[0:3]+fname[3:].replace(':','_').replace('.','_',1).replace('+','_').replace('-','_')
+                else:
+                    fname = f"{self.o.directory}{os.sep}ts_{ts_id}_{station_id}_{str(then).replace(' ','T')}_{str(now).replace(' ','T')}.csv" 
+
+                logger.info( f"Timeseries {ts_id} for station_id {station_id} to be written to: {fname}" )
+                f=open(fname,'w')
+                #ts=k.get_timeseries_values(ts_id = ts_id, to = date(2023,1,31), **{'from': date(2023,1,1)})
+                ts=k.get_timeseries_values(ts_id = ts_id, to = now, **{'from': then})
+                if len(ts) > 0:
+                    ts.to_csv(f)
+                else:
+                    logger.info( f"no data to write to {f}")
+                f.close() 
+                messages.append( sarracenia.Message.fromFileData( fname, self.o, os.stat(fname) ) )
+
+        return messages
+
+if __name__ == '__main__':
+
+    import sarracenia.config
+    import types
+    import sarracenia.flow
+
+    options = sarracenia.config.default_config()
+    flow = sarracenia.flow.Flow(options)
+    flow.o.scheduled_interval= 5
+    flow.o.pollUrl = "https://kiwis.opg.com"
+    if sys.platform.startswith( "win" ):
+        flow.o.directory = "C:\\temp\wiski"
+    else:
+        flow.o.directory = "/tmp/wiski"
+    logging.basicConfig(level=logging.DEBUG)
+
+    me = Wiski(flow.o)
+    me.gather()
+