filterfw/core/Filter.java

/*
 * Copyright (C) 2011 The Android Open Source Project
 *
 * Licensed 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.
 */


package android.filterfw.core;

import android.filterfw.core.FilterContext;
import android.filterfw.core.FilterPort;
import android.filterfw.core.KeyValueMap;
import android.filterfw.io.TextGraphReader;
import android.filterfw.io.GraphIOException;
import android.filterfw.format.ObjectFormat;
import android.util.Log;

import java.io.Serializable;
import java.lang.annotation.Annotation;
import java.lang.reflect.Field;
import java.lang.Thread;
import java.util.Collection;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Map.Entry;
import java.util.LinkedList;
import java.util.Set;

/**
 * @hide
 */
public abstract class Filter {

    static final int STATUS_PREINIT               = 0;
    static final int STATUS_UNPREPARED            = 1;
    static final int STATUS_PREPARED              = 2;
    static final int STATUS_PROCESSING            = 3;
    static final int STATUS_SLEEPING              = 4;
    static final int STATUS_FINISHED              = 5;
    static final int STATUS_ERROR                 = 6;
    static final int STATUS_RELEASED              = 7;

    private String mName;

    private int mInputCount = -1;
    private int mOutputCount = -1;

    private HashMap<String, InputPort> mInputPorts;
    private HashMap<String, OutputPort> mOutputPorts;

    private HashSet<Frame> mFramesToRelease;
    private HashMap<String, Frame> mFramesToSet;

    private int mStatus = 0;
    private boolean mIsOpen = false;
    private int mSleepDelay;

    private long mCurrentTimestamp;

    private boolean mLogVerbose;
    private static final String TAG = "Filter";

    public Filter(String name) {
        mName = name;
        mFramesToRelease = new HashSet<Frame>();
        mFramesToSet = new HashMap<String, Frame>();
        mStatus = STATUS_PREINIT;

        mLogVerbose = Log.isLoggable(TAG, Log.VERBOSE);
    }

    /** Tests to see if a given filter is installed on the system. Requires
     * full filter package name, including filterpack.
     */
    public static final boolean isAvailable(String filterName) {
        ClassLoader contextClassLoader = Thread.currentThread().getContextClassLoader();
        Class filterClass;
        // First see if a class of that name exists
        try {
            filterClass = contextClassLoader.loadClass(filterName);
        } catch (ClassNotFoundException e) {
            return false;
        }
        // Then make sure it's a subclass of Filter.
        try {
            filterClass.asSubclass(Filter.class);
        } catch (ClassCastException e) {
            return false;
        }
        return true;
    }

    public final void initWithValueMap(KeyValueMap valueMap) {
        // Initialization
        initFinalPorts(valueMap);

        // Setup remaining ports
        initRemainingPorts(valueMap);

        // This indicates that final ports can no longer be set
        mStatus = STATUS_UNPREPARED;
    }

    public final void initWithAssignmentString(String assignments) {
        try {
            KeyValueMap valueMap = new TextGraphReader().readKeyValueAssignments(assignments);
            initWithValueMap(valueMap);
        } catch (GraphIOException e) {
            throw new IllegalArgumentException(e.getMessage());
        }
    }

    public final void initWithAssignmentList(Object... keyValues) {
        KeyValueMap valueMap = new KeyValueMap();
        valueMap.setKeyValues(keyValues);
        initWithValueMap(valueMap);
    }

    public final void init() throws ProtocolException {
        KeyValueMap valueMap = new KeyValueMap();
        initWithValueMap(valueMap);
    }

    public String getFilterClassName() {
        return getClass().getSimpleName();
    }

    public final String getName() {
        return mName;
    }

    public boolean isOpen() {
        return mIsOpen;
    }

    public void setInputFrame(String inputName, Frame frame) {
        FilterPort port = getInputPort(inputName);
        if (!port.isOpen()) {
            port.open();
        }
        port.setFrame(frame);
    }

    public final void setInputValue(String inputName, Object value) {
        setInputFrame(inputName, wrapInputValue(inputName, value));
    }

    protected void prepare(FilterContext context) {
    }

    protected void parametersUpdated(Set<String> updated) {
    }

    protected void delayNextProcess(int millisecs) {
        mSleepDelay = millisecs;
        mStatus = STATUS_SLEEPING;
    }

    public abstract void setupPorts();

    public FrameFormat getOutputFormat(String portName, FrameFormat inputFormat) {
        return null;
    }

