hammerlib.pyx

Go to the documentation of this file.

#**** This file is a part of the HAMMER library
#**** Copyright (C) 2016 - 2020 The HAMMER Collaboration
#**** HAMMER is licensed under version 3 of the GPL; see COPYING for details
#**** Please note the MCnet academic guidelines; see GUIDELINES for details

## \file  hammerlib.pyx
#  \brief Cython class to wrap Hammer library
#

# distutils: language = c++
# cython: c_string_type=unicode, c_string_encoding=default

from typing import List, Set, Dict, Tuple, Optional, FrozenSet, AnyStr, Union, Any
from numbers import Number

EventUIDGroup = Set[FrozenSet[int]]
Vec3 = Tuple[float, float, float]

from cppdefs cimport Hammer as cpp_Hammer
from cppdefs cimport Process as cpp_Process
from cppdefs cimport Particle as cpp_Particle
from cppdefs cimport FourMomentum as cpp_FourMomentum
from cppdefs cimport IOBuffer as cpp_IOBuffer
from cppdefs cimport IOBuffers as cpp_IOBuffers
from cppdefs cimport WTerm as cpp_WTerm
from cppdefs cimport HistoInfo as cpp_HistoInfo
from cppdefs cimport RecordType as cpp_RecordType
from cppdefs cimport BinContents as cpp_BinContents
from cppdefs cimport UMap as cpp_UMap
from cppdefs cimport Log as cpp_Log
from cppdefs cimport version as cpp_version

from cpython cimport array
import array
from cpython cimport Py_buffer

from io import FileIO
from struct import pack, unpack, calcsize
from enum import Enum, IntEnum

from libc.stdint cimport uint8_t
from libcpp.string cimport string
from libcpp.map cimport map as cmap
from libcpp.pair cimport pair
from libcpp.vector cimport vector
from libcpp.memory cimport unique_ptr
from libcpp.set cimport set as cset
from cython.operator import dereference as deref, preincrement as inc
from cpython cimport bool

from cymove cimport cymove as move
from cpython.version cimport PY_MAJOR_VERSION

import cmath

HAVE_NUMPY = True
try:
    import numpy as np
except:
    HAVE_NUMPY = False

def _is_string(value: Any) -> bool:
    return isinstance(value, (str, bytes, bytearray))

def _is_set(value: Any) -> bool:
    return isinstance(value, (set, frozenset))

def _is_event_uid_group(value: Any) -> bool:
    return (_is_set(value) and
            all([(_is_set(elem) and
                all([(type(elem2) is int and elem2>0)
                for elem2 in elem]))
            for elem in value]))

def _is_list_of_string(value: Any) -> bool:
    return (isinstance(value, list) and all([_is_string(elem) for elem in value]))

def _is_list_of_list_of_string(value: Any) -> bool:
    return (isinstance(value, list) and all([_is_list_of_string(elem) for elem in value]))

def _is_vec_float(value: Any) -> bool:
    return (isinstance(value, list) and all([type(elem) is float for elem in value]))

def _is_vec_complex(value: Any) -> bool:
    return (isinstance(value, list) and all([isinstance(elem, Number) for elem in value]))

def _is_vec_int(value: Any) -> bool:
    return (isinstance(value, list) and all([type(elem) is int for elem in value]))

def _is_dict_float(value: Any) -> bool:
    return (isinstance(value, dict) and all([(type(elem_v) is float and _is_string(elem_k)) for elem_k, elem_v in value.iteritems()]))

def _is_dict_complex(value: Any) -> bool:
    return (isinstance(value, dict) and all([(isinstance(elem_v, Number) and _is_string(elem_k)) for elem_k, elem_v in value.iteritems()]))

cdef _to_set_of_sets(cset[cset[size_t]] value):
    return frozenset([frozenset([int(elem2) for elem2 in elem]) for elem in value])

class WTerm(Enum):
    COMMON = 0
    NUMERATOR = 1
    DENOMINATOR = 2

cdef cpp_WTerm _to_cpp_wterm(arg):
    if arg == WTerm.COMMON:
        return cpp_WTerm.COMMON
    elif arg == WTerm.NUMERATOR:
        return cpp_WTerm.NUMERATOR
    elif arg == WTerm.DENOMINATOR:
        return cpp_WTerm.DENOMINATOR
    else:
        return cpp_WTerm.COMMON

