Skip to content

Commit

Permalink
comm_sbs_bqctrl: Make main() usable as library call for external tools
Browse files Browse the repository at this point in the history 
Allow caller to pass args to main().
Cache parsed battery drivers to reduce overhead on multiple calls.
Make action parameter required.
Adjust tests for the main(argv) change.

---------

Co-authored-by: _AT_ <[email protected]>
  • Loading branch information
@Alter-1
Alter-1 and _AT_ authored Jul 29, 2023
1 parent c72b2a3 commit fdabd08
Show file tree
Hide file tree
Showing 2 changed files with 42 additions and 30 deletions.
52 changes: 32 additions & 20 deletions comm_sbs_bqctrl.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3448,12 +3448,16 @@ def parse_monitor_group(s):
"""
return s

driver_cache = dict()

def main():
def main(argv=sys.argv[1:]):
""" Main executable function.
Its task is to parse command line options and call a function which performs sniffing.
"""

global driver_cache

addrspace_datatypes = [ "int8", "uint8", "int16", "uint16", "int32", "uint32", "float", 'string[n]', 'byte[n]']

parser = argparse.ArgumentParser(description=__doc__.split('.')[0])
Expand Down Expand Up @@ -3492,7 +3496,7 @@ def main():
help="display version information and exit")

subparsers = parser.add_subparsers(dest='action', metavar='action',
help="action to take")
help="action to take", required=True)

subpar_info = subparsers.add_parser('info',
help=("displays information about specific command; when chip "
Expand Down Expand Up @@ -3631,7 +3635,7 @@ def main():
"(defaults to 0x{:08x} for FullAccess, otherwise to 0x{:08x})"
).format(0xffffffff,0x36720414))

po = parser.parse_args()
po = parser.parse_args(argv)

vals = {}

Expand All @@ -3649,26 +3653,34 @@ def main():
print("Opening {}".format(po.bus))
smbus_open(po.bus, po)

# Re-init global variables; then they are modified by specific chip driver, do it before detection
reset_default_driver(po)
if po.chip == CHIP_TYPE.AUTO:
po.chip = smart_battery_detect(vals, po)

# Re-init global variables; then they are modified by specific chip driver
reset_default_driver(po)
if po.chip in (CHIP_TYPE.BQ30z50, CHIP_TYPE.BQ30z55, CHIP_TYPE.BQ30z554,):
fnames = ["comm_sbs_chips/{}.py".format("BQ30z554")]
elif po.chip in (CHIP_TYPE.BQ40z50,):
fnames = ["comm_sbs_chips/{}.py".format("BQ40z50")]
if po.chip in driver_cache:
chip_file_code = driver_cache[po.chip]
exec(chip_file_code)
else:
fnames = ["comm_sbs_chips/{}.py".format(po.chip.name)]
for fname in fnames:
try:
with open(fname, "rb") as source_file:
chip_file_code = compile(source_file.read(), fname, "exec")
if (po.verbose > 0):
print("Importing {}".format(fname))
exec(chip_file_code)
except IOError:
print("Warning: Could not open chip definition file '{}'".format(fname))
if po.chip in (CHIP_TYPE.BQ30z50, CHIP_TYPE.BQ30z55, CHIP_TYPE.BQ30z554,):
fnames = ["comm_sbs_chips/{}.py".format("BQ30z554")]
elif po.chip in (CHIP_TYPE.BQ40z50,):
fnames = ["comm_sbs_chips/{}.py".format("BQ40z50")]
elif po.chip in ("SBS",): # default
pass # do nothing, already loaded with reset_default_driver(po)
else:
fnames = ["comm_sbs_chips/{}.py".format(po.chip.name)]
driver_cache[po.chip] = list()
for fname in fnames:
try:
with open(fname, "rb") as source_file:
chip_file_code = compile(source_file.read(), fname, "exec")
if (po.verbose > 0):
print("Importing {}".format(fname))
exec(chip_file_code)
driver_cache[po.chip] = chip_file_code
except IOError:
print("Warning: Could not open chip definition file '{}'".format(fname))

if po.action == 'info':
smart_battery_system_info(po.command, vals, po)
Expand Down Expand Up @@ -3728,7 +3740,7 @@ def main():
po.i32key = 0x36720414
smart_battery_system_sealing(po.sealstate, vals, po)
else:
raise NotImplementedError("Unsupported command.")
raise NotImplementedError("Unsupported or missing command.")

if not po.offline_mode:
smbus_close()
Expand Down
20 changes: 10 additions & 10 deletions tests/test_comm_sbs_bqctrl1.py
View file
Original file line number Diff line number Diff line change
Expand Up @@ -45,14 +45,14 @@ def test_comm_sbs_bqctrl_chip_detect(capsys):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-v", "--chip", "BQ30z55", "read-list"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()

# Do not provide chip - test auto-detect
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-v", "--dry-run", "info-list"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
pass

Expand All @@ -73,7 +73,7 @@ def test_comm_sbs_bqctrl_chip_info_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "--dry-run", "--chip", chip_name, "info-list"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
info_commands = list(capstdout.splitlines())
LOGGER.info("List of info commands: {}".format(', '.join(info_commands)))
Expand All @@ -85,7 +85,7 @@ def test_comm_sbs_bqctrl_chip_info_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-v", "--dry-run", "--chip", chip_name, "info", info_cmd]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
assert "Type:" in capstdout
assert "Description:" in capstdout
Expand All @@ -108,7 +108,7 @@ def test_comm_sbs_bqctrl_chip_read_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "--dry-run", "--chip", chip_name, "read-list"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
read_commands = list(capstdout.splitlines())
LOGGER.info("List of read commands: {}".format(', '.join(read_commands)))
Expand All @@ -125,7 +125,7 @@ def test_comm_sbs_bqctrl_chip_read_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-vv", "--dry-run", "--chip", chip_name, "read", read_cmd]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
pass

Expand All @@ -146,7 +146,7 @@ def test_comm_sbs_bqctrl_chip_write_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "--dry-run", "--chip", chip_name, "write-list"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
write_commands = list(capstdout.splitlines())
pytest.skip("no tests for write command as the command is unfinished")
Expand Down Expand Up @@ -176,7 +176,7 @@ def test_comm_sbs_bqctrl_chip_monitor_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-vvv", "--dry-run", "--chip", chip_name, "--short", "monitor", monitor_cmd]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
pass

Expand Down Expand Up @@ -207,7 +207,7 @@ def test_comm_sbs_bqctrl_chip_sealing_commands(capsys, chip_name, test_nth):
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-vvv", "--dry-run", "--chip", chip_name, "--short", "sealing", sealing_cmd]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
assert ".OperationStatus:" in capstdout
# TODO maybe check status value, ie assert " SEC=1" in capstdout
Expand All @@ -228,6 +228,6 @@ def test_comm_sbs_bqctrl_data_flash(capsys, chip_name, flash_start, flash_end, f
command = [os.path.join(".", "comm_sbs_bqctrl.py"), "-v", "--dry-run", "--chip", chip_name, "raw-read", "DataFlash", "0x{:X}".format(flash_offset), "string[32]"]
LOGGER.info(' '.join(command))
with patch.object(sys, 'argv', command):
comm_sbs_bqctrl_main()
comm_sbs_bqctrl_main(command[1:])
capstdout, _ = capsys.readouterr()
pass

0 comments on commit fdabd08

Please sign in to comment.