    public final FrameFormat getInputFormat(String portName) {
        InputPort inputPort = getInputPort(portName);
        return inputPort.getSourceFormat();
    }

    public void open(FilterContext context) {
    }

    public abstract void process(FilterContext context);

    public final int getSleepDelay() {
        return 250;
    }

    public void close(FilterContext context) {
    }

    public void tearDown(FilterContext context) {
    }

    public final int getNumberOfConnectedInputs() {
        int c = 0;
        for (InputPort inputPort : mInputPorts.values()) {
            if (inputPort.isConnected()) {
                ++c;
            }
        }
        return c;
    }

    public final int getNumberOfConnectedOutputs() {
        int c = 0;
        for (OutputPort outputPort : mOutputPorts.values()) {
            if (outputPort.isConnected()) {
                ++c;
            }
        }
        return c;
    }

    public final int getNumberOfInputs() {
        return mOutputPorts == null ? 0 : mInputPorts.size();
    }

    public final int getNumberOfOutputs() {
        return mInputPorts == null ? 0 : mOutputPorts.size();
    }

    public final InputPort getInputPort(String portName) {
        if (mInputPorts == null) {
            throw new NullPointerException("Attempting to access input port '" + portName
                + "' of " + this + " before Filter has been initialized!");
        }
        InputPort result = mInputPorts.get(portName);
        if (result == null) {
            throw new IllegalArgumentException("Unknown input port '" + portName + "' on filter "
                + this + "!");
        }
        return result;
    }

    public final OutputPort getOutputPort(String portName) {
        if (mInputPorts == null) {
            throw new NullPointerException("Attempting to access output port '" + portName
                + "' of " + this + " before Filter has been initialized!");
        }
        OutputPort result = mOutputPorts.get(portName);
        if (result == null) {
            throw new IllegalArgumentException("Unknown output port '" + portName + "' on filter "
                + this + "!");
        }
        return result;
    }

    protected final void pushOutput(String name, Frame frame) {
        if (frame.getTimestamp() == Frame.TIMESTAMP_NOT_SET) {
            if (mLogVerbose) Log.v(TAG, "Default-setting output Frame timestamp on port " + name + " to " + mCurrentTimestamp);
            frame.setTimestamp(mCurrentTimestamp);
        }
        getOutputPort(name).pushFrame(frame);
    }

    protected final Frame pullInput(String name) {
        Frame result = getInputPort(name).pullFrame();
        if (mCurrentTimestamp == Frame.TIMESTAMP_UNKNOWN) {
            mCurrentTimestamp = result.getTimestamp();
            if (mLogVerbose) Log.v(TAG, "Default-setting current timestamp from input port " + name + " to " + mCurrentTimestamp);
        }
        // As result is retained, we add it to the release pool here
        mFramesToRelease.add(result);

        return result;
    }

    public void fieldPortValueUpdated(String name, FilterContext context) {
    }

    /**
     * Transfers any frame from an input port to its destination. This is useful to force a
     * transfer from a FieldPort or ProgramPort to its connected target (field or program variable).
     */
    protected void transferInputPortFrame(String name, FilterContext context) {
        getInputPort(name).transfer(context);
    }

    /**
     * Assigns all program variables to the ports they are connected to. Call this after
     * constructing a Program instance with attached ProgramPorts.
     */
    protected void initProgramInputs(Program program, FilterContext context) {
        if (program != null) {
            for (InputPort inputPort : mInputPorts.values()) {
                if (inputPort.getTarget() == program) {
                    inputPort.transfer(context);
                }
            }
        }
    }

    /**
     * Adds an input port to the filter. You should call this from within setupPorts, if your
     * filter has input ports. No type-checking is performed on the input. If you would like to
     * check against a type mask, use
     * {@link #addMaskedInputPort(String, FrameFormat) addMaskedInputPort} instead.
     *
     * @param name the name of the input port
     */
    protected void addInputPort(String name) {
        addMaskedInputPort(name, null);
    }

    /**
     * Adds an input port to the filter. You should call this from within setupPorts, if your
     * filter has input ports. When type-checking is performed, the input format is
     * checked against the provided format mask. An exception is thrown in case of a conflict.
     *
     * @param name the name of the input port
     * @param formatMask a format mask, which filters the allowable input types
     */
    protected void addMaskedInputPort(String name, FrameFormat formatMask) {
        InputPort port = new StreamPort(this, name);
        if (mLogVerbose) Log.v(TAG, "Filter " + this + " adding " + port);
        mInputPorts.put(name, port);
        port.setPortFormat(formatMask);
    }