class RecordType(IntEnum):
    UNDEFINED = ord('u')
    HEADER = ord('b')
    EVENT = ord('e')
    HISTOGRAM = ord('h')
    RATE = ord('r')
    HISTOGRAM_DEFINITION = ord('d')

cdef class HistoInfo:

    cdef cpp_HistoInfo c_histo_info

    cdef from_cpp(self, cpp_HistoInfo info):
        self.c_histo_info = info

    # Attribute access
    @property
    def name(self) -> AnyStr:
        return self.c_histo_info.name
    @name.setter
    def name(self, name: AnyStr):
        self.c_histo_info.name = name

    @property
    def scheme(self) -> AnyStr:
        return self.c_histo_info.scheme
    @scheme.setter
    def scheme(self, scheme: AnyStr):
        self.c_histo_info.scheme = scheme

    @property
    def event_group_id(self) -> EventUIDGroup:
        return set([frozenset(elem) for elem in self.c_histo_info.eventGroupId])
    @event_group_id.setter
    def event_group_id(self, event_group_id: EventUIDGroup):
        self.c_histo_info.eventGroupId = event_group_id

cdef class BinContents:

    cdef cpp_BinContents c_bins

    @staticmethod
    cdef from_cpp(cpp_BinContents value):
        result = BinContents()
        result.c_bins = value
        return result

    @property
    def sum_wi(self) -> float:
        return self.c_bins.sumWi
    @sum_wi.setter
    def sum_wi(self, sum_wi: float):
        self.c_bins.sumWi = sum_wi

    @property
    def sum_wi2(self) -> float:
        return self.c_bins.sumWi2
    @sum_wi2.setter
    def sum_wi2(self, sum_wi2: float):
        self.c_bins.sumWi2 = sum_wi2

    @property
    def n(self) -> int:
        return self.c_bins.n
    @n.setter
    def n(self, n: int):
        self.c_bins.n = n

cdef class IOBuffers:

    cdef cpp_IOBuffers* p_io_buffers
    cdef cpp_IOBuffers.iterator curr_it

    def __cinit__(self):
        self.p_io_buffers = new cpp_IOBuffers()
        self.curr_it = self.p_io_buffers.begin()

    def __dealloc__(self):
        del self.p_io_buffers

    cdef from_cpp(self, cpp_IOBuffers buf):
        del self.p_io_buffers
        self.p_io_buffers = new cpp_IOBuffers(move(buf))
        self.curr_it = self.p_io_buffers.begin()

    def __iter__(self):
        self.curr_it = self.p_io_buffers.begin()
        return self

    def __next__(self):
        if self.curr_it == self.p_io_buffers.end():
            raise StopIteration
        result = IOBuffer()
        result.from_cpp(deref(self.curr_it))
        inc(self.curr_it)
        return result

    def save(self, file_handler : FileIO):
        cdef cpp_IOBuffers.iterator it = self.p_io_buffers.begin()
        while it != self.p_io_buffers.end():
            result = IOBuffer()
            result.from_cpp(deref(it))
            result.save(file_handler)
            inc(it)

