+ def putx(self, ss, es, cls, data):
+ self.put(ss, es, self.out_ann, [cls, data,])
+
+ def putb(self, ss, es, cls , data):
+ self.put(ss, es, self.out_binary, [cls, data,])
+
+ def snums_to_usecs(self, snum_count):
+ if not self.samplerate:
+ return None
+ snums_per_usec = self.samplerate / 1e6
+ usecs = snum_count / snums_per_usec
+ return usecs
+
+ def lookup_proto_ann_txt(self, key, variables):
+ ann = {
+ 'RESET_SYM': [Ann.RESET_SYM, 'Reset', 'R',],
+ 'INTR_SYM': [Ann.INTR_SYM, 'Interrupt', 'Intr', 'I',],
+ 'START_SYM': [Ann.START_SYM, 'Start', 'ST', 'S',],
+ 'STOP_SYM': [Ann.STOP_SYM, 'Stop', 'SP', 'P',],
+ 'BIT_SYM': [Ann.BIT_SYM, '{bit}',],
+ 'ATR_BYTE': [Ann.ATR_BYTE,
+ 'Answer To Reset: {data:02x}',
+ 'ATR: {data:02x}',
+ '{data:02x}',
+ ],
+ 'CMD_BYTE': [Ann.CMD_BYTE,
+ 'Command: {data:02x}',
+ 'Cmd: {data:02x}',
+ '{data:02x}',
+ ],
+ 'OUT_BYTE': [Ann.OUT_BYTE,
+ 'Outgoing data: {data:02x}',
+ 'Data: {data:02x}',
+ '{data:02x}',
+ ],
+ 'PROC_BYTE': [Ann.PROC_BYTE,
+ 'Internal processing: {data:02x}',
+ 'Proc: {data:02x}',
+ '{data:02x}',
+ ],
+ 'ATR_DATA': [Ann.ATR_DATA,
+ 'Answer To Reset: {data}',
+ 'ATR: {data}',
+ '{data}',
+ ],
+ 'CMD_DATA': [Ann.CMD_DATA,
+ 'Command: {data}',
+ 'Cmd: {data}',
+ '{data}',
+ ],
+ 'OUT_DATA': [Ann.OUT_DATA,
+ 'Outgoing: {data}',
+ 'Out: {data}',
+ '{data}',
+ ],
+ 'PROC_DATA': [Ann.PROC_DATA,
+ 'Processing: {data}',
+ 'Proc: {data}',
+ '{data}',
+ ],
+ }.get(key, None)
+ if ann is None:
+ return None, []
+ cls, texts = ann[0], ann[1:]
+ texts = [t.format(**variables) for t in texts]
+ return cls, texts
+
+ def text_for_accu_bytes(self, accu):
+ if not accu:
+ return None, None, None, None
+ ss, es = accu[0][1], accu[-1][2]
+ data = [a[0] for a in accu]
+ text = " ".join(['{:02x}'.format(a) for a in data])
+ return ss, es, data, text
+
+ def flush_queued(self):
+ '''Flush previously accumulated operations details.'''
+
+ # Can be called when either the completion of an operation got
+ # detected (reliably), or when some kind of reset condition was
+ # met while a potential previously observed operation has not
+ # been postprocessed yet (best effort). Should not harm when the
+ # routine gets invoked while no data was collected yet, or was
+ # flushed already.
+ # BEWARE! Will void internal state. Should really only get called
+ # "between operations", NOT between fields of an operation.
+
+ if self.atr_bytes:
+ key = 'ATR_DATA'
+ ss, es, _, text = self.text_for_accu_bytes(self.atr_bytes)
+ cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
+ self.putx(ss, es, cls, texts)
+
+ if self.cmd_bytes:
+ key = 'CMD_DATA'
+ ss, es, _, text = self.text_for_accu_bytes(self.cmd_bytes)
+ cls, texts = self.lookup_proto_ann_txt(key, {'data': text})
+ self.putx(ss, es, cls, texts)