extend the sniffing capability to head out to the endpoint to find out what it is

some comments:
- it may return an array of protocols, if multiple are supported
- for backwards compatibility, you need to set first=False to allow multiple (multiple is also slow)
- you need to set extended=True to activate the extended sniffing

Author: pvgenuchten

geolinks/__init__.py

@@ -28,12 +28,21 @@
 # =================================================================
 
 import logging
+from owslib.wms import WebMapService as WMS
+from owslib.wfs import WebFeatureService as WFS
+from owslib.ogcapi.features import Features as OAPIF
+from owslib.ogcapi.coverages import Coverages as OAPIC
+from owslib.ogcapi.records import Records as OAPIR
+from owslib.wcs import WebCoverageService as WCS
+from owslib.csw import CatalogueServiceWeb as CSW
+from owslib.wps import WebProcessingService as WPS
+from owslib.sos import SensorObservationService as SOS
+from owslib.wmts import WebMapTileService as WMTS
 
 LOGGER = logging.getLogger(__name__)
 
 __version__ = '0.2-dev'
 
-
 def inurl(needles, haystack, position='any'):
     """convenience function to make string.find return bool"""
 
@@ -60,7 +69,7 @@ def inurl(needles, haystack, position='any'):
     return False
 
 
-def sniff_link(url):
+def sniff_link(url, extended=False, first=True):
     """performs basic heuristics to detect what the URL is"""
 
     protocol = None
@@ -105,6 +114,105 @@ def sniff_link(url):
     elif inurl(['kml', 'kmz'], link, 'end'):
         protocol = 'OGC:KML'
     else:
-        LOGGER.info('No link type detected')
+        if (extended):
+            protocol = []
+            #for each servicetype, head out to see if it is valid
+            try: 
+                wms = WMS(link)
+                if (wms.identification.type == 'OGC:WMS'):
+                    if (first):
+                        return wms.identification.type
+                    else:
+                        protocol.append(wms.identification.type)
+            except:
+                pass # No need to log?
+            try: 
+                wmts = WMTS(link)        
+                if (wmts.identification.type == 'OGC:WMTS'):
+                    if (first):
+                        return wmts.identification.type
+                    else:
+                        protocol.append(wmts.identification.type)
+            except:
+                pass
+            try:
+                wps = WPS(link, verbose=False, skip_caps=True)
+                wps.getcapabilities()
+                if (wps.identification.type == 'OGC:WPS'):
+                    if (first):
+                        return wps.identification.type
+                    else:
+                        protocol.append(wps.identification.type)
+            except:
+                pass 
+            try:
+                wfs = WFS(link)
+                if (wfs.identification.type == 'OGC:WFS'):
+                    if (first):
+                        return wfs.identification.type
+                    else:
+                        protocol.append(wfs.identification.type)
+            except:
+                pass
+            try:
+                csw = CSW('http://geodiscover.cgdi.ca/wes/serviceManagerCSW/csw')
+                if (csw.identification.type == 'OGC:CSW'):
+                    if (first):
+                        return csw.identification.type
+                    else:
+                        protocol.append(csw.identification.type)
+            except:
+                pass
+            try:
+                wcs = WCS(link)
+                if (wcs.identification.type == 'OGC:WCS'):
+                    if (first):
+                        return wcs.identification.type
+                    else:
+                        protocol.append(wcs.identification.type)
+            except:
+                pass
+            try:
+                sos = SOS(link)
+                if (sos.identification.type == 'OGC:SOS'):
+                    if (first):
+                        return sos.identification.type
+                    else:
+                        protocol.append(sos.identification.type)
+            except:
+                pass
+            try: 
+                oapir = OAPIR(link)
+                if (oapir.conformance()):
+                    if (first):
+                        return "OGCAPI:records"
+                    else:
+                        protocol.append("OGCAPI:records")
+            except:
+                pass
+            try: 
+                oapif = OAPIF(link)
+                if (oapir.conformance()):
+                    if (first):
+                        return "OGCAPI:features"
+                    else:
+                        protocol.append("OGCAPI:features")
+            except:
+                pass
+            try: 
+                oapic = OAPIC(link)
+                if (oapir.conformance()):
+                    if (first):
+                        return "OGCAPI:coverages"
+                    else:
+                        protocol.append("OGCAPI:coverages")
+            except:
+                pass
+
+            if len(protocol) == 1:
+                protocol = protocol[0]
+
+        else:
+            LOGGER.info('No link type detected')
 
     return protocol

tests/run_tests.py

@@ -47,7 +47,7 @@ def test_link_types(self):
         """simple link type tests"""
 
         for test in self.test_data['test_data']:
-            self.assertEqual(sniff_link(test['link']), test['expected'],
+            self.assertEqual(sniff_link(test['link'],extended=True), test['expected'],
                              'Expected %s and %s to be equal' %
                              (test['link'], test['expected']))
 

tests/test_data.json

@@ -1,6 +1,7 @@
 {
     "test_data": [
         {"link": "http://host/wms?service=WMS", "expected": "OGC:WMS"},
+        {"link": "https://maps.isric.org/mapserv?map=/map/bdod.map", "expected": "OGC:WMS"},
         {"link": "http://host/ows?service=WFS", "expected": "OGC:WFS"},
         {"link": "http://host/ows?service=WCS", "expected": "OGC:WCS"},
         {"link": "http://host/ows?service=WPS", "expected": "OGC:WPS"},