cdef class IOBuffer:

    cdef cpp_IOBuffer c_buffer
    cdef array.array buf
    cdef int view_count
    cdef Py_ssize_t strides

    def __cinit__(self):
        self.buf = array.array('B', [0]*1000)
        self.c_buffer.kind = cpp_RecordType.UNDEFINED
        self.c_buffer.length = len(self.buf)
        self.c_buffer.start = <uint8_t*>(&self.buf.data.as_uchars[0])
        self.view_count = 0

    def __len__(self):
        return self.c_buffer.length

    def __getbuffer__(self, Py_buffer *view, int flags):
        cdef Py_ssize_t itemsize = 1
        view.buf = <unsigned char *> self.c_buffer.start
        view.format = 'B'
        view.internal = NULL
        view.itemsize = 1
        view.len = self.c_buffer.length
        view.ndim = 1
        view.obj = self
        view.readonly = 0
        view.shape = <Py_ssize_t*> &self.c_buffer.length
        view.strides = &itemsize
        view.suboffsets = NULL
        self.view_count += 1

    def __releasebuffer__(self, Py_buffer *buffer):
        self.view_count -= 1

    cdef from_cpp(self, cpp_IOBuffer buf):
        if self.view_count > 0:
            raise ValueError("can't edit buffer while it's accessed via buffer protocol")
        self.c_buffer = buf
        self.view_count = 0

    cdef cpp_IOBuffer* to_cpp(self):
        return &self.c_buffer

    cpdef init_from_size(self, int data_size = 0):
        if self.view_count > 0:
            raise ValueError("can't edit buffer while it's accessed via buffer protocol")
        array.resize(self.buf, data_size)
        self.c_buffer.kind = cpp_RecordType.UNDEFINED
        self.c_buffer.length = len(self.buf)
        self.c_buffer.start = <uint8_t*>(&self.buf.data.as_uchars[0])
        self.view_count = 0

    @property
    def kind(self) -> RecordType:
        return self.c_buffer.kind

    @kind.setter
    def kind(self, kind: RecordType):
        self.c_buffer.kind = <cpp_RecordType>(kind)

    def save(self, file_handler : FileIO):
        file_handler.write(pack(b'b', self.c_buffer.kind))
        file_handler.write(pack(b'<L', self.c_buffer.length))
        file_handler.write(self)

    def load(self, file_handler : FileIO):
        tmp = file_handler.read(1)
        if len(tmp) == 0:
            return False
        typebuf = unpack(b'b',tmp)
        form = b'<L'
        sizebuf = unpack(form, file_handler.read(calcsize(form)))[0]
        self.init_from_size(sizebuf)
        self.c_buffer.kind = <cpp_RecordType>(typebuf[0])
        file_handler.readinto(self)
        return True


cdef class FourMomentum:

    cdef cpp_FourMomentum c_mom

    def __cinit__(self, e: float = 0., px: float = 0., py: float = 0., pz: float = 0.):
        self.c_mom = cpp_FourMomentum(e, px, py, pz)

    @staticmethod
    def fromPtEtaPhiM(pt: float, eta: float, phi: float, m: float) -> FourMomentum:
        result = FourMomentum()
        result.c_mom = cpp_FourMomentum.fromPtEtaPhiM(pt, eta, phi, m)
        return result

    @staticmethod
    def fromEtaPhiME(eta: float, phi: float, m: float, e: float) -> FourMomentum:
        result = FourMomentum()
        result.c_mom = cpp_FourMomentum.fromEtaPhiME(eta, phi, m, e)
        return result

    @staticmethod
    def fromPM(px: float, py: float, pz: float, m: float) -> FourMomentum:
        result = FourMomentum()
        result.c_mom = cpp_FourMomentum.fromPM(px, py, pz, m)
        return result

    @property
    def px(self) -> float:
        return self.c_mom.px()
    @px.setter
    def px(self, value: float):
        self.c_mom.setPx(value)

    @property
    def py(self) -> float:
        return self.c_mom.py()
    @py.setter
    def py(self, value: float):
        self.c_mom.setPy(value)

    @property
    def pz(self) -> float:
        return self.c_mom.pz()
    @pz.setter
    def pz(self, value: float):
        self.c_mom.setPz(value)

    @property
    def e(self) -> float:
        return self.c_mom.E()
    @e.setter
    def e(self, value: float):
        self.c_mom.setE(value)

    def mass(self) -> float:
        return self.c_mom.mass()

    def mass2(self) -> float:
        return self.c_mom.mass2()

    def p(self) -> float:
        return self.c_mom.p()

    def p2(self) -> float:
        return self.c_mom.p2()

    def pt(self) -> float:
        return self.c_mom.pt()

    def rapidity(self) -> float:
        return self.c_mom.rapidity()

    def phi(self) -> float:
        return self.c_mom.phi()

    def eta(self) -> float:
        return self.c_mom.eta()

    def theta(self) -> float:
        return self.c_mom.theta()

    def p_vec(self) -> Vec3:
        res = self.c_mom.pVec()
        return (res[0], res[1], res[2])

    def gamma(self) -> float:
        return self.c_mom.gamma()

    def beta(self) -> float:
        return self.c_mom.beta()

    def boost_vector(self) -> Vec3:
        res = self.c_mom.boostVector()
        return (res[0], res[1], res[2])