    /**
     * Adds an output port to the filter with a fixed output format. You should call this from
     * within setupPorts, if your filter has output ports. You cannot use this method, if your
     * output format depends on the input format (e.g. in a pass-through filter). In this case, use
     * {@link #addOutputBasedOnInput(String, String) addOutputBasedOnInput} instead.
     *
     * @param name the name of the output port
     * @param format the fixed output format of this port
     */
    protected void addOutputPort(String name, FrameFormat format) {
        OutputPort port = new OutputPort(this, name);
        if (mLogVerbose) Log.v(TAG, "Filter " + this + " adding " + port);
        port.setPortFormat(format);
        mOutputPorts.put(name, port);
    }

    /**
     * Adds an output port to the filter. You should call this from within setupPorts, if your
     * filter has output ports. Using this method indicates that the output format for this
     * particular port, depends on the format of an input port. You MUST also override
     * {@link #getOutputFormat(String, FrameFormat) getOutputFormat} to specify what format your
     * filter will output for a given input. If the output format of your filter port does not
     * depend on the input, use {@link #addOutputPort(String, FrameFormat) addOutputPort} instead.
     *
     * @param outputName the name of the output port
     * @param inputName the name of the input port, that this output depends on
     */
    protected void addOutputBasedOnInput(String outputName, String inputName) {
        OutputPort port = new OutputPort(this, outputName);
        if (mLogVerbose) Log.v(TAG, "Filter " + this + " adding " + port);
        port.setBasePort(getInputPort(inputName));
        mOutputPorts.put(outputName, port);
    }

    protected void addFieldPort(String name,
                                Field field,
                                boolean hasDefault,
                                boolean isFinal) {
        // Make sure field is accessible
        field.setAccessible(true);

        // Create port for this input
        InputPort fieldPort = isFinal
            ? new FinalPort(this, name, field, hasDefault)
            : new FieldPort(this, name, field, hasDefault);

        // Create format for this input
        if (mLogVerbose) Log.v(TAG, "Filter " + this + " adding " + fieldPort);
        MutableFrameFormat format = ObjectFormat.fromClass(field.getType(),
                                                           FrameFormat.TARGET_SIMPLE);
        fieldPort.setPortFormat(format);

        // Add port
        mInputPorts.put(name, fieldPort);
    }

    protected void addProgramPort(String name,
                                  String varName,
                                  Field field,
                                  Class varType,
                                  boolean hasDefault) {
        // Make sure field is accessible
        field.setAccessible(true);

        // Create port for this input
        InputPort programPort = new ProgramPort(this, name, varName, field, hasDefault);

        // Create format for this input
        if (mLogVerbose) Log.v(TAG, "Filter " + this + " adding " + programPort);
        MutableFrameFormat format = ObjectFormat.fromClass(varType,
                                                           FrameFormat.TARGET_SIMPLE);
        programPort.setPortFormat(format);

        // Add port
        mInputPorts.put(name, programPort);
    }

    protected void closeOutputPort(String name) {
        getOutputPort(name).close();
    }

    /**
     * Specifies whether the filter should not be scheduled until a frame is available on that
     * input port. Note, that setting this to false, does not block a new frame from coming in
     * (though there is no necessity to pull that frame for processing).
     * @param portName the name of the input port.
     * @param waits true, if the filter should wait for a frame on this port.
     */
    protected void setWaitsOnInputPort(String portName, boolean waits) {
        getInputPort(portName).setBlocking(waits);
    }

    /**
     * Specifies whether the filter should not be scheduled until the output port is free, i.e.
     * there is no frame waiting on that output.
     * @param portName the name of the output port.
     * @param waits true, if the filter should wait for the port to become free.
     */
    protected void setWaitsOnOutputPort(String portName, boolean waits) {
        getOutputPort(portName).setBlocking(waits);
    }

    public String toString() {
        return "'" + getName() + "' (" + getFilterClassName() + ")";
    }

    // Core internal methods ///////////////////////////////////////////////////////////////////////
    final Collection<InputPort> getInputPorts() {
        return mInputPorts.values();
    }

    final Collection<OutputPort> getOutputPorts() {
        return mOutputPorts.values();
    }

    final synchronized int getStatus() {
        return mStatus;
    }

    final synchronized void unsetStatus(int flag) {
        mStatus &= ~flag;
    }

