lib/hmmmix/trellis/__init__.py
def pack_observations(states: numpy.typing.NDArray[numpy.int64],
                      weighted_obs_by_state: typing.Dict[numpy.int64, typing.Sequence[WeightedObservation]]):
    """
    pack weighted observations into C friendly compressed-sparse format that can
    be iterated over without hashing or attribute access

    :param states: list of state indices
    :param outgoing_observations: lists of WeightedObservation keyed by state index
    :return: (PackedObservations
    """

    n_states = len(states)
    n = sum(len(weighted_obs_by_state[s]) for s in states)

    start_indices = numpy.zeros(shape=(n_states, ), dtype=int)
    end_indices = numpy.zeros(shape=(n_states, ), dtype=int)

    delta_es = numpy.zeros(shape=(n, ), dtype=int)
    weights = numpy.zeros(shape=(n, ), dtype=numpy.float64)

    i = 0
    for s in states:
        s = int(s)
        assert 0 <= s and s < n_states
        start_indices[s] = i
        for edge in weighted_obs_by_state[s]:
            weights[i] = edge.weight
            delta_es[i] = edge.delta_e
            i += 1
        end_indices[s] = i

    return PackedObservations(
        start_indices=start_indices,
        end_indices=end_indices,
        delta_es=delta_es,
        weights=weights,
    )

    pack weighted observations into C friendly compressed-sparse format that can
    be iterated over without hashing or attribute access

    :param states: list of state indices
    :param outgoing_observations: lists of WeightedObservation keyed by state index
    :return: (PackedObservations
    
pack weighted observations into C friendly compressed-sparse format that can
be iterated over without hashing or attribute access
pack weighted observations into C friendly compressed - sparse format that can be iterated over without hashing or attribute access
def pack_observations(states: numpy.typing.NDArray[numpy.int64],
                      weighted_obs_by_state: typing.Dict[numpy.int64, typing.Sequence[WeightedObservation]]):
    n_states = len(states)
    n = sum(len(weighted_obs_by_state[s]) for s in states)
    start_indices = numpy.zeros(shape=(n_states, ), dtype=int)
    end_indices = numpy.zeros(shape=(n_states, ), dtype=int)
    delta_es = numpy.zeros(shape=(n, ), dtype=int)
    weights = numpy.zeros(shape=(n, ), dtype=numpy.float64)
    i = 0
    for s in states:
        s = int(s)
        assert 0 <= s and s < n_states
        start_indices[s] = i
        for edge in weighted_obs_by_state[s]:
            weights[i] = edge.weight
            delta_es[i] = edge.delta_e
            i += 1
        end_indices[s] = i
    return PackedObservations(
        start_indices=start_indices,
        end_indices=end_indices,
        delta_es=delta_es,
        weights=weights,
    )
def pack_observations ( states : numpy . typing . NDArray [ numpy . int64 ] , weighted_obs_by_state : typing . Dict [ numpy . int64 , typing . Sequence [ WeightedObservation ] ] ) : n_states = len ( states ) n = sum ( len ( weighted_obs_by_state [ s ] ) for s in states ) start_indices = numpy . zeros ( shape = ( n_states , ) , dtype = int ) end_indices = numpy . zeros ( shape = ( n_states , ) , dtype = int ) delta_es = numpy . zeros ( shape = ( n , ) , dtype = int ) weights = numpy . zeros ( shape = ( n , ) , dtype = numpy . float64 ) i = 0 for s in states : s = int ( s ) assert 0 <= s and s < n_states start_indices [ s ] = i for edge in weighted_obs_by_state [ s ] : weights [ i ] = edge . weight delta_es [ i ] = edge . delta_e i += 1 end_indices [ s ] = i return PackedObservations ( start_indices = start_indices , end_indices = end_indices , delta_es = delta_es , weights = weights , )
pack weighted observations into C friendly compressed-sparse format that can
be iterated over without hashing or attribute access
pack weighted observations into C friendly compressed - sparse format that can be iterated over without hashing or attribute access
"""
    pack weighted observations into C friendly compressed-sparse format that can
    be iterated over without hashing or attribute access

    :param states: list of state indices
    :param outgoing_observations: lists of WeightedObservation keyed by state index
    :return: (PackedObservations
    """
None
states
numpy.typing.NDArray[numpy.int64]
list of state indices
list of state indices
weighted_obs_by_state
typing.Dict[numpy.int64, typing.Sequence[WeightedObservation]]
outgoing_observations
lists of WeightedObservation keyed by state index
lists of WeightedObservation keyed by state index