linse/frappy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Manually convert most remaining format statements

Browse Source 
%d accepts floats and other things, so manual fixes are needed after
conversion.

After flynt -ll 2000 --aggressive, each was manually checked if the
casts with int() are needed.

Two statements are still missing in ls370res

Change-Id: I2651ddbe60695aa19582882a97d0f71bcb05c1ef
Reviewed-on: https://forge.frm2.tum.de/review/c/secop/frappy/+/30901
Tested-by: Jenkins Automated Tests <pedersen+jenkins@frm2.tum.de>
Reviewed-by: Alexander Zaft <a.zaft@fz-juelich.de>
This commit is contained in:
Alexander Zaft 2023-04-14 09:43:08 +02:00 committed by Markus Zolliker
parent 42c5e97122
commit a334cc4f0a
15 changed files with 58 additions and 64 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

22
frappy/datatypes.py
View File

@ -324,8 +324,7 @@ class IntRange(DataType):
# check the limits
if self.min <= value <= self.max:
return value
raise RangeError('%r must be between %d and %d' %
(value, self.min, self.max))
raise RangeError(f'{value!r} must be between {self.min} and {self.max}')
def __repr__(self):
args = (self.min, self.max)
@ -348,7 +347,7 @@ class IntRange(DataType):
return self(value)
def format_value(self, value, unit=None):
return '%d' % value
return f'{value}'
def compatible(self, other):
if isinstance(other, (IntRange, FloatRange, ScaledInteger)):
@ -564,7 +563,7 @@ class BLOBType(DataType):
return self.get_info(type='blob')
def __repr__(self):
return 'BLOBType(%d, %d)' % (self.minbytes, self.maxbytes)
return f'BLOBType({self.minbytes}, {self.maxbytes})'
def __call__(self, value):
"""accepts bytes only"""
@ -573,10 +572,10 @@ class BLOBType(DataType):
size = len(value)
if size < self.minbytes:
raise RangeError(
'%r must be at least %d bytes long!' % (value, self.minbytes))
f'{value!r} must be at least {self.minbytes} bytes long!')
if size > self.maxbytes:
raise RangeError(
'%r must be at most %d bytes long!' % (value, self.maxbytes))
f'{value!r} must be at most {self.maxbytes} bytes long!')
return value
def export_value(self, value):
@ -643,10 +642,10 @@ class StringType(DataType):
size = len(value)
if size < self.minchars:
raise RangeError(
'%s must be at least %d chars long!' % (shortrepr(value), self.minchars))
f'{shortrepr(value)} must be at least {self.minchars} chars long!')
if size > self.maxchars:
raise RangeError(
'%s must be at most %d chars long!' % (shortrepr(value), self.maxchars))
f'{shortrepr(value)} must be at most {self.maxchars} chars long!')
if '\0' in value:
raise RangeError(
'Strings are not allowed to embed a \\0! Use a Blob instead!')
@ -690,7 +689,7 @@ class TextType(StringType):
def __repr__(self):
if self.maxchars == UNLIMITED:
return 'TextType()'
return 'TextType(%d)' % self.maxchars
return f'TextType({self.maxchars})'
def copy(self):
# DataType.copy will not work, because it is exported as 'string'
@ -799,11 +798,10 @@ class ArrayOf(DataType):
# check number of elements
if self.minlen is not None and len(value) < self.minlen:
raise RangeError(
'array too small, needs at least %d elements!' %
self.minlen)
f'array too small, needs at least {self.minlen} elements!')
if self.maxlen is not None and len(value) > self.maxlen:
raise RangeError(
'array too big, holds at most %d elements!' % self.maxlen)
f'array too big, holds at most {self.maxlen} elements!')
except TypeError:
raise WrongTypeError(f'{type(value).__name__} can not be converted to ArrayOf DataType!') from None

2
frappy/gui/cfg_editor/mainwindow.py
View File

@ -54,7 +54,7 @@ class MainWindow(QMainWindow):
self.new_files = 0
def on_actionNew(self):
name = 'unnamed_%i.cfg' % self.new_files if self.new_files else \
name = f'unnamed_{self.new_files}.cfg' if self.new_files else \
'unnamed.cfg'
self.new_node(name)
self.new_files += 1

3
frappy/gui/cfg_editor/tree_widget_item.py
View File