    final synchronized void performOpen(FilterContext context) {
        if (!mIsOpen) {
            if (mStatus == STATUS_UNPREPARED) {
                if (mLogVerbose) Log.v(TAG, "Preparing " + this);
                prepare(context);
                mStatus = STATUS_PREPARED;
            }
            if (mStatus == STATUS_PREPARED) {
                if (mLogVerbose) Log.v(TAG, "Opening " + this);
                open(context);
                mStatus = STATUS_PROCESSING;
            }
            if (mStatus != STATUS_PROCESSING) {
                throw new RuntimeException("Filter " + this + " was brought into invalid state during "
                    + "opening (state: " + mStatus + ")!");
            }
            mIsOpen = true;
        }
    }

    final synchronized void performProcess(FilterContext context) {
        if (mStatus == STATUS_RELEASED) {
            throw new RuntimeException("Filter " + this + " is already torn down!");
        }
        transferInputFrames(context);
        if (mStatus < STATUS_PROCESSING) {
            performOpen(context);
        }
        if (mLogVerbose) Log.v(TAG, "Processing " + this);
        mCurrentTimestamp = Frame.TIMESTAMP_UNKNOWN;
        process(context);
        releasePulledFrames(context);
        if (filterMustClose()) {
            performClose(context);
        }
    }

    final synchronized void performClose(FilterContext context) {
        if (mIsOpen) {
            if (mLogVerbose) Log.v(TAG, "Closing " + this);
            mIsOpen = false;
            mStatus = STATUS_PREPARED;
            close(context);
            closePorts();
        }
    }

    final synchronized void performTearDown(FilterContext context) {
        performClose(context);
        if (mStatus != STATUS_RELEASED) {
            tearDown(context);
            mStatus = STATUS_RELEASED;
        }
    }

    synchronized final boolean canProcess() {
        if (mLogVerbose) Log.v(TAG, "Checking if can process: " + this + " (" + mStatus + ").");
        if (mStatus <= STATUS_PROCESSING) {
            return inputConditionsMet() && outputConditionsMet();
        } else {
            return false;
        }
    }

    final void openOutputs() {
        if (mLogVerbose) Log.v(TAG, "Opening all output ports on " + this + "!");
        for (OutputPort outputPort : mOutputPorts.values()) {
            if (!outputPort.isOpen()) {
                outputPort.open();
            }
        }
    }

    final void clearInputs() {
        for (InputPort inputPort : mInputPorts.values()) {
            inputPort.clear();
        }
    }

    final void clearOutputs() {
        for (OutputPort outputPort : mOutputPorts.values()) {
            outputPort.clear();
        }
    }

    final void notifyFieldPortValueUpdated(String name, FilterContext context) {
        if (mStatus == STATUS_PROCESSING || mStatus == STATUS_PREPARED) {
            fieldPortValueUpdated(name, context);
        }
    }

    final synchronized void pushInputFrame(String inputName, Frame frame) {
        FilterPort port = getInputPort(inputName);
        if (!port.isOpen()) {
            port.open();
        }
        port.pushFrame(frame);
    }

    final synchronized void pushInputValue(String inputName, Object value) {
        pushInputFrame(inputName, wrapInputValue(inputName, value));
    }

    // Filter internal methods /////////////////////////////////////////////////////////////////////
    private final void initFinalPorts(KeyValueMap values) {
        mInputPorts = new HashMap<String, InputPort>();
        mOutputPorts = new HashMap<String, OutputPort>();
        addAndSetFinalPorts(values);
    }

    private final void initRemainingPorts(KeyValueMap values) {
        addAnnotatedPorts();
        setupPorts();   // TODO: rename to addFilterPorts() ?
        setInitialInputValues(values);
    }

    private final void addAndSetFinalPorts(KeyValueMap values) {
        Class filterClass = getClass();
        Annotation annotation;
        for (Field field : filterClass.getDeclaredFields()) {
            if ((annotation = field.getAnnotation(GenerateFinalPort.class)) != null) {
                GenerateFinalPort generator = (GenerateFinalPort)annotation;
                String name = generator.name().isEmpty() ? field.getName() : generator.name();
                boolean hasDefault = generator.hasDefault();
                addFieldPort(name, field, hasDefault, true);
                if (values.containsKey(name)) {
                    setImmediateInputValue(name, values.get(name));
                    values.remove(name);
                } else if (!generator.hasDefault()) {
                    throw new RuntimeException("No value specified for final input port '"
                        + name + "' of filter " + this + "!");
                }
            }
        }
    }