cdef class Particle:

    cdef cpp_Particle c_part

    def __cinit__(self, p: FourMomentum = FourMomentum(), pdg: int = 0):
        self.c_part = cpp_Particle(p.c_mom, pdg)

    @staticmethod
    cdef Particle from_cpp(cpp_Particle p):
        part = Particle()
        part.c_part = p
        return part

    @property
    def momentum(self) -> FourMomentum:
        result = FourMomentum()
        result.c_mom = self.c_part.momentum()
        return result

    @momentum.setter
    def momentum(self, val: FourMomentum):
        self.c_part.setMomentum(val.c_mom)

    @property
    def p(self) -> FourMomentum:
        result = FourMomentum()
        result.c_mom = self.c_part.momentum()
        return result

    @p.setter
    def p(self, val: FourMomentum):
        self.c_part.setMomentum(val.c_mom)

    @property
    def pdg_id(self) -> int:
        return self.c_part.pdgId()

    @pdg_id.setter
    def pdg_id(self, val: int):
        self.c_part.setPdgId(val)


cdef class Process:

    cdef cpp_Process c_proc

    def add_particle(self, value: Particle):
        return self.c_proc.addParticle(value.c_part)

    def add_vertex(self, parent: int, daughters: List[int]):
        self.c_proc.addVertex(parent, daughters)

    def remove_vertex(self, vertex_id: int, prune: bool = False):
        self.c_proc.removeVertex(vertex_id, prune)

    def get_daughters_ids(self, parent: int = 0):
        return self.c_proc.getDaughtersIds(parent)

    def get_parent_id(self, daughter: int):
        return self.c_proc.getParentId(daughter)

    def get_siblings_ids(self, particle: int = 0):
        return self.c_proc.getSiblingsIds(particle)

    def get_particle(self, id: int):
        result = Particle()
        result.c_part = self.c_proc.getParticle(id)
        return result

    def get_daughters(self, parent: int = 0, sorted: bool = False):
        result = self.c_proc.getDaughters(parent, sorted)
        return [Particle.from_cpp(elem) for elem in result]

    def get_siblings(self, particle: int = 0, sorted: bool = False):
        result = self.c_proc.getSiblings(particle, sorted)
        return [Particle.from_cpp(elem) for elem in result]

    def get_parent(self, daughter: int):
        result = Particle()
        result.c_part = self.c_proc.getParent(daughter)
        return result

    def is_parent(self, particle: int):
        return self.c_proc.isParent(particle)

    def get_first_vertex(self):
        return self.c_proc.getFirstVertex()

    def get_id(self):
        return self.c_proc.getId()

    def full_id(self):
        return self.c_proc.fullId()

    def get_vertex_id(self, vertex):
        return self.c_proc.getVertexId(vertex)

    def num_particles(self, without_photons: bool = False):
        return self.c_proc.numParticles(without_photons)

## \brief main Hammer python class
#
#  Provides the functionalities of `Hammer::Hammer` visible to a python program
#
#  \ingroup PyExt

