refactor: simplify exception handling

examples/async.py

@@ -79,6 +79,27 @@ async def get_info(worker) -> None:
     print(f"{'Firmware':<15}: {firmware[0]}")
 
 
+async def run(worker) -> None:
+    await worker.init_async()
+
+    await get_info(worker)
+    await get_network_info(worker)
+
+    await send_message_async(worker, "6700", "BAL")
+
+    # Just a busy waiting for event
+    # We need to keep communication with phone to get notifications
+    print("Press Ctrl+C to interrupt")
+    while 1:
+        try:
+            signal = await worker.get_signal_quality_async()
+            print(f"Signal is at {signal['SignalPercent']:d}%")
+        except Exception as e:  # ruff: ignore[blind-except]
+            print(f"Exception reading signal: {e}")
+
+        await asyncio.sleep(10)
+
+
 async def main() -> None:
     gammu.SetDebugFile(sys.stderr)
     gammu.SetDebugLevel("textall")
@@ -88,25 +109,7 @@ async def main() -> None:
     worker.configure(config)
 
     try:
-        await worker.init_async()
-
-        await get_info(worker)
-        await get_network_info(worker)
-
-        await send_message_async(worker, "6700", "BAL")
-
-        # Just a busy waiting for event
-        # We need to keep communication with phone to get notifications
-        print("Press Ctrl+C to interrupt")
-        while 1:
-            try:
-                signal = await worker.get_signal_quality_async()
-                print(f"Signal is at {signal['SignalPercent']:d}%")
-            except Exception as e:  # ruff: ignore[blind-except]
-                print(f"Exception reading signal: {e}")
-
-            await asyncio.sleep(10)
-
+        await run(worker)
     except Exception as e:  # ruff: ignore[blind-except]
         print("Exception:")
         print(e)

examples/filesystem_test.py

@@ -53,6 +53,16 @@
 import gammu
 
 
+def print_root_folders(state_machine) -> None:
+    file_obj = state_machine.GetNextRootFolder("")
+    while 1:
+        print(f"{file_obj['ID_FullName']} - {file_obj['Name']}")
+        try:
+            file_obj = state_machine.GetNextRootFolder(file_obj["ID_FullName"])
+        except gammu.ERR_EMPTY:
+            break
+
+
 def main() -> None:  # ruff: ignore[too-many-branches, too-many-statements, complex-structure]
     parser = argparse.ArgumentParser(usage="usage: %(prog)s [options]")
 
@@ -164,13 +174,7 @@ def main() -> None:  # ruff: ignore[too-many-branches, too-many-statements, comp
     # Check GetNextRootFolder
     print("\n\nExpectation: Root Folder List")
     try:
-        file_obj = state_machine.GetNextRootFolder("")
-        while 1:
-            print(f"{file_obj['ID_FullName']} - {file_obj['Name']}")
-            try:
-                file_obj = state_machine.GetNextRootFolder(file_obj["ID_FullName"])
-            except gammu.ERR_EMPTY:
-                break
+        print_root_folders(state_machine)
     except gammu.ERR_NOTSUPPORTED:
         print("Not supported...")
 

examples/getallsms.py

@@ -24,6 +24,12 @@
 import gammu
 
 
+def get_next_sms(state_machine, previous):
+    if previous is None:
+        return state_machine.GetNextSMS(Start=True, Folder=0)
+    return state_machine.GetNextSMS(Location=previous[0]["Location"], Folder=0)
+
+
 def main() -> None:
     state_machine = gammu.StateMachine()
     state_machine.ReadConfig()
@@ -33,27 +39,23 @@ def main() -> None:
 
     remain = status["SIMUsed"] + status["PhoneUsed"] + status["TemplatesUsed"]
 
-    start = True
-
-    try:
-        while remain > 0:
-            if start:
-                sms = state_machine.GetNextSMS(Start=True, Folder=0)
-                start = False
-            else:
-                sms = state_machine.GetNextSMS(Location=sms[0]["Location"], Folder=0)
-            remain -= len(sms)
-
-            for m in sms:
-                print()
-                print(f"{'Number':<15}: {m['Number']}")
-                print(f"{'Date':<15}: {m['DateTime']!s}")
-                print(f"{'State':<15}: {m['State']}")
-                print(f"\n{m['Text']}")
-    except gammu.ERR_EMPTY:
-        # This error is raised when we've reached last entry
-        # It can happen when reported status does not match real counts
-        print("Failed to read all messages!")
+    sms = None
+    while remain > 0:
+        try:
+            sms = get_next_sms(state_machine, sms)
+        except gammu.ERR_EMPTY:
+            # This error is raised when we've reached last entry
+            # It can happen when reported status does not match real counts
+            print("Failed to read all messages!")
+            break
+        remain -= len(sms)
+
+        for message in sms:
+            print()
+            print(f"{'Number':<15}: {message['Number']}")
+            print(f"{'Date':<15}: {message['DateTime']!s}")
+            print(f"{'State':<15}: {message['State']}")
+            print(f"\n{message['Text']}")
 
 
 if __name__ == "__main__":

examples/getallsms_decode.py

@@ -23,6 +23,13 @@
 
 import gammu
 
+
+def get_next_sms(state_machine, previous):
+    if previous is None:
+        return state_machine.GetNextSMS(Start=True, Folder=0)
+    return state_machine.GetNextSMS(Location=previous[0]["Location"], Folder=0)
+
+
 state_machine = gammu.StateMachine()
 state_machine.ReadConfig()
 state_machine.Init()
