MC_MocapStreamImpl.h

Go to the documentation of this file.

/*
 * PROPRIETARY INFORMATION.  This software is proprietary to
 * Side Effects Software Inc., and is not to be reproduced,
 * transmitted, or disclosed in any way without written permission.
 *
 * NAME:        MC_MocapStreamImpl.h (C++)
 *
 */
#ifndef __MC_MOCAPSTREAMIMPL_H_INCLUDED__
#define __MC_MOCAPSTREAMIMPL_H_INCLUDED__

#include "MC_API.h"
#include "MC_MocapStreamLog.h"

#include <GU/GU_Detail.h>
#include <GU/GU_MotionClip.h>
#include <UT/UT_Array.h>
#include <UT/UT_ConcurrentQueue.h>
#include <UT/UT_Function.h>
#include <UT/UT_StopWatch.h>
#include <UT/UT_WorkBuffer.h>

#include <hboost/system/system_error.hpp>

#include <atomic>
#include <vector>

const UT_StringHolder theActorAttribName = "actorname";

extern "C" {
    /// Function called in DSO to install new functions
    /// All files found in $HOUDINI_MOCAPSTREAM_DSO_PATH are automatically
    /// scanned for this function.
    SYS_VISIBILITY_EXPORT extern void newMocapStreamDevice(void *);
}

/// A structure for passing the values of the MocapStream SOPs parameters
/// and node path to the MocapStream devices.
/// MC_MocapStreamImpl::updateJoints(...) receives the current value of
/// these parameters as well as the value of these parameters the last
/// time the joints were updated for that node.
struct MC_API MC_MocapStreamCookParms
{
    UT_StringHolder myNodePath;
    UT_StringHolder myActorPattern;
    UT_StringHolder myConfigAttrib;
    UT_StringHolder myFacialAttribs;
    bool            myDoConvertUnits = false;
    bool            myDoConvertUpAxis = false;
};

/// A structure used to store the MotionClip geometry that is created 
/// when recording a motion capture stream.
/// This structure also has its own MC_MocapStreamCookparms which is a
/// snapshot of the value of those parms on the MocapStream SOP from when
/// the recording started.
struct MC_API MC_MocapRecorder
{
    GU_Detail myEvalTarget;
    GU_Detail myClipGeo;
    UT_UniquePtr<GU_MotionClipRW> myClip;

    MC_MocapStreamCookParms myStreamCookParms;
    
    UT_WorkBuffer myErrorMsg;
    bool myHasStartTime = false;
    fpreal64 myStartTime = 0.0f;
    uint64 mySkeletonVersion = 0;
};

class MC_API MC_MocapStreamImpl
{
public:
    // we want to access connect methods in the manager but not
    // in the implementations
    friend class MC_MocapStreamManager;


    class RegisterBase
    {
    public:
        virtual UT_UniquePtr<MC_MocapStreamImpl> getInstance() = 0;
        virtual ~RegisterBase() {};
    };

    /// Create a static instance of this to register an implementation
    ///
    /// The first time a SOP_MocapStream is initialised it will solidify the
    /// list of implementations and ignore any more registrations/unregistrations
    template <typename T>
    class Register : public RegisterBase
    {
    public:
        Register()
        {
            MC_MocapStreamImpl::registerDevice(this);
        }
        ~Register() override
        {
            MC_MocapStreamImpl::unregisterDevice(this);
        }
        UT_UniquePtr<MC_MocapStreamImpl> getInstance() override
        {
            return UTmakeUnique<T>();
        }
    };


    /// Keeps track of the state of a mocap stream server and is used as a
    /// method of communcating between a server function and the base class.
    ///
    /// For thread safety, the server function should only call setConnected(),
    /// setPortOpened(), setError(), gotPacket(), and killed(). While the server
    /// function is running, it should be the only thread calling these methods.
    struct ServerState {
        UT_ConcurrentQueue<std::pair<UT_StringHolder, fpreal64>> myPackets;
        UT_ConcurrentQueue<UT_Function<void(void*)>> myTasks;
        
        ServerState() : myConnected(false), myError(false),
                        myKilled(false) { }

        /// Whether a connection is currently established.
        bool connected() { return myConnected; }
        void setConnected(bool connected) { myConnected = connected; }

        /// Whether the connection has been killed.
        bool killed() { return myKilled; }
        void setKilled(bool killed) { myKilled = killed; }

        /// Whether the port for receiving data has been opened.
        /// setPortOpened(True) must be called for the data to begin parsing.
        bool portOpened() { return myPortOpened; }
        void setPortOpened(bool port_opened) { myPortOpened = port_opened; }
        
        /// Sends an error message from the server function to be added to the
        /// log.
        void setError(const char* msg)
        {
            if (!myError)
            {
                myErrorMsg = msg;
                myError = true;
            }
        }

        /// Reads the error message if one has been added.
        bool getError(UT_StringHolder& res) const
        {
            if (myError)
            {
                res = myErrorMsg;
                return true;
            }
            return false;
        }