cdef class Hammer:

    ## pointer to Hammer::Hammer class
    cdef cpp_Hammer *wrapped

    ## base constructor
    def __cinit__ (self):
        self.wrapped = new cpp_Hammer()
        if self.wrapped is NULL:
            raise MemoryError()

    ## destructor
    def __dealloc__ (self):
        if self.wrapped is not NULL:
            del self.wrapped

    def init_run(self):
        self.wrapped.initRun()

    def init_event(self, weight: float = 1.0):
        self.wrapped.initEvent(weight)

    def add_process(self, proc: Process) -> int:
        return self.wrapped.addProcess(proc.c_proc)

    def remove_process(self, proc_id: int):
        self.wrapped.removeProcess(proc_id)

    def set_event_histogram_bin(self, name: AnyStr, bins: List[int]):
        self.wrapped.setEventHistogramBin(name, bins)

    def fill_event_histogram(self, name: AnyStr, values: List[float]):
        self.wrapped.fillEventHistogram(name, values)

    def set_event_base_weight(self, weight: float):
        self.wrapped.setEventBaseWeight(weight)

    def process_event(self):
        self.wrapped.processEvent()

    def load_event_weights(self, buf: IOBuffer, merge: bool = False) -> bool:
        return self.wrapped.loadEventWeights(deref(buf.to_cpp()), merge)

    def save_event_weights(self) -> IOBuffer:
        result = IOBuffer()
        result.from_cpp(self.wrapped.saveEventWeights())
        return result

    def write_event_weights(self, file_handler: FileIO):
        file_handler.write(self.save_event_weights())

    def load_run_header(self, buf: IOBuffer, merge: bool = False) -> bool:
        return self.wrapped.loadRunHeader(deref(buf.to_cpp()), merge)

    def save_run_header(self) -> IOBuffer:
        result = IOBuffer()
        result.from_cpp(self.wrapped.saveRunHeader())
        return result

    def write_run_header(self, file_handler: FileIO):
        file_handler.write(self.save_run_header())

    def load_histogram_definition(self, buf: IOBuffer, merge: bool = False) -> AnyStr:
        return self.wrapped.loadHistogramDefinition(deref(buf.to_cpp()), merge)

    def load_histogram(self, buf: IOBuffer, merge: bool = False) -> HistoInfo:
        result = HistoInfo()
        result.from_cpp(self.wrapped.loadHistogram(deref(buf.to_cpp()), merge))
        return result

    def save_histogram(self, name_or_info: Union[AnyStr, HistoInfo]) -> IOBuffers:
        result = IOBuffers()
        cdef string s1
        if _is_string(name_or_info):
            s1 = name_or_info
            result.from_cpp(self.wrapped.saveHistogram(s1))
        elif isinstance(name_or_info, HistoInfo):
            result.from_cpp(self.wrapped.saveHistogram(((HistoInfo)(name_or_info)).c_histo_info))
        else:
            print("Invalid Type")
        return result

    def write_histogram(self, file_handler: FileIO, name_or_info: Union[AnyStr, HistoInfo]):
        for elem in self.save_histogram(name_or_info):
            file_handler.write(elem)

    def save_histogram(self, name: AnyStr, scheme_or_event_ids: Union[AnyStr, EventUIDGroup]) -> IOBuffers:
        result = IOBuffers()
        cdef string s1 = name
        cdef string s2
        cdef cset[cset[size_t]] s3
        if _is_string(scheme_or_event_ids) == AnyStr:
            s2 = scheme_or_event_ids
            result.from_cpp(self.wrapped.saveHistogram(s1, s2))
        elif _is_event_uid_group(scheme_or_event_ids):
            s3 = scheme_or_event_ids
            result.from_cpp(self.wrapped.saveHistogram(s1, s3))
        else:
            print("Invalid Type")
        return result

    def write_histogram(self, file_handler: FileIO, name: AnyStr, scheme_or_event_ids: Union[AnyStr, EventUIDGroup]):
        for elem in self.save_histogram(name, scheme_or_event_ids):
            file_handler.write(elem)

    def save_histogram(self, name: AnyStr, scheme: AnyStr, event_ids: EventUIDGroup) -> IOBuffers:
        result = IOBuffers()
        result.from_cpp(self.wrapped.saveHistogram(name, scheme, event_ids))
        return result

    def write_histogram(self, file_handler: FileIO, name: AnyStr, scheme: AnyStr, event_ids: EventUIDGroup):
        for elem in self.save_histogram(name, scheme, event_ids):
            file_handler.write(elem)

    def load_rates(self, buf: IOBuffer, merge: bool = False) -> bool:
        return self.wrapped.loadRates(deref(buf.to_cpp()), merge)

    def save_rates(self) -> IOBuffer:
        result = IOBuffer()
        result.from_cpp(self.wrapped.saveRates())
        return result

    def write_rates(self, file_handler: FileIO):
        file_handler.write(self.save_rates())

    def read_cards(self, file_decays: AnyStr, file_options: AnyStr):
        self.wrapped.readCards(file_decays, file_options)

    def save_option_card(self, file_options: AnyStr, use_default: bool = True):
        self.wrapped.saveOptionCard(file_options, use_default)

    def save_header_card(self, file_decays: AnyStr):
        self.wrapped.saveHeaderCard(file_decays)

    def save_references(self, file_refs: AnyStr):
        self.wrapped.saveReferences(file_refs)

    def set_options(self, options: AnyStr):
        self.wrapped.setOptions(options)

    def set_header(self, options: AnyStr):
        self.wrapped.setHeader(options)

    def add_total_sum_of_weights(self, compress: bool = False, with_errors: bool = False):
        self.wrapped.addTotalSumOfWeights(compress, with_errors)

    def add_histogram(self, name: AnyStr, bin_sizes: List[int], has_under_over_flow: bool = True, ranges: List[Tuple[float, float]] = []):
        self.wrapped.addHistogram(name, bin_sizes, has_under_over_flow, ranges)

    def add_histogram(self, name: AnyStr, bin_edges: List[List[float]], has_under_over_flow: bool):
        self.wrapped.addHistogram(name, bin_edges, has_under_over_flow)

    def collapse_processes_in_histogram(self, name: AnyStr):
        self.wrapped.collapseProcessesInHistogram(name)

    def keep_errors_in_histogram(self, name: AnyStr, value: bool = True):
        self.wrapped.keepErrorsInHistogram(name, value)

    def specialize_WC_in_weights(self, process: AnyStr, values_or_settings: Union[List[complex], Dict[AnyStr, complex]]):
        cdef string s1 = process
        cdef vector[double complex] v
        cdef cmap[string, double complex] d
        if _is_vec_complex(values_or_settings):
            v = values_or_settings
            self.wrapped.specializeWCInWeights(s1, v)
        elif _is_dict_complex(values_or_settings):
            d = values_or_settings
            self.wrapped.specializeWCInWeights(s1, d)
        else:
            print("Invalid Type")

    def reset_specialize_WC_in_weights(self, process: AnyStr):
        self.wrapped.resetSpecializeWCInWeights(process)

    def specialize_wc_in_histogram(self, name: AnyStr, process: AnyStr, values_or_settings: Union[List[complex], Dict[AnyStr, complex]]):
        cdef string s1 = name
        cdef string s2 = process
        cdef vector[double complex] v
        cdef cmap[string, double complex] d
        if _is_vec_complex(values_or_settings):
            v = values_or_settings
            self.wrapped.specializeWCInHistogram(s1, s2, v)
        elif _is_dict_complex(values_or_settings):
            d = values_or_settings
            self.wrapped.specializeWCInHistogram(s1, s2, d)
        else:
            print("Invalid Type")

    def specialize_ff_in_histogram(self, name: AnyStr, process: AnyStr, group: AnyStr, values_or_settings: Union[List[float], Dict[AnyStr, float]]):
        cdef string s1 = name
        cdef string s2 = process
        cdef string s3 = group
        cdef vector[double] v
        cdef cmap[string, double] d
        if _is_vec_float(values_or_settings):
            v = values_or_settings
            self.wrapped.specializeFFInHistogram(s1, s2, s3, v)
        elif _is_dict_float(values_or_settings):
            d = values_or_settings
            self.wrapped.specializeFFInHistogram(s1, s2, s3, d)
        else:
            print("Invalid Type")

    def reset_specialization_in_histogram(self, name):
        self.wrapped.resetSpecializationInHistogram(name)

    def create_projected_histogram(self, old_name, new_name, collapsed_index_positions: Set[int]):
        self.wrapped.createProjectedHistogram(old_name, new_name, collapsed_index_positions)

    def remove_histogram(self, name):
        self.wrapped.removeHistogram(name)

    def add_ff_scheme(self, scheme_name, schemes: Dict[AnyStr, AnyStr]):
        self.wrapped.addFFScheme(scheme_name, schemes)

    def set_ff_input_scheme(self, schemes: Dict[AnyStr, AnyStr]):
        self.wrapped.setFFInputScheme(schemes)

    def get_ff_scheme_names(self) -> List[str]:
        res = self.wrapped.getFFSchemeNames()
        return [str(elem) for elem in res]

    def include_decay(self, name_or_names: Union[List[AnyStr], AnyStr]) -> None:
        cdef string s
        cdef vector[string] v
        if _is_list_of_string(name_or_names):
            v = name_or_names
            self.wrapped.includeDecay(v)
        elif _is_string(name_or_names):
            s = name_or_names
            self.wrapped.includeDecay(s)
        else:
            print("Invalid Type")

    def forbid_decay(self, name_or_names: Union[List[AnyStr], AnyStr]) -> None:
        cdef string s
        cdef vector[string] v
        if _is_list_of_string(name_or_names):
            v = name_or_names
            self.wrapped.forbidDecay(v)
        elif _is_string(name_or_names):
            s = name_or_names
            self.wrapped.forbidDecay(s)
        else:
            print("Invalid Type")

    def add_pure_ps_vertices(self, vertices: Set[AnyStr], what: WTerm = WTerm.NUMERATOR):
        self.wrapped.addPurePSVertices(vertices, _to_cpp_wterm(what))

    def clear_pure_ps_vertices(self, what: WTerm):
        self.wrapped.clearPurePSVertices(_to_cpp_wterm(what))

    def set_units(self, name: AnyStr = "GeV"):
        self.wrapped.setUnits(name)

    def set_wilson_coefficients(self, process: AnyStr, values_or_settings: Union[List[complex], Dict[AnyStr, complex]], what = WTerm.NUMERATOR):
        cdef string s1 = process
        cdef vector[double complex] v
        cdef cmap[string, double complex] d
        if _is_vec_complex(values_or_settings):
            v = values_or_settings
            self.wrapped.setWilsonCoefficients(s1, v, _to_cpp_wterm(what))
        elif _is_dict_complex(values_or_settings):
            d = values_or_settings
            self.wrapped.setWilsonCoefficients(s1, d, _to_cpp_wterm(what))
        else:
            print("Invalid Type")

    def set_wilson_coefficients_local(self, process: AnyStr, values_or_settings: Union[List[complex], Dict[AnyStr, complex]]):
        cdef string s1 = process
        cdef vector[double complex] v
        cdef cmap[string, double complex] d
        if _is_vec_complex(values_or_settings):
            v = values_or_settings
            self.wrapped.setWilsonCoefficientsLocal(s1, v)
        elif _is_dict_complex(values_or_settings):
            d = values_or_settings
            self.wrapped.setWilsonCoefficientsLocal(s1, d)
        else:
            print("Invalid Type")

    def reset_wilson_coefficients(self, process: AnyStr, what: WTerm = WTerm.NUMERATOR):
        self.wrapped.resetWilsonCoefficients(process, _to_cpp_wterm(what))

    def set_ff_eigenvectors(self, process: AnyStr, group: AnyStr, values_or_settings: Union[List[float], Dict[AnyStr, float]]):
        cdef string s1 = process
        cdef string s2 = group
        cdef vector[double] v
        cdef cmap[string, double] d
        if _is_vec_float(values_or_settings):
            v = values_or_settings
            self.wrapped.setFFEigenvectors(s1, s2, v)
        elif _is_dict_float(values_or_settings):
            d = values_or_settings
            self.wrapped.setFFEigenvectors(s1, s2, d)
        else:
            print("Invalid Type")

    def set_ff_eigenvectors_local(self, process: AnyStr, group: AnyStr, values_or_settings: Union[List[float], Dict[AnyStr, float]]):
        cdef string s1 = process
        cdef string s2 = group
        cdef vector[double] v
        cdef cmap[string, double] d
        if _is_vec_float(values_or_settings):
            v = values_or_settings
            self.wrapped.setFFEigenvectorsLocal(s1, s2, v)
        elif _is_dict_float(values_or_settings):
            d = values_or_settings
            self.wrapped.setFFEigenvectorsLocal(s1, s2, d)
        else:
            print("Invalid Type")

    def reset_ff_eigenvectors(self, process: AnyStr, group: AnyStr):
        self.wrapped.resetFFEigenvectors(process, group)

    def get_weight(self, scheme: AnyStr, processes: Union[List[int], List[List[AnyStr]]] = []) -> float:
        cdef string s1 = scheme
        cdef vector[size_t] v1
        cdef vector[vector[string]] v2
        if _is_vec_int(processes):
            v1 = processes
            return self.wrapped.getWeight(s1, v1)
        elif _is_list_of_list_of_string(processes):
            v2 = processes
            return self.wrapped.getWeight(s1, v2)
        else:
            print("Invalid Type")
            return 0.

    def get_weights(self, scheme: AnyStr) -> Dict[int, float]:
        return self.wrapped.getWeights(scheme)

    def get_rate(self, id_or_vertex: Union[int, AnyStr], scheme: AnyStr) -> float:
        cdef string s2 = scheme
        cdef string s1
        cdef Py_ssize_t i1
        if _is_string(id_or_vertex):
            s1 = id_or_vertex
            return self.wrapped.getRate(s1, s2)
        elif type(id_or_vertex) is int:
            i1 = int(id_or_vertex)
            return self.wrapped.getRate(i1, s2)
        else:
            print("Invalid Type")
            return 0.

    def get_rate(self, parent: int, daughters: List[int], scheme: AnyStr) -> float:
        return self.wrapped.getRate(parent, daughters, scheme)

    def get_denominator_rate(self, id_or_vertex: Union[int, AnyStr]) -> float:
        cdef string s1
        cdef Py_ssize_t i1
        if _is_string(id_or_vertex):
            s1 = id_or_vertex
            return self.wrapped.getDenominatorRate(s1)
        elif type(id_or_vertex) is int:
            i1 = int(id_or_vertex)
            return self.wrapped.getDenominatorRate(i1)
        else:
            print("Invalid Type")
            return 0.

    def get_denominator_rate(self, parent: int, daughters: List[int]) -> float:
        return self.wrapped.getDenominatorRate(parent, daughters)

    def get_histogram(self, name: AnyStr, scheme: AnyStr) -> List[BinContents]:
        result = self.wrapped.getHistogram(name, scheme)
        return [BinContents.from_cpp(elem) for elem in result]

    def get_histograms(self, name: AnyStr, scheme: AnyStr):
        cdef cpp_UMap[cset[cset[size_t]], vector[cpp_BinContents]] res
        cdef cpp_UMap[cset[cset[size_t]], vector[cpp_BinContents]].iterator it
        res = self.wrapped.getHistograms(name, scheme)
        results = dict()
        it = res.begin()
        while (it != res.end()):
            tmp_v = [BinContents.from_cpp(elem) for elem in deref(it).second]
            tmp_k = _to_set_of_sets(deref(it).first)
            results.update({tmp_k, tmp_v})
        return results

    if HAVE_NUMPY:

        def get_numpy_histogram(self, name: AnyStr, scheme: AnyStr) -> List[BinContents]:
            shape = self.wrapped.getHistogramShape(name)
            res = self.wrapped.getHistogram(name, scheme)
            results = np.array([BinContents.from_cpp(o) for o in res], dtype=BinContents)
            results.reshape(list(shape))
            return results

        def get_numpy_histograms(self, name: AnyStr, scheme: AnyStr):
            cdef cpp_UMap[cset[cset[size_t]], vector[cpp_BinContents]] res
            cdef cpp_UMap[cset[cset[size_t]], vector[cpp_BinContents]].iterator it
            shape = self.wrapped.getHistogramShape(name)
            res = self.wrapped.getHistograms(name, scheme)
            results = dict()
            it = res.begin()
            while (it != res.end()):
                tmp_v = np.array([BinContents.from_cpp(o) for o in deref(it).second], dtype=BinContents)
                tmp_k = _to_set_of_sets(deref(it).first)
                results.update({tmp_k, tmp_v})
            return results

    def get_histogram_event_ids(self, name: AnyStr, scheme: AnyStr) -> EventUIDGroup:
        return self.wrapped.getHistogramEventIds(name, scheme)

    def get_histogram_shape(self, name: AnyStr) -> List[int]:
        return self.wrapped.getHistogramShape(name)

    def get_histogram_bin_edges(self, name: AnyStr) -> List[List[float]]:
        return self.wrapped.getHistogramBinEdges(name)

    def histogram_has_under_over_flows(self, name: AnyStr) -> bool:
        return self.wrapped.histogramHasUnderOverFlows(name)

def version() -> str:
    return str(cpp_version())

class LogLevel(Enum):
    TRACE = 0
    DEBUG = 10
    INFO = 20
    WARN = 30
    WARNING = 30
    ERROR = 40
    CRITICAL = 50
    ALWAYS = 50

def set_log_level(name: AnyStr, level: LogLevel):
    cpp_Log.setLevel(name, int(level))

def set_log_levels(level_dict: Dict[AnyStr, LogLevel]):
    cpp_Log.setLevels(level_dict)

def set_log_show_timestamp(value: bool):
    cpp_Log.setShowTimestamp(value)

def set_log_show_level(value: bool):
    cpp_Log.setShowLevel(value)

def set_log_show_logger_name(value: bool):
    cpp_Log.setShowLoggerName(value)

def set_log_use_colors(value: bool):
    cpp_Log.setUseColors(value)

def set_log_warning_max_count(name: AnyStr, max_count: int):
    cpp_Log.setWarningMaxCount(name, max_count)

def reset_log_warning_counters():
    cpp_Log.resetWarningCounters()