@@ -32,21 +39,18 @@
 remain = status["SIMUsed"] + status["PhoneUsed"] + status["TemplatesUsed"]
 
 sms = []
-start = True
+current_sms = None
 
-try:
-    while remain > 0:
-        if start:
-            cursms = state_machine.GetNextSMS(Start=True, Folder=0)
-            start = False
-        else:
-            cursms = state_machine.GetNextSMS(Location=cursms[0]["Location"], Folder=0)
-        remain -= len(cursms)
-        sms.append(cursms)
-except gammu.ERR_EMPTY:
-    # This error is raised when we've reached last entry
-    # It can happen when reported status does not match real counts
-    print("Failed to read all messages!")
+while remain > 0:
+    try:
+        current_sms = get_next_sms(state_machine, current_sms)
+    except gammu.ERR_EMPTY:
+        # This error is raised when we've reached last entry
+        # It can happen when reported status does not match real counts
+        print("Failed to read all messages!")
+        break
+    remain -= len(current_sms)
+    sms.append(current_sms)
 
 data = gammu.LinkSMS(sms)
 

examples/savesmspercontact.py

@@ -69,73 +69,74 @@ def saveSMS(mysms, all_contacts) -> None:
         handle.write("\n")
 
 
+def getNextMemory(state_machine, previous):
+    if previous is None:
+        return state_machine.GetNextMemory(Start=True, Type="SM")
+    return state_machine.GetNextMemory(Location=previous["Location"], Type="SM")
+
+
 def getContacts(state_machine):
     # Get all contacts
     remaining = state_machine.GetMemoryStatus(Type="SM")["Used"]
     contacts = {}
 
-    start = True
+    memory_entry = None
+
+    while remaining > 0:
+        first = memory_entry is None
+        try:
+            memory_entry = getNextMemory(state_machine, memory_entry)
+        except gammu.ERR_EMPTY:
+            # error is raised if memory is empty (this induces wrong reported
+            # memory status)
+            print("Failed to read contacts!")
+            break
+        if not first:
+            remaining -= 1
 
-    try:
-        while remaining > 0:
-            if start:
-                memory_entry = state_machine.GetNextMemory(Start=True, Type="SM")
-                start = False
+        numbers = []
+        name = "Unknown"
+        for entry in memory_entry["Entries"]:
+            if entry["Type"] == "Text_FirstName":
+                name = entry["Value"]
             else:
-                memory_entry = state_machine.GetNextMemory(
-                    Location=memory_entry["Location"], Type="SM"
-                )
-                remaining -= 1
-
-            numbers = []
-            name = "Unknown"
-            for entry in memory_entry["Entries"]:
-                if entry["Type"] == "Text_FirstName":
-                    name = entry["Value"]
-                else:
-                    numbers.append(getInternationalizedNumber(entry["Value"]))
-
-            for number in numbers:
-                contacts[number] = name
-
-    except gammu.ERR_EMPTY:
-        # error is raised if memory is empty (this induces wrong reported
-        # memory status)
-        print("Failed to read contacts!")
+                numbers.append(getInternationalizedNumber(entry["Value"]))
+
+        for number in numbers:
+            contacts[number] = name
 
     return contacts
 
 
+def getNextSMS(state_machine, previous):
+    if previous is None:
+        return state_machine.GetNextSMS(Folder=0, Start=True)
+    return state_machine.GetNextSMS(Folder=0, Location=previous[0]["Location"])
+
+
 def getAndDeleteAllSMS(state_machine):
     # Read SMS memory status ...
     memory = state_machine.GetSMSStatus()
     # ... and calculate number of messages
     remaining = memory["SIMUsed"] + memory["PhoneUsed"]
 
     # Get all sms
-    start = True
     entries = []
+    entry = None
 
-    try:
-        while remaining > 0:
-            if start:
-                entry = state_machine.GetNextSMS(Folder=0, Start=True)
-                start = False
-            else:
-                entry = state_machine.GetNextSMS(
-                    Folder=0, Location=entry[0]["Location"]
-                )
-
+    while remaining > 0:
+        try:
+            entry = getNextSMS(state_machine, entry)
             remaining -= 1
             entries.append(entry)
 
             # delete retrieved sms
             state_machine.DeleteSMS(Folder=0, Location=entry[0]["Location"])
-
-    except gammu.ERR_EMPTY:
-        # error is raised if memory is empty (this induces wrong reported
-        # memory status)
-        print("Failed to read messages!")
+        except gammu.ERR_EMPTY:
+            # error is raised if memory is empty (this induces wrong reported
+            # memory status)
+            print("Failed to read messages!")
+            break
 
     # Link all SMS when there are concatenated messages
     return gammu.LinkSMS(entries)

gammu/asyncworker.py

@@ -36,12 +36,7 @@ def _do_command(self, future, cmd, params, percentage=100) -> None:
         func = getattr(self._sm, cmd)
         result = None
         try:
-            if params is None:
-                result = func()
-            elif isinstance(params, dict):
-                result = func(**params)
-            else:
-                result = func(*params)
+            result = gammu.worker._execute_command(func, params)
         except gammu.GSMError as info:
             errcode = info.args[0]["Code"]
             error = gammu.ErrorNumbers[errcode]