小编Ser*_*gio的帖子

我有两个通过外键相关的表,这里他们使用声明映射

class Task(DeclarativeBase):
    __tablename__ = 'task'
    id = Column(Integer, primary_key=True)
    state = Column(Integer, default=0)
    obs_id = Column(Integer, ForeignKey('obs.id'), nullable=False)

class Obs(DeclarativeBase):
    __tablename__ = 'obs'
    id = Column(Integer, primary_key=True)
    state = Column(Integer, default=0)

因此,当obs.state更改为值2时,我想更新相关的task.state.目前我正在手动执行(使用称为task的关系)

obs.state = 2
obs.task.state = 2

但我更喜欢使用触发器.我已经检查过这可以在sqlite中使用

CREATE TRIGGER update_task_state UPDATE OF state ON obs
  BEGIN
    UPDATE task SET state = 2 WHERE (obs_id = old.id) and (new.state = 2);
  END;

但我找不到如何在sqlalchemy中表达这一点.我已多次读取插入更新默认值,但无法找到方法.我不知道是否有可能.

我试图理解如何使用nditer进行减少,在我的情况下将3d数组转换为2d数组.

我按照这里的帮助 http://docs.scipy.org/doc/numpy/reference/arrays.nditer.html设法创建了一个在输入的最后一个轴上应用缩减的函数.有了这个功能

def nditer_sum(data, red_axes):
    it = numpy.nditer([data, None],
            flags=['reduce_ok', 'external_loop'],
            op_flags=[['readonly'], ['readwrite', 'allocate']],
            op_axes=[None, red_axes])
    it.operands[1][...] = 0

    for x, y in it:
        y[...] = x.sum()

    return it.operands[1]

我可以获得与data.sum等效的东西(axis = 2)

>>> data = numpy.arange(2*3*4).reshape((2,3,4))
>>> nditer_sum(data, [0, 1, -1])
[[ 6 22 38]
[54 70 86]]
>>> data.sum(axis=2)
[[ 6 22 38]
[54 70 86]]

所以为了获得与data.sum(axis = 0)相当的东西,我认为将参数red_axes更改为[-1,0,1]就足够了但结果却完全不同了.

>>> data = numpy.arange(2*3*4).reshape((2,3,4))
>>> data.sum(axis=0)
[[12 14 16 18]
 [20 22 24 26]
 [28 30 32 34]] …

我正在使用dbus来传达两个程序.一个人创建一个大图像,然后将其发送给其他程序进行进一步处理.我将图像作为ByteArray传递.

使用2000x2000图像,我的程序可以工作,但是对于4000x4000,它会用以下方法进行操作:

process 2283: arguments to dbus_message_iter_append_fixed_array() were       
incorrect,assertion "n_elements <= DBUS_MAXIMUM_ARRAY_LENGTH / _dbus_type_get_alignment  
(element_type)" failed in file dbus-message.c line 2628.

我明白这意味着我传递的数组大于允许的数组.还有其他方法可以在dbus中传递大数据结构吗？

这是我正在使用的代码的摘录:

handle = StringIO()
# hdulist contains the large data structure
hdulist.writeto(handle)
hdub = dbus.ByteArray(handle.getvalue())
# this sends the image via dbus
self.dbi.store_image(hdub)

在另一端,我有类似的东西

def store_image(self, bindata):
    # Convert binary data back to HDUList
    handle = StringIO.StringIO(bindata)
    hdulist = pyfits.open(handle)

我在线程中调用dbus方法时遇到段错误.这是我的场景:我有一个程序Service1公开方法测试.第二个程序Service2公开了一个方法公开.由于这个方法做了一些严肃的数值计算,我将一些params从expose传递给正在运行的线程读取器.反过来,该线程在结束其工作时调用Service1的方法测试.我在最后一次dbus调用中发现了段错误.

代码:

# Service1.py
class Service1(Object):
    def __init__(self, bus):
        name = BusName('com.example.Service1', bus)
        path = '/'
       super(Service1, self).__init__(name, path)

    @method(dbus_interface='com.example.Service1',
        in_signature='s', out_signature='s')
    def test(self, test):
        print 'test being called'
        return test

dbus_loop = DBusGMainLoop()
dsession = SessionBus(mainloop=dbus_loop)
loop = gobject.MainLoop()
gobject.threads_init()

im = Service1(dsession)
loop.run()


# Service2.py
dbus_loop = DBusGMainLoop()
dsession = SessionBus(mainloop=dbus_loop)

class Service2(Object):
def __init__(self, bus):
    name = BusName('com.example.Service2', bus)
    super(Service2, self).__init__(name, '/')

    self.queue = Queue()
    self.db = bus.get_object('com.example.Service1', '/')
    self.dbi = dbus.Interface(self.db, dbus_interface='com.example.Service1')

@method(dbus_interface='com.example.Service2',
        in_signature='', out_signature='')
def …