@ -72,8 +72,7 @@ class TreeWidgetItem(QTreeWidgetItem):
def duplicate(self):
self.duplicates += 1
duplicate = TreeWidgetItem(self.kind, '%s_%i' % (self.name,
self.duplicates), self.get_value(),
duplicate = TreeWidgetItem(self.kind, f'{self.name}_{self.duplicates}', self.get_value(),
self.class_object)
self.parent().addChild(duplicate)
for i in range(self.childCount()):

5
frappy/gui/tabwidget.py
View File

@ -180,9 +180,8 @@ class TearOffTabWidget(QTabWidget):
self.detached = detached
def __repr__(self):
return 'index %d, widget %r, title %s, visible %r, detached %r' % (
self.index, self.widget, self.title, self.visible,
self.detached)
return f'index {self.index}, widget {self.widget!r}, title {self.title},' \
f' visible {self.visible!r}, detached {self.detached!r}'
def __init__(self, item, window, menuwindow, parent=None):
QTabWidget.__init__(self, parent)

5
frappy/lib/__init__.py
View File

@ -245,8 +245,7 @@ def formatExtendedTraceback(exc_info=None):
while tb is not None:
frame = tb.tb_frame
filename = frame.f_code.co_filename
item = ' File "%s", line %d, in %s\n' % (filename, tb.tb_lineno,
frame.f_code.co_name)
item = f' File "{filename}", line {tb.tb_lineno}, in {frame.f_code.co_name}\n'
linecache.checkcache(filename)
line = linecache.getline(filename, tb.tb_lineno, frame.f_globals)
if line:
@ -267,7 +266,7 @@ def formatExtendedStack(level=1):
co = f.f_code
filename = co.co_filename
name = co.co_name
item = ' File "%s", line %d, in %s\n' % (filename, lineno, name)
item = f' File "{filename}", line {lineno}, in {name}\n'
linecache.checkcache(filename)
line = linecache.getline(filename, lineno, f.f_globals)
if line:

6
frappy/lib/enum.py
View File

@ -104,7 +104,7 @@ class EnumMember:
# be human readable (for debugging)
def __repr__(self):
return '<%s%s (%d)>' % (self.enum.name + '.' if self.enum.name else '', self.name, self.value)
return f"<{self.enum.name + '.' if self.enum.name else ''}{self.name} ({self.value})>"
def __bool__(self):
return bool(self.value)
@ -285,9 +285,9 @@ class Enum(dict):
# check for duplicates
if self.get(k, v) != v:
raise TypeError('%s=%d conflicts with %s=%d' % (k, v, k, self[k]))
raise TypeError(f'{k}={v} conflicts with {k}={self[k]}')
if self.get(v, k) != k:
raise TypeError('%s=%d conflicts with %s=%d' % (k, v, self[v].name, v))
raise TypeError(f'{k}={v} conflicts with {self[v].name}={v}')
# remember it
self[v] = self[k] = EnumMember(self, k, v)

2
frappy/lib/pidfile.py
View File

@ -46,7 +46,7 @@ def remove_pidfile(pidfile):
def write_pidfile(pidfile, pid):
"""write the given pid to the given pidfile"""
with open(pidfile, 'w', encoding='utf-8') as f:
f.write('%d\n' % pid)
f.write(f'{pid}\n')
atexit.register(remove_pidfile, pidfile)

4
frappy/protocol/interface/tcp.py
View File

@ -94,7 +94,7 @@ class TCPRequestHandler(socketserver.BaseRequestHandler):
if origin in (HELP, b''): # empty string -> send help message
for idx, line in enumerate(HelpMessage.splitlines()):
# not sending HELPREPLY here, as there should be only one reply for every request
self.send_reply(('_', '%d' % (idx+1), line))
self.send_reply(('_', f'{idx + 1}', line))
# ident matches request
self.send_reply((HELPREPLY, None, None))
continue
@ -181,7 +181,7 @@ class TCPServer(socketserver.ThreadingTCPServer):
# for cfg-editor
configurables = {
'uri': Property('hostname or ip address for binding', StringType(),
default='tcp://%d' % DEF_PORT, export=False),
default=f'tcp://{DEF_PORT}', export=False),
'detailed_errors': Property('Flag to enable detailed Errorreporting.', BoolType(),
default=False, export=False),
}

2
frappy/protocol/router.py
View File

@ -97,7 +97,7 @@ class Router(frappy.protocol.dispatcher.Dispatcher):
self.nodes = [SecopClient(uri, logger.getChild('routed'), self)]
self.singlenode = self.nodes[0]
else:
self.nodes = [SecopClient(uri, logger.getChild('routed%d' % i), self) for i, uri in enumerate(uris)]
self.nodes = [SecopClient(uri, logger.getChild(f'routed{i}'), self) for i, uri in enumerate(uris)]
# register callbacks
for node in self.nodes:
node.register_callback(None, node.updateEvent, node.descriptiveDataChange, node.nodeStateChange)

