[AnomalyDetection] Support offline detectors

sdks/python/apache_beam/ml/anomaly/detectors/offline.py

@@ -0,0 +1,59 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements.  See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License.  You may obtain a copy of the License at
+#
+#    http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from typing import Any
+from typing import Dict
+from typing import Optional
+
+import apache_beam as beam
+from apache_beam.ml.anomaly.base import AnomalyDetector
+from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.inference.base import KeyedModelHandler
+
+
+@specifiable
+class OfflineDetector(AnomalyDetector):
+  """A offline anomaly detector that uses a provided model handler for scoring.
+
+  Args:
+    keyed_model_handler: The model handler to use for inference.
+      Requires a `KeyModelHandler[Any, Row, float, Any]` instance.
+    run_inference_args: Optional arguments to pass to RunInference
+    **kwargs: Additional keyword arguments to pass to the base
+      AnomalyDetector class.
+  """
+  def __init__(
+      self,
+      keyed_model_handler: KeyedModelHandler[Any, beam.Row, float, Any],
+      run_inference_args: Optional[Dict[str, Any]] = None,
+      **kwargs):
+    super().__init__(**kwargs)
+
+    # TODO: validate the model handler type
+    self._keyed_model_handler = keyed_model_handler
+    self._run_inference_args = run_inference_args or {}
+
+    # always override model_identifier with model_id from the detector
+    self._run_inference_args["model_identifier"] = self._model_id
+
+  def learn_one(self, x: beam.Row) -> None:
+    """Not implemented since CustomDetector invokes RunInference directly."""
+    raise NotImplementedError
+
+  def score_one(self, x: beam.Row) -> Optional[float]:
+    """Not implemented since CustomDetector invokes RunInference directly."""
+    raise NotImplementedError

sdks/python/apache_beam/ml/anomaly/specifiable.py

@@ -21,21 +21,22 @@
 
 from __future__ import annotations
 
+import abc
 import collections
 import dataclasses
 import inspect
 import logging
 import os
 from typing import Any
+from typing import Callable
 from typing import ClassVar
 from typing import Dict
 from typing import List
 from typing import Optional
-from typing import Protocol
 from typing import Type
 from typing import TypeVar
 from typing import Union
-from typing import runtime_checkable
+from typing import overload
 
 from typing_extensions import Self
 
@@ -59,7 +60,7 @@
 #: `spec_type` when applying the `specifiable` decorator to an existing class.
 _KNOWN_SPECIFIABLE = collections.defaultdict(dict)
 
-SpecT = TypeVar('SpecT', bound='Specifiable')
+T = TypeVar('T', bound=type)
 
 
 def _class_to_subspace(cls: Type) -> str:
@@ -104,8 +105,7 @@ class Spec():
   config: Optional[Dict[str, Any]] = dataclasses.field(default_factory=dict)
 
 
-@runtime_checkable
-class Specifiable(Protocol):
+class Specifiable(abc.ABC):
   """Protocol that a specifiable class needs to implement."""
   #: The value of the `type` field in the object's spec for this class.
   spec_type: ClassVar[str]
@@ -130,7 +130,9 @@ def _from_spec_helper(v, _run_init):
     return v
 
   @classmethod
-  def from_spec(cls, spec: Spec, _run_init: bool = True) -> Union[Self, type]:
+  def from_spec(cls,
+                spec: Spec,
+                _run_init: bool = True) -> Union[Self, type[Self]]:
     """Generate a `Specifiable` subclass object based on a spec.
 
     Args:
@@ -250,13 +252,35 @@ def _get_init_kwargs(inst, init_method, *args, **kwargs):
   return params
 
 
+@overload
 def specifiable(
-    my_cls=None,
+    my_cls: None = None,
     /,
     *,
-    spec_type=None,
-    on_demand_init=True,
-    just_in_time_init=True):
+    spec_type: Optional[str] = None,
+    on_demand_init: bool = True,
+    just_in_time_init: bool = True) -> Callable[[T], T]:
+  ...
+
+
+@overload
+def specifiable(
+    my_cls: T,
+    /,
+    *,
+    spec_type: Optional[str] = None,
+    on_demand_init: bool = True,
+    just_in_time_init: bool = True) -> T:
+  ...
+
+
+def specifiable(
+    my_cls: Optional[T] = None,
+    /,
+    *,
+    spec_type: Optional[str] = None,
+    on_demand_init: bool = True,
+    just_in_time_init: bool = True) -> Union[T, Callable[[T], T]]:
   """A decorator that turns a class into a `Specifiable` subclass by
   implementing the `Specifiable` protocol.
 