    private final void addAnnotatedPorts() {
        Class filterClass = getClass();
        Annotation annotation;
        for (Field field : filterClass.getDeclaredFields()) {
            if ((annotation = field.getAnnotation(GenerateFieldPort.class)) != null) {
                GenerateFieldPort generator = (GenerateFieldPort)annotation;
                addFieldGenerator(generator, field);
            } else if ((annotation = field.getAnnotation(GenerateProgramPort.class)) != null) {
                GenerateProgramPort generator = (GenerateProgramPort)annotation;
                addProgramGenerator(generator, field);
            } else if ((annotation = field.getAnnotation(GenerateProgramPorts.class)) != null) {
                GenerateProgramPorts generators = (GenerateProgramPorts)annotation;
                for (GenerateProgramPort generator : generators.value()) {
                    addProgramGenerator(generator, field);
                }
            }
        }
    }

    private final void addFieldGenerator(GenerateFieldPort generator, Field field) {
        String name = generator.name().isEmpty() ? field.getName() : generator.name();
        boolean hasDefault = generator.hasDefault();
        addFieldPort(name, field, hasDefault, false);
    }

    private final void addProgramGenerator(GenerateProgramPort generator, Field field) {
        String name = generator.name();
        String varName = generator.variableName().isEmpty() ? name
                                                            : generator.variableName();
        Class varType = generator.type();
        boolean hasDefault = generator.hasDefault();
        addProgramPort(name, varName, field, varType, hasDefault);
    }

    private final void setInitialInputValues(KeyValueMap values) {
        for (Entry<String, Object> entry : values.entrySet()) {
            setInputValue(entry.getKey(), entry.getValue());
        }
    }

    private final void setImmediateInputValue(String name, Object value) {
        if (mLogVerbose) Log.v(TAG, "Setting immediate value " + value + " for port " + name + "!");
        FilterPort port = getInputPort(name);
        port.open();
        port.setFrame(SimpleFrame.wrapObject(value, null));
    }

    private final void transferInputFrames(FilterContext context) {
        for (InputPort inputPort : mInputPorts.values()) {
            inputPort.transfer(context);
        }
    }

    private final Frame wrapInputValue(String inputName, Object value) {
        MutableFrameFormat inputFormat = ObjectFormat.fromObject(value, FrameFormat.TARGET_SIMPLE);
        if (value == null) {
            // If the value is null, the format cannot guess the class, so we adjust it to the
            // class of the input port here
            FrameFormat portFormat = getInputPort(inputName).getPortFormat();
            Class portClass = (portFormat == null) ? null : portFormat.getObjectClass();
            inputFormat.setObjectClass(portClass);
        }

        // Serialize if serializable, and type is not an immutable primitive.
        boolean shouldSerialize = !(value instanceof Number)
            && !(value instanceof Boolean)
            && !(value instanceof String)
            && value instanceof Serializable;

        // Create frame wrapper
        Frame frame = shouldSerialize
            ? new SerializedFrame(inputFormat, null)
            : new SimpleFrame(inputFormat, null);
        frame.setObjectValue(value);
        return frame;
    }

    private final void releasePulledFrames(FilterContext context) {
        for (Frame frame : mFramesToRelease) {
            context.getFrameManager().releaseFrame(frame);
        }
        mFramesToRelease.clear();
    }

    private final boolean inputConditionsMet() {
        for (FilterPort port : mInputPorts.values()) {
            if (!port.isReady()) {
                if (mLogVerbose) Log.v(TAG, "Input condition not met: " + port + "!");
                return false;
            }
        }
        return true;
    }

    private final boolean outputConditionsMet() {
        for (FilterPort port : mOutputPorts.values()) {
            if (!port.isReady()) {
                if (mLogVerbose) Log.v(TAG, "Output condition not met: " + port + "!");
                return false;
            }
        }
        return true;
    }

    private final void closePorts() {
        if (mLogVerbose) Log.v(TAG, "Closing all ports on " + this + "!");
        for (InputPort inputPort : mInputPorts.values()) {
            inputPort.close();
        }
        for (OutputPort outputPort : mOutputPorts.values()) {
            outputPort.close();
        }
    }

    private final boolean filterMustClose() {
        for (InputPort inputPort : mInputPorts.values()) {
            if (inputPort.filterMustClose()) {
                if (mLogVerbose) Log.v(TAG, "Filter " + this + " must close due to port " + inputPort);
                return true;
            }
        }
        for (OutputPort outputPort : mOutputPorts.values()) {
            if (outputPort.filterMustClose()) {
                if (mLogVerbose) Log.v(TAG, "Filter " + this + " must close due to port " + outputPort);
                return true;
            }
        }
        return false;
    }
}