mlm_insights.core.post_processors package

Subpackages

Submodules

mlm_insights.core.post_processors.local_writer_post_processor module

class mlm_insights.core.post_processors.local_writer_post_processor.LocalWriterPostProcessor(file_name: str, file_location: str, is_critical: bool = False)

Bases: PostProcessor

Local writer Post Processor. This will store the profile in user provided local file location.
Output will be in form of Profile object in serialized format, this object can be deserialized back using profile unmarshall method.
User need to pass valid file location for this post processor

Configuration

file_name: str

  • User provided file name for the profile which needs to be written by post processor

file_location: str

  • User provided file location where this profile needs to be written by post processor

is_critical: bool

  • An optional boolean to mark the run as failed in case of any exception in the post processor run. By default the flag is set to false.

Sample code

def test_local_profile_writer():
    profile = Profile()
    feature_metadata = FeatureMetadata(feature_name='age',
                                       feature_type=FeatureType(DataType.FLOAT, VariableType.CONTINUOUS))
    feature = Feature(feature_metadata)
    feature.add_metric(MetricMetadata(klass=Sum))
    profile.add_feature(feature)
    input_data = pd.DataFrame({'age': [1, 2, 3, 4]})
    profile.compute(input_data, input_schema_for_test)

    postProcessorRequest  = PostProcessorRequest(profile)
    file_path = get_root_path() + "/" + "data"
    file_name = "profile"
    localWriterPostProcessor = LocalWriterPostProcessor(file_name, file_path)
    result = localWriterPostProcessor.process(postProcessorRequest)

For reading from file_path which is a string
    with open(file_path, "rb") as file:
        data: Profile = file.read()

    recreated_profile = Profile.unmarshall(profile_with_feature.marshall())
process(postProcessorRequest: PostProcessorRequest, **kwargs: Any) → PostProcessorResult

Fetch the Profile and store it in object store bucket

Parameters

postProcessorRequest: PostProcessorRequest

Returns

bool : Status of storing the Profile in local file location

This returns True when the Profile is successfully uploaded in local file location and False otherwise.

mlm_insights.core.post_processors.object_storage_writer_post_processor module

class mlm_insights.core.post_processors.object_storage_writer_post_processor.ObjectStorageWriterPostProcessor(bucket_name: str, object_name: str, prefix: str, namespace: str, storage_options: Dict[str, Any] = {}, is_critical: bool = False)

Bases: PostProcessor

A class for interacting with Oracle Cloud Infrastructure Object Storage as a post-processor.
Object Storage Writer Post Processor. This will store the profile in user provided Object storage location.
Output will be in form of Profile object in serialized format, this object can be deserialized back using profile unmarshall method.

Configuration

namespace: str

  • Namespace of tenancy

bucket_name: str

  • User configured Bucket name where this profile needs to be written by post processor

object_name: str

  • user provided object name

prefix: str

  • user provided object prefix

storage_options’: Dict[str, Any]

  • {“config”: “~/.oci/config”} to authenticate the file systems

is_critical: bool

  • An optional boolean to mark the run as failed in case of any exception in the post processor run. By default the flag is set to false.

Sample code

def test_object_storage_postprocessor(mocked_auth):
    profile = Profile()
    feature_metadata = FeatureMetadata(feature_name='age',
                                       feature_type=FeatureType(DataType.FLOAT, VariableType.CONTINUOUS))
    feature = Feature(feature_metadata)
    feature.add_metric(MetricMetadata(klass=Sum))
    profile.add_feature(feature)
    input_data = pd.DataFrame({'age': [1, 2, 3, 4]})
    profile.compute(input_data, input_schema_for_test)

    objectStorageWriterPostProcessor = ObjectStorageWriterPostProcessor("root", "bucket", "prefix", "namespace")
    post_processor_request =  PostProcessorRequest(profile)
    result = objectStorageWriterPostProcessor.process(post_processor_request)


For reading using some datasource
    data_source_args = {
        'bucket_name': bucket_name,
        'namespace': namespace,
        'object_prefix': object_prefix,
        'storage_options' : {"config": "~/.oci/config"} to authenticate the file systems
    }
    file_location = 'oci://%s@%s/%s' % (bucket_name, namespace, object_prefix)
    fs: ocifs.OCIFileSystem = ocifs.OCIFileSystem(**data_csv)
    assert fs.exists(file_location)

    profile_file = fs.oci_client.get_object(self, namespace_name, bucket_name, object_name, **kwargs)

    # read data from profile file and unmarshall it
    with open(profile_file, "rb") as file:
        data: Profile = file.read()

    recreated_profile = Profile.unmarshall(profile_with_feature.marshall())
process(postProcessorRequest: PostProcessorRequest, **kwargs: Any) → PostProcessorResult

Fetches the Profile from PostProcessorRequest and stores it in Object Storage using default or user provided authentication policy

Parameters

postProcessorRequest: PostProcessorRequest

Returns

bool : Status of storing the Profile in Object Storage

This returns True when the Profile is successfully uploaded in Object Storage and False otherwise.

mlm_insights.core.post_processors.post_processor_request module

class mlm_insights.core.post_processors.post_processor_request.PostProcessorRequest(profile: Profile)

Bases: ABC

Class for defining Post Processor Request needed for Insights PostProcessor Module.This will have getter’s and setter’s method for reference and prediction profile

property profile: Profile

Get the profile which is set during initialization.

Returns

Profile

An Instance of Profile.

mlm_insights.core.post_processors.post_processor_result module

class mlm_insights.core.post_processors.post_processor_result.PostProcessorResult(status: PostProcessorResultStatus, message: str, is_critical: bool = False)

Bases: ABC

Class for capturing Post Processor response .This will post processor run status and message .