colors spatial feature

3 files changed, 169 insertions, 0 deletions

diff --git a/bsie/extractor/image/__init__.py b/bsie/extractor/image/__init__.py
new file mode 100644
index 0000000..75b118d
--- /dev/null
+++ b/bsie/extractor/image/__init__.py
@@ -0,0 +1,13 @@
+"""
+
+Part of the bsie module.
+A copy of the license is provided with the project.
+Author: Matthias Baumgartner, 2022
+"""
+# standard imports
+import typing
+
+# exports
+__all__: typing.Sequence[str] = []
+
+## EOF ##
diff --git a/bsie/extractor/image/colors_spatial.py b/bsie/extractor/image/colors_spatial.py
new file mode 100644
index 0000000..fa31ea7
--- /dev/null
+++ b/bsie/extractor/image/colors_spatial.py
@@ -0,0 +1,155 @@
+"""Spatial color features.
+
+Part of the bsie module.
+A copy of the license is provided with the project.
+Author: Matthias Baumgartner, 2022
+"""
+# standard imports
+import typing
+
+# external imports
+import PIL.Image
+import numpy as np
+
+# bsie imports
+from bsie.utils import bsfs, node, ns
+
+# inner-module imports
+from .. import base
+
+# constants
+FEATURE_NAME = ns.bsf + 'ColorsSpatial'
+PREDICATE_NAME = ns.bse + 'colors_spatial'
+
+# exports
+__all__: typing.Sequence[str] = (
+    'ColorsSpatial',
+    )
+
+
+## code ##
+
+class ColorsSpatial(base.Extractor):
+    """Determine dominant colors of subregions in the image.
+
+    Computes the domiant color of increasingly smaller subregions of the image.
+    """
+
+    CONTENT_READER = 'bsie.reader.image.Image'
+
+    # Initial subregion width.
+    width: int
+
+    # Initial subregion height.
+    height: int
+
+    # Decrement exponent.
+    exp: float
+
+    # Principal predicate's URI.
+    _predicate_name: bsfs.URI
+
+    def __init__(
+            self,
+            width: int = 32,
+            height: int = 32,
+            exp: float = 4.,
+            ):
+        # instance identifier
+        uuid = bsfs.uuid.UCID.from_dict({
+            'width': width,
+            'height': height,
+            'exp': exp,
+            })
+        # determine symbol names
+        instance_name = FEATURE_NAME[uuid]
+        predicate_name = PREDICATE_NAME[uuid]
+        # get vector dimension
+        dimension = self.dimension(width, height, exp)
+        # initialize parent with the schema
+        super().__init__(bsfs.schema.from_string(base.SCHEMA_PREAMBLE + f'''
+            <{FEATURE_NAME}> rdfs:subClassOf bsfs:Feature ;
+                # annotations
+                rdfs:label "Spatially dominant colors"^^xsd:string ;
+                schema:description "Domiant colors of subregions in an image."^^xsd:string .
+
+            <{instance_name}> rdfs:subClassOf <{FEATURE_NAME}> ;
+                bsfs:dimension "{dimension}"^^xsd:integer ;
+                # annotations
+                <{FEATURE_NAME}/args#width> "{width}"^^xsd:integer ;
+                <{FEATURE_NAME}/args#height> "{height}"^^xsd:integer ;
+                <{FEATURE_NAME}/args#exp> "{exp}"^^xsd:float .
+
+            <{predicate_name}> rdfs:subClassOf bsfs:Predicate ;
+                rdfs:domain bsfs:File ;
+                rdfs:range <{instance_name}> ;
+                bsfs:unique "true"^^xsd:boolean .
+
+            '''))
+        # assign extra members
+        self.width = width
+        self.height = height
+        self.exp = exp
+        self._predicate_name = predicate_name
+
+    def __repr__(self) -> str:
+        return f'{bsfs.typename(self)}({self.width}, {self.height}, {self.exp})'
+
+    def __eq__(self, other: typing.Any) -> bool:
+        return super().__eq__(other) \
+          and self.width == other.width \
+          and self.height == other.height \
+          and self.exp == other.exp
+
+    def __hash__(self) -> int:
+        return hash((super().__hash__(), self.width, self.height, self.exp))
+
+    @staticmethod
+    def dimension(width: int, height: int, exp: float) -> int:
+        """Return the feature vector dimension."""
+        # FIXME: replace with a proper formula
+        dim = 0
+        while width >= 1 and height >= 1:
+            dim += width * height
+            width = np.floor(width / exp)
+            height = np.floor(height / exp)
+        dim *= 3 # per band
+        return int(dim)
+
+    def extract(
+            self,
+            subject: node.Node,
+            content: PIL.Image,
+            principals: typing.Iterable[bsfs.schema.Predicate],
+            ) -> typing.Iterator[typing.Tuple[node.Node, bsfs.schema.Predicate, typing.List[float]]]:
+        # check principals
+        if self.schema.predicate(self._predicate_name) not in principals:
+            # nothing to do; abort
+            return
+
+        # convert to HSV
+        content = content.convert('HSV')
+
+        # get dimensions
+        width, height = self.width, self.height
+        num_bands = len(content.getbands()) # it's three since we converted to HSV before
+
+        features = []
+        while width >= 1 and height >= 1:
+            # downsample
+            img = content.resize((width, height), resample=PIL.Image.Resampling.BOX)
+            # feature vector
+            features.append(
+                np.array(img.getdata()).reshape((width * height, num_bands)))
+            # iterate
+            width = int(np.floor(width / self.exp))
+            height = int(np.floor(height / self.exp))
+
+        # combine features
+        value = np.vstack(features)
+        # convert features
+        value = value.reshape(-1).tolist() # several bands
+        # return triple with feature vector as value
+        yield subject, self.schema.predicate(self._predicate_name), value
+
+## EOF ##
diff --git a/bsie/utils/namespaces.py b/bsie/utils/namespaces.py
index 2d0b535..393b436 100644
--- a/bsie/utils/namespaces.py
+++ b/bsie/utils/namespaces.py
@@ -15,6 +15,7 @@ bse = _bsfs.Namespace('http://bsfs.ai/schema/Entity')
 bsfs = _bsfs.Namespace('http://bsfs.ai/schema', fsep='/')
 bsm = _bsfs.Namespace('http://bsfs.ai/schema/Meta')
 xsd = _bsfs.Namespace('http://www.w3.org/2001/XMLSchema')
+bsf = _bsfs.Namespace('http://ie.bsfs.ai/schema/Feature')
 
 # export
 __all__: typing.Sequence[str] = (