@@ -285,8 +309,8 @@ class Bar():
       original `__init__` method will be called when the first time an attribute
       is accessed.
   """
-  def _wrapper(cls):
-    def new_init(self: Specifiable, *args, **kwargs):
+  def _wrapper(cls: T) -> T:
+    def new_init(self, *args, **kwargs):
       self._initialized = False
       self._in_init = False
 
@@ -361,9 +385,24 @@ def new_getattr(self, name):
     # start of the function body of _wrapper
     _register(cls, spec_type)
 
+    # register the original class as a virtual subclass of Specifiable
+    # so issubclass(cls, Specifiable) and isinstance(cls(), Specifiable) are
+    # true
+    Specifiable.register(cls)
+
     class_name = cls.__name__
-    original_init = cls.__init__
-    cls.__init__ = new_init
+
+    # if a class has `<class_name>__original_init` function, it means it has
+    # already been registered as specifiable. In this case, we skip the
+    # following modification.
+    if hasattr(cls, cls.__name__ + '__original_init'):
+      return cls
+
+    original_init = cls.__init__  # type: ignore[misc]
+    # saved the class init function in a name that won't conflict with its
+    # parents.
+    setattr(cls, cls.__name__ + '__original_init', cls.__init__)  # type: ignore[misc]
+    cls.__init__ = new_init  # type: ignore[misc]
     if just_in_time_init:
       cls.__getattr__ = new_getattr
 
@@ -386,3 +425,16 @@ def new_getattr(self, name):
   # When this decorator is called without an argument, i.e. "@specifiable",
   # we return the augmented class.
   return _wrapper(my_cls)
+
+
+def unspecifiable(cls) -> None:
+  cls.__init__ = getattr(cls, cls.__name__ + '__original_init')
+  # cls.__getattr__ = None
+  delattr(cls, '__getattr__')
+  delattr(cls, cls.__name__ + '__original_init')
+  delattr(cls, 'spec_type')
+  delattr(cls, 'run_original_init')
+  delattr(cls, 'to_spec')
+  delattr(cls, '_to_spec_helper')
+  delattr(cls, 'from_spec')
+  delattr(cls, '_from_spec_helper')

sdks/python/apache_beam/ml/anomaly/specifiable_test.py

@@ -30,6 +30,7 @@
 from apache_beam.ml.anomaly.specifiable import Spec
 from apache_beam.ml.anomaly.specifiable import Specifiable
 from apache_beam.ml.anomaly.specifiable import specifiable
+from apache_beam.ml.anomaly.specifiable import unspecifiable
 
 
 class TestSpecifiable(unittest.TestCase):
@@ -585,6 +586,32 @@ def apply(self, x, y):
     self.assertEqual(w_2.run_func_in_class(5, 3), 150)
 
 
+class TestUncommonUsages(unittest.TestCase):
+  def test_double_specifiable(self):
+    @specifiable
+    @specifiable
+    class ZZ():
+      def __init__(self, a):
+        self.a = a
+
+    c = ZZ("b")
+    c.run_original_init()
+    self.assertEqual(c.a, "b")
+
+  def test_unspecifiable(self):
+    class YY():
+      def __init__(self, x):
+        self.x = x
+        assert False
+
+    YY = specifiable(YY)
+    y = YY(1)
+    self.assertRaises(AssertionError, lambda: y.x)
+
+    unspecifiable(YY)
+    self.assertRaises(AssertionError, YY, 1)
+
+
 if __name__ == '__main__':
   logging.getLogger().setLevel(logging.INFO)
   unittest.main()