6
frappy_demo/lscsim.py
View File

@ -59,11 +59,11 @@ class Ls370Sim(Communicator):
def simulate(self):
# not really a simulation. just for testing RDGST
for channel in self.CHANNELS:
_, _, _, _, excoff = self._data['RDGRNG?%d' % channel].split(',')
_, _, _, _, excoff = self._data[f'RDGRNG?{channel}'].split(',')
if excoff == '1':
self._data['RDGST?%d' % channel] = '6'
self._data[f'RDGST?{channel}'] = '6'
else:
self._data['RDGST?%d' % channel] = '0'
self._data[f'RDGST?{channel}'] = '0'
def communicate(self, command):
self.comLog(f'> {command}')

2
frappy_psi/ccu4.py
View File

@ -96,4 +96,4 @@ class HeLevel(HasIO, Readable):
return self.query('hf')
def write_sample_rate(self, value):
return self.query('hf=%d' % value)
return self.query(f'hf={int(value)}')

22
frappy_psi/ls370res.py
View File

@ -64,7 +64,7 @@ class Switcher(HasIO, ChannelSwitcher):
# disable unused channels
for ch in range(1, 16):
if ch not in self._channels:
self.communicate('INSET %d,0,0,0,0,0;INSET?%d' % (ch, ch))
self.communicate(f'INSET {ch},0,0,0,0,0;INSET?{ch}')
channelno, autoscan = literal_eval(self.communicate('SCAN?'))
if channelno in self._channels and self._channels[channelno].enabled:
if not autoscan:
@ -74,7 +74,7 @@ class Switcher(HasIO, ChannelSwitcher):
if channelno is None:
self.status = 'ERROR', 'no enabled channel'
return
self.communicate('SCAN %d,0;SCAN?' % channelno)
self.communicate(f'SCAN {channelno},0;SCAN?')
def doPoll(self):
"""poll buttons
@ -87,7 +87,7 @@ class Switcher(HasIO, ChannelSwitcher):
if autoscan:
# pressed autoscan button: switch off HW autoscan and toggle soft autoscan
self.autoscan = not self.autoscan
self.communicate('SCAN %d,0;SCAN?' % self.value)
self.communicate(f'SCAN {self.value},0;SCAN?')
if channelno != self.value:
# channel changed by keyboard, do not yet return new channel
self.write_target(channelno)
@ -135,7 +135,7 @@ class Switcher(HasIO, ChannelSwitcher):
self.measure_delay = chan.dwell
def set_active_channel(self, chan):
self.communicate('SCAN %d,0;SCAN?' % chan.channel)
self.communicate(f'SCAN {chan.channel},0;SCAN?')
chan._last_range_change = time.monotonic()
self.set_delays(chan)
@ -187,7 +187,7 @@ class ResChannel(Channel):
return [self.Status.DISABLED, 'disabled']
if not self.channel == self.switcher.value == self.switcher.target:
return Done
result = int(self.communicate('RDGST?%d' % self.channel))
result = int(self.communicate(f'RDGST?{self.channel}'))
result &= 0x37 # mask T_OVER and T_UNDER (change this when implementing temperatures instead of resistivities)
statustext = ' '.join(formatStatusBits(result, STATUS_BIT_LABELS))
if statustext:
@ -199,7 +199,7 @@ class ResChannel(Channel):
now = time.monotonic()
if now + 0.5 < max(self._last_range_change, self.switcher._start_switch) + self.pause:
return None
result = self.communicate('RDGR?%d' % self.channel)
result = self.communicate(f'RDGR?{self.channel}')
result = float(result)
if self.autorange:
self.fix_autorange()
@ -236,7 +236,7 @@ class ResChannel(Channel):
@CommonReadHandler(rdgrng_params)
def read_rdgrng(self):
iscur, exc, rng, autorange, excoff = literal_eval(
self.communicate('RDGRNG?%d' % self.channel))
self.communicate(f'RDGRNG?{self.channel}'))
self._prev_rdgrng = iscur, exc
if autorange: # pressed autorange button
if not self._toggle_autorange:
@ -283,7 +283,7 @@ class ResChannel(Channel):
def read_inset(self):
# ignore curve no and temperature coefficient
enabled, dwell, pause, _, _ = literal_eval(
self.communicate('INSET?%d' % self.channel))
self.communicate(f'INSET?{self.channel}'))
self.enabled = enabled
self.dwell = dwell
self.pause = pause
@ -291,7 +291,7 @@ class ResChannel(Channel):
@CommonWriteHandler(inset_params)
def write_inset(self, change):
_, _, _, curve, tempco = literal_eval(
self.communicate('INSET?%d' % self.channel))
self.communicate(f'INSET?{self.channel}'))
self.enabled, self.dwell, self.pause, _, _ = literal_eval(
self.communicate('INSET %d,%d,%d,%d,%d,%d;INSET?%d' % (
self.channel, change['enabled'], change['dwell'], change['pause'], curve, tempco,
@ -303,14 +303,14 @@ class ResChannel(Channel):
self.switcher.set_delays(self)
def read_filter(self):
on, settle, _ = literal_eval(self.communicate('FILTER?%d' % self.channel))
on, settle, _ = literal_eval(self.communicate(f'FILTER?{self.channel}'))
return settle if on else 0
def write_filter(self, value):
on = 1 if value else 0
value = max(1, value)
on, settle, _ = literal_eval(self.communicate(
'FILTER %d,%d,%g,80;FILTER?%d' % (self.channel, on, value, self.channel)))
f'FILTER {self.channel},{on},{value:g},80;FILTER?{self.channel}'))
if not on:
settle = 0
return settle

30
frappy_psi/ppms.py
View File

@ -94,7 +94,7 @@ class Main(Communicator):
if channel.enabled:
mask |= 1 << self._channel_to_index.get(channelname, 0)
# send, read and convert to floats and ints
data = self.communicate('GETDAT? %d' % mask)
data = self.communicate(f'GETDAT? {mask}')
reply = data.split(',')
mask = int(reply.pop(0))
reply.pop(0) # pop timestamp
@ -204,7 +204,7 @@ class DriverChannel(Channel):
@CommonReadHandler(param_names)
def read_params(self):
no, self.current, self.powerlimit = literal_eval(
self.communicate('DRVOUT? %d' % self.no))
self.communicate(f'DRVOUT? {self.no}'))
if self.no != no:
raise HardwareError('DRVOUT command: channel number in reply does not match')
@ -215,7 +215,7 @@ class DriverChannel(Channel):
:param values: a dict like object containing the parameters to be written
"""
self.read_params() # make sure parameters are up to date
self.comm_write('DRVOUT %(no)d,%(current)g,%(powerlimit)g' % values)
self.comm_write(f"DRVOUT {values['no']},{values['current']:g},{values['powerlimit']:g}")
self.read_params() # read back
@ -238,7 +238,7 @@ class BridgeChannel(Channel):
@CommonReadHandler(param_names)
def read_params(self):
no, excitation, powerlimit, self.dcflag, self.readingmode, voltagelimit = literal_eval(
self.communicate('BRIDGE? %d' % self.no))
self.communicate(f'BRIDGE? {self.no}'))
if self.no != no:
raise HardwareError('DRVOUT command: channel number in reply does not match')
self.enabled = excitation != 0 and powerlimit != 0 and voltagelimit != 0
@ -260,8 +260,9 @@ class BridgeChannel(Channel):
values['excitation'] = 0
values['powerlimit'] = 0
values['voltagelimit'] = 0
self.comm_write('BRIDGE %(no)d,%(enabled)g,%(powerlimit)g,%(dcflag)d,'
'%(readingmode)d,%(voltagelimit)g' % values)
self.comm_write(f"BRIDGE {values['no']},{values['enabled']:g},' \
f'{values['powerlimit']:g},{int(values['dcflag'])},{int(values['readingmode'])},' \
f'{values['voltagelimit']:g}")
self.read_params() # read back
@ -322,7 +323,7 @@ class Chamber(PpmsDrivable):
self.status = self.status_map[status_code]
else:
self.value = self.value_map['unknown']
self.status = (StatusType.ERROR, 'unknown status code %d' % status_code)
self.status = (StatusType.ERROR, f'unknown status code {status_code}')
def read_target(self):
opcode = int(self.communicate('CHAMBER?'))
@ -332,7 +333,7 @@ class Chamber(PpmsDrivable):
if value == self.target.noop:
return self.target.noop
opcode = self.name2opcode[self.target.enum(value).name]
assert self.communicate('CHAMBER %d' % opcode) == 'OK'
assert self.communicate(f'CHAMBER {opcode}') == 'OK'
return self.read_target()
@ -408,7 +409,7 @@ class Temp(PpmsDrivable):
self.calc_expected(setpoint, ramp)
self.log.debug(
'change_temp v %r s %r r %r w %r l %r', self.value, setpoint, ramp, wait_at10, ramp_at_limit)
self.comm_write('TEMP %g,%g,%d' % (setpoint, ramp, approachmode))
self.comm_write(f'TEMP {setpoint:g},{ramp:g},{int(approachmode)}')
self.read_params()
def update_value_status(self, value, packed_status):
@ -417,7 +418,7 @@ class Temp(PpmsDrivable):
return
self.value = value
status_code = packed_status & 0xf
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, 'unknown status code %d' % status_code))
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, f'unknown status code {status_code}'))
now = time.time()
if value > 11:
# when starting from T > 50, this will be 15 min.
@ -542,8 +543,7 @@ class Field(PpmsDrivable):
self.ramp = ramp * 6e-3
def _write_params(self, target, ramp, approachmode, persistentmode):
self.comm_write('FIELD %g,%g,%d,%d' % (
target * 1e+4, ramp / 6e-3, approachmode, persistentmode))
self.comm_write(f'FIELD {target * 10000.0:g},{ramp / 0.006:g},{int(approachmode)},{int(persistentmode)}')
self.read_params()
def update_value_status(self, value, packed_status):
@ -552,7 +552,7 @@ class Field(PpmsDrivable):
return
self.value = round(value * 1e-4, 7)
status_code = (packed_status >> 4) & 0xf
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, 'unknown status code %d' % status_code))
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, f'unknown status code {status_code}'))
now = time.time()
if self._last_change: # there was a change, which is not yet confirmed by hw
if status_code == 1: # persistent mode
@ -661,7 +661,7 @@ class Position(PpmsDrivable):
def _write_params(self, target, speed):
speed = int(round(min(14, max(0, 15 - speed / 0.8)), 0))
self.comm_write('MOVE %g,%d,%d' % (target, 0, speed))
self.comm_write(f'MOVE {target:g},{0},{speed}')
return self.read_params()
def update_value_status(self, value, packed_status):
@ -673,7 +673,7 @@ class Position(PpmsDrivable):
return
self.value = value
status_code = (packed_status >> 12) & 0xf
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, 'unknown status code %d' % status_code))
status = self.STATUS_MAP.get(status_code, (StatusType.ERROR, f'unknown status code {status_code}'))
if self._last_change: # there was a change, which is not yet confirmed by hw
now = time.time()
if now > self._last_change + 5:

2
frappy_psi/ppmssim.py
View File

@ -198,7 +198,7 @@ class PpmsSim:
def getdat(self, mask):
mask = int(mask) & 0x8000ff # all channels up to i2 plus ts
output = ['%d' % mask, f'{time.time() - self.start:.2f}']
output = [f'{mask}', f'{time.time() - self.start:.2f}']
for i, chan in self.CHANNELS.items():
if (1 << i) & mask:
output.append(f"{getattr(self, chan):.7g}")

9
frappy_psi/trinamic.py
View File

@ -128,7 +128,7 @@ class Motor(PersistentMixin, HasIO, Drivable):
power_down_delay = writable('', FloatRange(0, 60., unit='sec', fmtstr='%.2f'),
default=0.1, group='motorparam')
baudrate = Parameter('', EnumType({'%d' % v: i for i, v in enumerate(BAUDRATES)}),
baudrate = Parameter('', EnumType({f'{v}': i for i, v in enumerate(BAUDRATES)}),
readonly=False, default=0, visibility=3, group='more')
pollinterval = Parameter(group='more')
@ -154,7 +154,7 @@ class Motor(PersistentMixin, HasIO, Drivable):
baudrate = getattr(self.io._conn.connection, 'baudrate', None)
if baudrate:
if baudrate not in BAUDRATES:
raise CommunicationFailedError('unsupported baud rate: %d' % baudrate)
raise CommunicationFailedError(f'unsupported baud rate: {int(baudrate)}')
self.io.timeout = 0.03 + 200 / baudrate
exc = None
@ -178,7 +178,7 @@ class Motor(PersistentMixin, HasIO, Drivable):
if status != 100:
self.log.warning('bad status from cmd %r %s: %d', cmd, adr, status)
if radr != 2 or modadr != self.address or cmd != rcmd:
raise CommunicationFailedError('bad reply %r to command %s %d' % (reply, cmd, adr))
raise CommunicationFailedError(f'bad reply {reply!r} to command {cmd} {adr}')
return result
def startModule(self, start_events):
@ -229,8 +229,7 @@ class Motor(PersistentMixin, HasIO, Drivable):
if readback:
result = self.comm(GET_AXIS_PAR, adr)
if result != rawvalue:
raise HardwareError('result for adr=%d scale=%g does not match %g != %g'
% (adr, scale, result * scale, value))
raise HardwareError(f'result for adr={adr} scale={scale:g} does not match {result * scale:g} != {value:g}')
return result * scale
return rawvalue * scale