        /// Resets the error state.
        void resetError()
        {
            myError = false;
        }

        /// Resets the server state to its default values.
        void reset()
        {
            myConnected = false; 
            myKilled = false;
            myPortOpened = false;
            myError = false;
            myTasks.clear();
            myPackets.clear();
            myTimer.start();
        }

        /// Called in the server function to pass a packet along to be parsed
        /// once it has been received.
        void gotPacket(UT_StringHolder&& packet)
        {
            if (myPackets.unsafe_size() > 250)
            {
                myPackets.clear();
            }
            myPackets.push(std::make_pair<UT_StringHolder, fpreal64>(
                std::move(packet),
                myTimer.getTime()
            ));
        }

    private:
        std::atomic<bool> myConnected;
        std::atomic<bool> myKilled;
        std::atomic<bool> myPortOpened;
        std::atomic<bool> myError; 
        UT_StringHolder myErrorMsg;
        UT_StopWatch myTimer;
    };
    
    /// Keeps track of the options used to connect to a server and defines which
    /// parameters should be enabled for the device on the MocapStream SOP.
    struct ServerOptions
    {
        // These are the value for the corresponding parameters on the Mocap
        // Stream SOP. When a subclass of MC_MocapStreamImpl defines the values
        // for these variables in its constructor, that defines the default
        // values for the corresponding parameters for that device.
        int myPort;
        UT_StringHolder myLocalIP;
        UT_StringHolder myHostIP;
        UT_StringHolder myMulticastIP;
        int myDataPort;
        int myCommandPort;
        int myBufferSize;

        // This structure defines which parameters are made available for any
        // mocap stream device defined by a subclass of MC_MocapStreamImpl.
        struct Enabled
        {
            bool localPort;
            bool localIP;
            bool hostIP;
            bool multicastIP;
            bool dataPort;
            bool commandPort;
            bool facialAttribs;
            bool configAttrib;
            bool doConvertUnits;
            bool doConvertUpAxis;
        } myEnabled;
        
        bool isSameConnection(const ServerOptions& other) const
        {
            return myPort == other.myPort
                   && myHostIP == other.myHostIP
                   && myMulticastIP == other.myMulticastIP
                   && myDataPort == other.myDataPort
                   && myCommandPort == other.myCommandPort
                   && myBufferSize == other.myBufferSize;
        }
    };

    typedef void (*MC_MocapServerFunc)(ServerState& state, const ServerOptions& options,
                                    MC_MocapStreamImpl *data);

    /// A function for establishing a basic UDP connection to a server,
    /// receiving motion capture data, and passing it along to be parsed.
    static void basicUdpReceiver(ServerState& state, const ServerOptions& options,
                                 MC_MocapStreamImpl *data);

    /// A function for establishing a basic TPC connection, receiving motion
    /// capture data, and passing it along to be parsed.
    static void basicTcpClient(ServerState& state, const ServerOptions& options,
                               MC_MocapStreamImpl *data);

    /// Gets the error message corresponding to an hboost error code.
    static void getBoostErrorMessage(hboost::system::error_code e,
                                     UT_StringHolder &err_msg);
    /// Gets the error message corresponding to an hboost error code.
    static void getBoostErrorMessage(hboost::system::system_error& e,
                                     UT_StringHolder &err_msg);
    /// Adds an error message to the log corresponding to an hboost error code.
    static void handleBoostError(hboost::system::error_code e,
                                 ServerState& state);
    /// Adds an error message to the log corresponding to an hboost error code.
    static void handleBoostError(hboost::system::system_error& e,
                                 ServerState& state);


    MC_MocapStreamImpl(MC_MocapServerFunc f, const ServerOptions& myServerOptions);
    virtual ~MC_MocapStreamImpl();
    
    virtual UT_UniquePtr<MC_MocapStreamImpl> clone(const ServerOptions& opts) = 0;

    /// Called for each packet received from the server.
    /// Return true when a packet which defines a pose is valid.
    /// parse_incomplete is set to true when a portion of the packet
    /// remains to be parsed. In this case, packet is set to the appropriate
    /// value to call this method again.
    ///
    /// NOTE: If this returns true and parse_incomplete is set to true, that means
    /// that the "packet" that was passed in contained parts of more that one packet.
    /// This can occur when the data is from a constant stream that has been cut up.
    virtual bool parsePacket(UT_StringHolder &packet,
                             fpreal packet_time,
                             bool &parse_incomplete) = 0;

    /// Appends the skeleton's geometry to the gdp
    /// The theActorAttribName attribute should store the
    /// name of the skeleton, and the name attribute will store the names
    /// of each joint. 
    virtual void buildSkeleton(GU_Detail*, UT_StringHolder name) = 0;

    /// Removes the stored data for a skeleton.
    virtual void removeSkeleton(const UT_StringHolder& name);

    // Indicates whether a skeleton that has expired should be removed
    virtual bool removeExpiredSkeletons() const { return true; }
    
