diff --git a/secop/datatypes.py b/secop/datatypes.py index 0662325..55ed4a1 100644 --- a/secop/datatypes.py +++ b/secop/datatypes.py @@ -705,7 +705,7 @@ class ArrayOf(DataType): raise BadValueError( 'Array too big, holds at most %d elements!' % self.minlen) # apply subtype valiation to all elements and return as list - return [self.members(elem) for elem in value] + return tuple(self.members(elem) for elem in value) raise BadValueError( 'Can not convert %s to ArrayOf DataType!' % repr(value)) @@ -715,7 +715,7 @@ class ArrayOf(DataType): def import_value(self, value): """returns a python object from serialisation""" - return [self.members.import_value(elem) for elem in value] + return tuple(self.members.import_value(elem) for elem in value) def from_string(self, text): value, rem = Parser.parse(text) @@ -772,8 +772,8 @@ class TupleOf(DataType): 'Illegal number of Arguments! Need %d arguments.' % (len(self.members))) # validate elements and return as list - return [sub(elem) - for sub, elem in zip(self.members, value)] + return tuple(sub(elem) + for sub, elem in zip(self.members, value)) except Exception as exc: raise BadValueError('Can not validate:', str(exc)) @@ -783,7 +783,7 @@ class TupleOf(DataType): def import_value(self, value): """returns a python object from serialisation""" - return [sub.import_value(elem) for sub, elem in zip(self.members, value)] + return tuple(sub.import_value(elem) for sub, elem in zip(self.members, value)) def from_string(self, text): value, rem = Parser.parse(text) @@ -804,6 +804,11 @@ class TupleOf(DataType): a.compatible(b) +class ImmutableDict(dict): + def _no(self, *args, **kwds): + raise TypeError('a struct can not be modified, please overwrite instead') + __setitem__ = __delitem__ = clear = pop = popitem = setdefault = update = _no + class StructOf(DataType): @@ -843,7 +848,10 @@ class StructOf(DataType): ['%s=%s' % (n, repr(st)) for n, st in list(self.members.items())]), opt) def __call__(self, value): - """return the validated value or raise""" + """return the validated value or raise + + in principle, we should make it some sort of frozen dict + """ try: # XXX: handle optional elements !!! if len(list(value.keys())) != len(list(self.members.keys())): @@ -851,8 +859,8 @@ class StructOf(DataType): 'Illegal number of Arguments! Need %d arguments.' % len(list(self.members.keys()))) # validate elements and return as dict - return dict((str(k), self.members[k](v)) - for k, v in list(value.items())) + return ImmutableDict((str(k), self.members[k](v)) + for k, v in list(value.items())) except Exception as exc: raise BadValueError('Can not validate %s: %s' % (repr(value), str(exc))) @@ -871,8 +879,8 @@ class StructOf(DataType): raise BadValueError( 'Illegal number of Arguments! Need %d arguments.' % len( list(self.members.keys()))) - return dict((str(k), self.members[k].import_value(v)) - for k, v in list(value.items())) + return ImmutableDict((str(k), self.members[k].import_value(v)) + for k, v in list(value.items())) def from_string(self, text): value, rem = Parser.parse(text) diff --git a/secop_psi/ppms.py b/secop_psi/ppms.py index 3c596ba..1b0bbd2 100644 --- a/secop_psi/ppms.py +++ b/secop_psi/ppms.py @@ -163,13 +163,13 @@ class PpmsMixin(HasIodev, Module): to be reimplemented for modules looking at packed_status """ if not self.enabled: - self.status = [self.Status.DISABLED, 'disabled'] + self.status = (self.Status.DISABLED, 'disabled') return if value is None: - self.status = [self.Status.ERROR, 'invalid value'] + self.status = (self.Status.ERROR, 'invalid value') else: self.value = value - self.status = [self.Status.IDLE, ''] + self.status = (self.Status.IDLE, '') class Channel(PpmsMixin, Readable): @@ -305,7 +305,7 @@ class Level(PpmsMixin, Readable): def analyze_level(self, level, status): # ignore 'old reading' state of the flag, as this happens only for a short time # during measuring - return dict(value=level, status=[self.Status.IDLE, '']) + return dict(value=level, status=(self.Status.IDLE, '')) class Chamber(PpmsMixin, Drivable): @@ -352,17 +352,17 @@ class Chamber(PpmsMixin, Drivable): Override(visibility=3), } STATUS_MAP = { - StatusCode.unknown: [Status.WARN, 'unknown'], - StatusCode.purged_and_sealed: [Status.IDLE, 'purged and sealed'], - StatusCode.vented_and_sealed: [Status.IDLE, 'vented and sealed'], - StatusCode.sealed_unknown: [Status.WARN, 'sealed unknown'], - StatusCode.purge_and_seal: [Status.BUSY, 'purge and seal'], - StatusCode.vent_and_seal: [Status.BUSY, 'vent and seal'], - StatusCode.pumping_down: [Status.BUSY, 'pumping down'], - StatusCode.at_hi_vacuum: [Status.IDLE, 'at hi vacuum'], - StatusCode.pumping_continuously: [Status.IDLE, 'pumping continuously'], - StatusCode.venting_continuously: [Status.IDLE, 'venting continuously'], - StatusCode.general_failure: [Status.ERROR, 'general failure'], + StatusCode.unknown: (Status.WARN, 'unknown'), + StatusCode.purged_and_sealed: (Status.IDLE, 'purged and sealed'), + StatusCode.vented_and_sealed: (Status.IDLE, 'vented and sealed'), + StatusCode.sealed_unknown: (Status.WARN, 'sealed unknown'), + StatusCode.purge_and_seal: (Status.BUSY, 'purge and seal'), + StatusCode.vent_and_seal: (Status.BUSY, 'vent and seal'), + StatusCode.pumping_down: (Status.BUSY, 'pumping down'), + StatusCode.at_hi_vacuum: (Status.IDLE, 'at hi vacuum'), + StatusCode.pumping_continuously: (Status.IDLE, 'pumping continuously'), + StatusCode.venting_continuously: (Status.IDLE, 'venting continuously'), + StatusCode.general_failure: (Status.ERROR, 'general failure'), } channel = 'chamber' @@ -428,15 +428,15 @@ class Temp(PpmsMixin, Drivable): general_failure=15, ) STATUS_MAP = { - 0: [Status.ERROR, 'unknown'], - 1: [Status.IDLE, 'stable at target'], - 2: [Status.RAMPING, 'ramping'], - 5: [Status.STABILIZING, 'within tolerance'], - 6: [Status.STABILIZING, 'outside tolerance'], - 10: [Status.WARN, 'standby'], - 13: [Status.WARN, 'control disabled'], - 14: [Status.ERROR, 'can not complete'], - 15: [Status.ERROR, 'general failure'], + 0: (Status.ERROR, 'unknown'), + 1: (Status.IDLE, 'stable at target'), + 2: (Status.RAMPING, 'ramping'), + 5: (Status.STABILIZING, 'within tolerance'), + 6: (Status.STABILIZING, 'outside tolerance'), + 10: (Status.WARN, 'standby'), + 13: (Status.WARN, 'control disabled'), + 14: (Status.ERROR, 'can not complete'), + 15: (Status.ERROR, 'general failure'), } properties = { 'iodev': Attached(), @@ -454,7 +454,7 @@ class Temp(PpmsMixin, Drivable): def update_value_status(self, value, packed_status): """update value and status""" if value is None: - self.status = [self.Status.ERROR, 'invalid value'] + self.status = (self.Status.ERROR, 'invalid value') return self.value = value status = self.STATUS_MAP[packed_status & 0xf] @@ -466,7 +466,7 @@ class Temp(PpmsMixin, Drivable): self.log.debug('time needed to change to busy: %.3g', now - self._last_change) self._last_change = 0 else: - status = [self.Status.BUSY, 'changed target'] + status = (self.Status.BUSY, 'changed target') if abs(self.value - self.target) < self.target * 0.01: self._last_target = self.target elif self._last_target is None: @@ -474,14 +474,14 @@ class Temp(PpmsMixin, Drivable): if self._stopped: # combine 'stopped' with current status text if status[0] == self.Status.IDLE: - status = [status[0], 'stopped'] + status = (status[0], 'stopped') else: - status = [status[0], 'stopping (%s)' % status[1]] + status = (status[0], 'stopping (%s)' % status[1]) if self._expected_target_time: # handle timeout if self.isDriving(status): if now > self._expected_target_time + self.timeout: - status = [self.Status.WARN, 'timeout while %s' % status[1]] + status = (self.Status.WARN, 'timeout while %s' % status[1]) else: self._expected_target_time = 0 self.status = status @@ -503,7 +503,7 @@ class Temp(PpmsMixin, Drivable): if abs(self.target - self.value) <= 2e-5 * target and target == self.target: return None self._status_before_change = self.status - self.status = [self.Status.BUSY, 'changed target'] + self.status = (self.Status.BUSY, 'changed target') self._last_change = time.time() self.temp.write(self, 'target', target) return Done @@ -532,7 +532,7 @@ class Temp(PpmsMixin, Drivable): if newtarget != self.target: self.log.debug('stop at %s K', newtarget) self.write_target(newtarget) - self.status = [self.status[0], 'stopping (%s)' % self.status[1]] + self.status = self.status[0], 'stopping (%s)' % self.status[1] self._stopped = True @@ -571,16 +571,16 @@ class Field(PpmsMixin, Drivable): } STATUS_MAP = { - 0: [Status.ERROR, 'unknown'], - 1: [Status.IDLE, 'persistent mode'], - 2: [Status.PREPARING, 'switch warming'], - 3: [Status.FINALIZING, 'switch cooling'], - 4: [Status.IDLE, 'driven stable'], - 5: [Status.FINALIZING, 'driven final'], - 6: [Status.RAMPING, 'charging'], - 7: [Status.RAMPING, 'discharging'], - 8: [Status.ERROR, 'current error'], - 15: [Status.ERROR, 'general failure'], + 0: (Status.ERROR, 'unknown'), + 1: (Status.IDLE, 'persistent mode'), + 2: (Status.PREPARING, 'switch warming'), + 3: (Status.FINALIZING, 'switch cooling'), + 4: (Status.IDLE, 'driven stable'), + 5: (Status.FINALIZING, 'driven final'), + 6: (Status.RAMPING, 'charging'), + 7: (Status.RAMPING, 'discharging'), + 8: (Status.ERROR, 'current error'), + 15: (Status.ERROR, 'general failure'), } channel = 'field' @@ -698,12 +698,12 @@ class Position(PpmsMixin, Drivable): Override(visibility=3), } STATUS_MAP = { - 0: [Status.ERROR, 'unknown'], - 1: [Status.IDLE, 'at target'], - 5: [Status.BUSY, 'moving'], - 8: [Status.IDLE, 'at limit'], - 9: [Status.IDLE, 'at index'], - 15: [Status.ERROR, 'general failure'], + 0: (Status.ERROR, 'unknown'), + 1: (Status.IDLE, 'at target'), + 5: (Status.BUSY, 'moving'), + 8: (Status.IDLE, 'at limit'), + 9: (Status.IDLE, 'at index'), + 15: (Status.ERROR, 'general failure'), } channel = 'position' @@ -714,10 +714,10 @@ class Position(PpmsMixin, Drivable): def update_value_status(self, value, packed_status): """update value and status""" if not self.enabled: - self.status = [self.Status.DISABLED, 'disabled'] + self.status = (self.Status.DISABLED, 'disabled') return if value is None: - self.status = [self.Status.ERROR, 'invalid value'] + self.status = (self.Status.ERROR, 'invalid value') return self.value = value status = self.STATUS_MAP[(packed_status >> 12) & 0xf] @@ -774,5 +774,5 @@ class Position(PpmsMixin, Drivable): if newtarget != self.target: self.log.debug('stop at %s T', newtarget) self.write_target(newtarget) - self.status = [self.status[0], 'stopping (%s)' % self.status[1]] + self.status = (self.status[0], 'stopping (%s)' % self.status[1]) self._stopped = True diff --git a/test/test_datatypes.py b/test/test_datatypes.py index a317232..dddab7a 100644 --- a/test/test_datatypes.py +++ b/test/test_datatypes.py @@ -381,10 +381,10 @@ def test_ArrayOf(): with pytest.raises(ValueError): dt('av') - assert dt([1, 2, 3]) == [1, 2, 3] + assert dt([1, 2, 3]) == (1, 2, 3) assert dt.export_value([1, 2, 3]) == [1, 2, 3] - assert dt.import_value([1, 2, 3]) == [1, 2, 3] + assert dt.import_value([1, 2, 3]) == (1, 2, 3) assert dt.format_value([1,2,3]) == '[1, 2, 3] Z' assert dt.format_value([1,2,3], '') == '[1, 2, 3]' @@ -419,10 +419,10 @@ def test_TupleOf(): with pytest.raises(ValueError): dt([99, 'X']) - assert dt([1, True]) == [1, True] + assert dt([1, True]) == (1, True) assert dt.export_value([1, True]) == [1, True] - assert dt.import_value([1, True]) == [1, True] + assert dt.import_value([1, True]) == (1, True) assert dt.format_value([3,0]) == "(3, False)" diff --git a/test/test_modules.py b/test/test_modules.py index 3c96779..900eae6 100644 --- a/test/test_modules.py +++ b/test/test_modules.py @@ -156,7 +156,7 @@ def test_ModuleMeta(): # check for inital updates working properly o1 = Newclass1('o1', logger, {'.description':''}, srv) - expectedBeforeStart = {'target': 0.0, 'status': [Drivable.Status.IDLE, ''], + expectedBeforeStart = {'target': 0.0, 'status': (Drivable.Status.IDLE, ''), 'param1': False, 'param2': 1.0, 'a1': 0.0, 'a2': True, 'pollinterval': 5.0, 'value': 'first'} assert updates.pop('o1') == expectedBeforeStart @@ -165,7 +165,7 @@ def test_ModuleMeta(): o1.startModule(event.set) event.wait() # should contain polled values - expectedAfterStart = {'status': [Drivable.Status.IDLE, ''], + expectedAfterStart = {'status': (Drivable.Status.IDLE, ''), 'value': 'second'} assert updates.pop('o1') == expectedAfterStart