    /// Each time we sucessfully parse a packet we increment a counter,
    /// if receivedSkeleton has not been called the last 'expirationCount'
    /// of successful parses then we will remove the skeleton.
    /// Return -1 to disable this.
    virtual int expirationCount();

    /// Updates the joints' attributes in place based on the results of parsePacket
    /// you can assume that all desired skeletons have already been built in the gdp,
    /// so you should keep the same geometry, just update the attributes you need.
    virtual void updateJoints(GU_Detail *gdp,
                              const MC_MocapStreamCookParms &cookparms,
                              const MC_MocapStreamCookParms &prev_cookparms) = 0;
    
    /// Should be called when an actor's data has been parsed.
    /// The base class will keep track of the actors and and call
    /// build/removeSkeleton when needed.
    void receivedSkeleton(const UT_StringHolder& name);

    /// Call this to let the base class know an actor's name has changed
    void renamedSkeleton(const UT_StringHolder& old_name, const UT_StringHolder& new_name);

    /// Call this in parsePacket to specify the time in seconds of the
    /// received packet. Otherwise it will default to the time it was received
    /// This is used for timing in the recorded MotionClip.
    void setPacketTime(fpreal64 time);
    
    /// Call this to report an error when parsing.
    template <typename... Args>
    void addParseError(const char* fmt, Args&&... args)
    {
        MC_MocapStream_Log(UT_ERROR_WARNING, "Parse Error", fmt, std::forward<Args>(args)...);
        myHitError = true;
    }

    /// Call this to report a warning when parsing.
    template <typename... Args>
    void addParseWarning(const char* fmt, Args&&... args)
    {
        MC_MocapStream_Log(UT_ERROR_WARNING, "Parse Warning", fmt, std::forward<Args>(args)...);
    }
    
    // Call this to trigger a rebuild of the skeleton geometries.
    void rebuildSkeletons() { mySkeletonVersion++; }
    
    /// Used by the main thread to queue a task for the server.
    /// The server function will access these tasks via state.myTasks
    /// so it can call them. Ensure your task functions
    /// accept the same data type your server passes into them.
    ///
    /// Your functions should not bind any data that would be unsafe
    /// to access within the server thread.
    void queueServerTask(const UT_Function<void(void*)> &task);
    
    /// The unique name used to identify a subclass's mocap stream device.
    virtual const char* name() const = 0;

    /// The label for a subclass's mocap stream device which appears in the "Device"
    /// menu on the MocapStream SOP.
    virtual const char* label() const = 0;

    /// The main entry point into MocapStreamImpl for creating or updating
    /// a SOP's skeleton geometry.
    /// Returns true if it has successfully parsed new data.
    bool cook(GU_Detail *gdp, uint64 &skeleton_version, fpreal time,
              const MC_MocapStreamCookParms &cookparms,
              const MC_MocapStreamCookParms &prev_cookparms,
              bool &is_killed);
    
    /// Removes all actors/skeletons.
    void clear();

    /// Starts recording the motion capture data and stores it in a MotionClip.
    void startRecording(UT_SharedPtr<MC_MocapRecorder> recorder,
                        const MC_MocapStreamCookParms &cookparms);

    /// Stops recording the motion capture data.
    void stopRecording(UT_SharedPtr<MC_MocapRecorder> recorder);
    
    // Maps each actor name to the number of packets processed
    // since it was last received.
    const UT_ArrayStringMap<exint>& getActorMap();
    
    /// Used to determine if the stream has run into an error.
    bool hasHitError();
protected:
    
    /// Call this from updateJoints to rename any of the attributes for storing
    /// the parsed blendshapes.
    UT_StringHolder updateAttribName(GU_Detail *gdp,
                                     const UT_StringHolder &base_name,
                                     const UT_StringHolder &rename_pattern,
                                     const UT_StringHolder &prev_rename_pattern);

private:

    static void registerDevice(RegisterBase*);
    static void unregisterDevice(RegisterBase*);

    static void* threadBody(void* data);

    bool connect();
    void disconnect();
    void endServerThread();

    void updateOutput(GU_Detail *gdp,
                      uint64 &skeleton_ver,
                      const MC_MocapStreamCookParms &cookparms,
                      const MC_MocapStreamCookParms &prev_cookparms,
                      const UT_ArrayStringMap<exint> &old_actors);

    // maps each actor name to the number of packets processed
    // since it was last received.
    UT_ArrayStringMap<exint> myActors;
    uint64 mySkeletonVersion;

    bool myHitError;

    const MC_MocapServerFunc myServerFunc;
    const ServerOptions myServerOptions;
    ServerState myServerState;
    unsigned long long myParseCount; 
    
    UT_Array<UT_WeakPtr<MC_MocapRecorder>> myRecorders;
    fpreal64 myPacketTime;

    UT_UniquePtr<UT_Thread> myServerThread;
    UT_Array<UT_Function<void(void*)>> myTasks;
};

#endif