fix ipc api

Reviewed-on: #19

README.md

@@ -8,7 +8,7 @@
 
 ## Install ##
 
-`pip install qrcode sysv-ipc pycstruct brother-ql`
+`pip install qrcode brother-ql`
 
 ```
 cd /root/

layouter.py

@@ -21,10 +21,9 @@ FONT_PATH = "/usr/share/fonts"
 DEFAULT_FONT_REGULAR = "DejaVuSans.ttf"
 DEFAULT_FONT_BOLD = "DejaVuSans-Bold.ttf"
 OUTPUT_WIDTH = 696  # px
-OUTPUT_HEIGHT = 300  # px
+OUTPUT_HEIGHT = 190  # px
-TEXT_X_OFFSET = 300  # px
+TEXT_X_OFFSET = 190  # px
-QR_CODE_SIZE = 289  # px
+QR_CODE_SIZE = 179  # px
-
 # Configure logging
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
@@ -46,6 +45,20 @@ class DriveData:
     shred_duration: int
 
 
+@dataclasses.dataclass
+class DriveDataJson:
+    state: str
+    fam: str
+    name: str
+    cap: int
+    sn: str
+    poh: int
+    pc: int
+    err: int
+    time: int
+    dur: int
+
+
 @dataclasses.dataclass
 class DriveDataPrintable:
     modelfamily: str
@@ -66,8 +79,8 @@ class ReHddInfo:
 
 
 @dataclasses.dataclass
-class DriveDataJson:
+class QrDataJson:
-    drive: DriveData
+    drive: DriveDataJson
     rehdd: ReHddInfo
 
 
@@ -122,7 +135,7 @@ def format_to_printable(drive):
         cut_string(20, re.sub(r"[^a-zA-Z0-9. ]", "", drive.modelfamily), "end"),
         cut_string(20, re.sub(r"[^a-zA-Z0-9. ]", "", drive.modelname), "end"),
         cut_string(20, human_readable_capacity(drive.capacity), "end"),
-        cut_string(16, re.sub(r"[^a-zA-Z0-9.-_]", "", drive.serialnumber), "start"),
+        cut_string(20, re.sub(r"[^a-zA-Z0-9.-_]", "", drive.serialnumber), "start"),
         cut_string(30, human_readable_power_on_hours(drive.power_on_hours), "end"),
         cut_string(10, str(drive.power_cycle), "end"),
         cut_string(10, str(drive.smart_error_count), "end"),
@@ -138,40 +151,98 @@ def format_to_printable(drive):
 
 
 def draw_text(drawable, printable_data, font_regular, font_bold, font_bold_bigger):
-    """Draws the formatted text onto the image."""
+    """Draws formatted text with Cycles and Errors on one row."""
-    text_y_offset = 10
-    value_column_x_offset = 120
-    font_size = 20
 
+    line_height = 26
+    label_x = TEXT_X_OFFSET
+    value_offset = 115
+    right_field_spacing = 200  # Horizontal gap between Cycles and Errors
+    x_capacity = 520
+    y_capacity = 142
+    y_start = 4
+
+    # Serial Number
+    drawable.text((label_x, y_start), "Serial:", fill=0, font=font_bold)
     drawable.text(
-        (TEXT_X_OFFSET, text_y_offset), printable_data.serialnumber, (0), font=font_bold
+        (label_x + value_offset, y_start),
+        printable_data.serialnumber,
+        fill=0,
+        font=font_bold,
     )
-    text_y_offset += 25
 
-    fields = [
+    y1 = y_start + line_height
-        ("Family:", printable_data.modelfamily, font_regular),
+    y2 = y1 + line_height
-        ("Model:", printable_data.modelname, font_regular),
+    y3 = y2 + line_height
-        ("Hours:", printable_data.power_on_hours, font_regular),
+    y4 = y3 + line_height
-        ("Cycles:", printable_data.power_cycle, font_regular),
+    y5 = y4 + line_height
-        ("Errors:", printable_data.smart_error_count, font_regular),
+    y6 = y5 + line_height
-        ("Shred on:", printable_data.shred_timestamp, font_regular),
-        ("Duration:", printable_data.shred_duration, font_regular),
-    ]
 
-    for label, value, font in fields:
+    # Left-Aligned Fields (One per row)
-        drawable.text((TEXT_X_OFFSET, text_y_offset), label, fill=0, font=font_bold)
+    drawable.text((label_x, y1), "Family:", fill=0, font=font_bold)
-        drawable.text(
+    drawable.text(
-            (TEXT_X_OFFSET + value_column_x_offset, text_y_offset),
+        (label_x + value_offset, y1),
-            value,
+        printable_data.modelfamily,
-            fill=0,
+        fill=0,
-            font=font,
+        font=font_regular,
-        )
+    )
-        text_y_offset += 25
+
+    drawable.text((label_x, y2), "Model:", fill=0, font=font_bold)
+    drawable.text(
+        (label_x + value_offset, y2),
+        printable_data.modelname,
+        fill=0,
+        font=font_regular,
+    )
+
+    drawable.text((label_x, y3), "Hours:", fill=0, font=font_bold)
+    drawable.text(
+        (label_x + value_offset, y3),
+        printable_data.power_on_hours,
+        fill=0,
+        font=font_regular,
+    )
+
+    # Cycles and Errors on the same line
+    drawable.text((label_x, y4), "Cycles:", fill=0, font=font_bold)
+    drawable.text(
+        (label_x + value_offset, y4),
+        printable_data.power_cycle,
+        fill=0,
+        font=font_regular,
+    )
 
     drawable.text(
-        (TEXT_X_OFFSET, text_y_offset),
+        (label_x + right_field_spacing, y4), "Errors:", fill=0, font=font_bold
+    )
+    drawable.text(
+        (label_x + right_field_spacing + value_offset, y4),
+        printable_data.smart_error_count,
+        fill=0,
+        font=font_regular,
+    )
+
+    # Continue remaining fields
+    drawable.text((label_x, y5), "Shred on:", fill=0, font=font_bold)
+    drawable.text(
+        (label_x + value_offset, y5),
+        printable_data.shred_timestamp,
+        fill=0,
+        font=font_regular,
+    )
+
+    drawable.text((label_x, y6), "Duration:", fill=0, font=font_bold)
+    drawable.text(
+        (label_x + value_offset, y6),
+        printable_data.shred_duration,
+        fill=0,
+        font=font_regular,
+    )
+
+    # Capacity at the bottom
+    drawable.text(
+        (x_capacity, y_capacity),
         printable_data.capacity,
-        (0),
+        fill=0,
         font=font_bold_bigger,
     )
 
@@ -232,7 +303,21 @@ def draw_outline(drawable, margin, width, output_width, output_height):
 def generate_image(drive, rehdd_info, output_file):
     """Generates an image containing drive data and a QR code."""
     try:
-        qr_data = json.dumps(dataclasses.asdict(DriveDataJson(drive, rehdd_info)))
+
+        drive_json = DriveDataJson(
+            state=drive.drive_state,
+            fam=drive.modelfamily,
+            name=drive.modelname,
+            cap=drive.capacity,
+            sn=drive.serialnumber,
+            poh=drive.power_on_hours,
+            pc=drive.power_cycle,
+            err=drive.smart_error_count,
+            time=int(drive.shred_timestamp),
+            dur=drive.shred_duration,
+        )
+
+        qr_data = json.dumps(dataclasses.asdict(QrDataJson(drive_json, rehdd_info)))
         printable_data = format_to_printable(drive)
     except Exception as e:
         logging.error(f"Error preparing data: {e}")
@@ -243,9 +328,9 @@ def generate_image(drive, rehdd_info, output_file):
 
     font_regular = ImageFont.truetype(find_font_path(DEFAULT_FONT_REGULAR), 20)
     font_bold = ImageFont.truetype(find_font_path(DEFAULT_FONT_BOLD), 20)
-    font_bold_bigger = ImageFont.truetype(find_font_path(DEFAULT_FONT_BOLD), 60)
+    font_bold_bigger = ImageFont.truetype(find_font_path(DEFAULT_FONT_BOLD), 42)
 
-    draw_outline(draw, 1, 4, OUTPUT_WIDTH, OUTPUT_HEIGHT)
+    draw_outline(draw, 0, 3, OUTPUT_WIDTH + 1, OUTPUT_HEIGHT + 1)
     draw_text(draw, printable_data, font_regular, font_bold, font_bold_bigger)
     draw_qr_code(output_image, qr_data)
 
@@ -264,7 +349,7 @@ def main():
         drive_state="shredded",
         modelfamily='Toshiba 2.5" HDD MK..65GSSX',
         modelname="TOSHIBA MK3265GSDX",
-        capacity=343597383680,
+        capacity=343597383000,
         serialnumber="YG6742U56UDRL123456789ABCDEFGJKL",
         power_on_hours=7074,
         power_cycle=4792,
@@ -275,5 +360,6 @@ def main():
 
     generate_image(temp_drive, rehdd_info, "output.png")
 
+
 if __name__ == "__main__":
     main()

reHDDPrinter.py

@@ -3,75 +3,215 @@
 
 """ Author:                     Hendrik Schutter, localhorst@mosad.xyz
     Date of creation:           2022/11/23
-    Date of last modification:  2022/11/23
+    Date of last modification:  2025/06/15
 """
 
-import sysv_ipc 
+import ctypes
-import pycstruct
 import os
 import time
+import signal
+import argparse
+import warnings
+import logging
+from PIL import Image
 from brother_ql.brother_ql_create import create_label
 from brother_ql.raster import BrotherQLRaster
 import layouter
 
-str_buffer_size = 64        #keep this synchronous to reHDD
+# Suppress deprecation and printer warnings
-msg_queue_key = 0x1B11193C0 #keep this synchronous to reHDD
+warnings.filterwarnings("ignore", category=DeprecationWarning)
+logging.getLogger("brother_ql").setLevel(logging.ERROR)
+
+# Constants
+STR_BUFFER_SIZE = 64
+MSG_QUEUE_KEY = 0x1B11193C0
+IPC_CREAT = 0o1000
 
 file_name = "output.png"
 printer_path = "/dev/usb/lp0"
 
-def get_struct_format():
+terminate = False
-    #keep this synchronous to struct in reHDD
+
-    driveData = pycstruct.StructDef()
+
-    driveData.add('utf-8', 'driveIndex', length=str_buffer_size)
+class TDriveData(ctypes.Structure):
-    driveData.add('utf-8', 'driveHours', length=str_buffer_size)
+    _fields_ = [
-    driveData.add('utf-8', 'driveCycles', length=str_buffer_size)
+        ("caDriveIndex", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveErrors', length=str_buffer_size)
+        ("caDriveHours", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveShredTimestamp', length=str_buffer_size)
+        ("caDriveCycles", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveShredDuration', length=str_buffer_size)
+        ("caDriveErrors", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveCapacity', length=str_buffer_size)
+        ("caDriveShredTimestamp", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveState', length=str_buffer_size)
+        ("caDriveShredDuration", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveModelFamiliy', length=str_buffer_size)
+        ("caDriveCapacity", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveModelName', length=str_buffer_size)
+        ("caDriveState", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveSerialnumber', length=str_buffer_size)
+        ("caDriveModelFamily", ctypes.c_char * STR_BUFFER_SIZE),
-    driveData.add('utf-8', 'driveReHddVersion', length=str_buffer_size)
+        ("caDriveModelName", ctypes.c_char * STR_BUFFER_SIZE),
-    return driveData
+        ("caDriveSerialnumber", ctypes.c_char * STR_BUFFER_SIZE),
+        ("caDriveReHddVersion", ctypes.c_char * STR_BUFFER_SIZE),
+    ]
+
+
+class TMsgQueueData(ctypes.Structure):
+    _fields_ = [
+        ("msg_queue_type", ctypes.c_long),
+        ("driveData", TDriveData),
+    ]
+
+
+# IPC bindings
+libc = ctypes.CDLL("libc.so.6")
+msgget = libc.msgget
+msgrcv = libc.msgrcv
+
+msgget.argtypes = [ctypes.c_int, ctypes.c_int]
+msgget.restype = ctypes.c_int
+
+msgrcv.argtypes = [
+    ctypes.c_int,
+    ctypes.POINTER(TMsgQueueData),
+    ctypes.c_size_t,
+    ctypes.c_long,
+    ctypes.c_int,
+]
+msgrcv.restype = ctypes.c_ssize_t
+
+
+def signal_handler(signum, frame):
+    global terminate
+    print(f"Signal {signum} received, terminating...")
+    terminate = True
+
+
+signal.signal(signal.SIGINT, signal_handler)
+signal.signal(signal.SIGTERM, signal_handler)
+
+
+def wait_for_printer():
+    while not os.path.exists(printer_path):
+        print("Printer not found, waiting ...")
+        time.sleep(30)
+    return True
+
+
+def create_drive_objects(drive_info):
+    """Convert dictionary to layouter-compatible DriveData and ReHddInfo objects"""
+    drive = layouter.DriveData(
+        drive_index=int(drive_info["driveIndex"]),
+        drive_state=drive_info["driveState"],
+        modelfamily=drive_info["driveModelFamily"],
+        modelname=drive_info["driveModelName"],
+        capacity=int(drive_info["driveCapacity"]),
+        serialnumber=drive_info["driveSerialnumber"],
+        power_on_hours=int(drive_info["driveHours"]),
+        power_cycle=int(drive_info["driveCycles"]),
+        smart_error_count=int(drive_info["driveErrors"]),
+        shred_timestamp=int(drive_info["driveShredTimestamp"]),
+        shred_duration=int(drive_info["driveShredDuration"]),
+    )
+
+    rehdd_info = layouter.ReHddInfo(
+        link="https://git.mosad.xyz/localhorst/reHDD",
+        version=drive_info["driveReHddVersion"],
+    )
+
+    return drive, rehdd_info
+
+
+def worker(queue_id, test_mode=False):
+    try:
+        while not terminate:
+            if test_mode:
+                drive_info = {
+                    "driveIndex": "42",
+                    "driveHours": 44,
+                    "driveCycles": 45,
+                    "driveErrors": 43,
+                    "driveShredTimestamp": int(time.time()),
+                    "driveShredDuration": 0,
+                    "driveCapacity": 42,
+                    "driveState": "shredded",
+                    "driveModelFamily": "modelFamily",
+                    "driveModelName": "modelName",
+                    "driveSerialnumber": "serial",
+                    "driveReHddVersion": "V1.1.2",
+                }
+            else:
+                msg = TMsgQueueData()
+                result = msgrcv(
+                    queue_id,
+                    ctypes.byref(msg),
+                    ctypes.sizeof(TMsgQueueData) - ctypes.sizeof(ctypes.c_long),
+                    0,
+                    0,
+                )
+                if result == -1:
+                    err = ctypes.get_errno()
+                    print(f"Error reading from message queue: {os.strerror(err)}")
+                    break
+
+                d = msg.driveData
+                drive_info = {
+                    "driveIndex": d.caDriveIndex.decode().strip("\x00"),
+                    "driveHours": int(d.caDriveHours.decode().strip("\x00")),
+                    "driveCycles": int(d.caDriveCycles.decode().strip("\x00")),
+                    "driveErrors": int(d.caDriveErrors.decode().strip("\x00")),
+                    "driveShredTimestamp": int(
+                        d.caDriveShredTimestamp.decode().strip("\x00")
+                    ),
+                    "driveShredDuration": int(
+                        d.caDriveShredDuration.decode().strip("\x00")
+                    ),
+                    "driveCapacity": int(d.caDriveCapacity.decode().strip("\x00")),
+                    "driveState": d.caDriveState.decode().strip("\x00"),
+                    "driveModelFamily": d.caDriveModelFamily.decode().strip("\x00"),
+                    "driveModelName": d.caDriveModelName.decode().strip("\x00"),
+                    "driveSerialnumber": d.caDriveSerialnumber.decode().strip("\x00"),
+                    "driveReHddVersion": d.caDriveReHddVersion.decode().strip("\x00"),
+                }
+                time.sleep(3) 
+
+            print(f"Received Drive Data: {drive_info}")
+
+            drive_obj, rehdd_info = create_drive_objects(drive_info)
+            layouter.generate_image(drive_obj, rehdd_info, file_name)
+
+            if wait_for_printer():
+                qlr = BrotherQLRaster("QL-570")
+                image = Image.open(file_name)
+                create_label(qlr, image, "62")
+                with open(printer_path, "wb") as file:
+                    file.write(qlr.data)
+                os.remove(file_name)
+            else:
+                print("Skipping printing due to printer unavailability.")
+
+            if test_mode:
+                break
+
+    except Exception as e:
+        print(f"Worker encountered an error: {e}")
+
 
 def main():
-    try:
+    parser = argparse.ArgumentParser()
-        mq = sysv_ipc.MessageQueue(msg_queue_key, sysv_ipc.IPC_CREAT)
+    parser.add_argument(
+        "--test", action="store_true", help="Run in test mode with fake data"
+    )
+    args = parser.parse_args()
 
-        while True:
+    if args.test:
-            message, mtype = mq.receive()
+        print("Running in test mode.")
-            driveData = get_struct_format().deserialize(message)
+        worker(None, test_mode=True)
+        return
+    while True:
+        try:
+            queue_id = msgget(MSG_QUEUE_KEY, 0)
+            if queue_id == -1:
+                raise RuntimeError("Failed to connect to the existing message queue.")
+            worker(queue_id)
+        except Exception as e:
+            print(f"Main process encountered an error: {e}")
+            time.sleep(30)
 
-            rehdd_info = layouter.ReHddInfo("https://git.mosad.xyz/localhorst/reHDD", driveData['driveReHddVersion'])
-            drive = layouter.DriveData(
-            drive_index=int(driveData['driveIndex']),\
-            drive_state=str(driveData['driveState']),\
-            modelfamiliy=str(driveData['driveModelFamiliy']),\
-            modelname=str(driveData['driveModelName']),\
-            capacity=int(driveData['driveCapacity']),\
-            serialnumber=str(driveData['driveSerialnumber']),\
-            power_on_hours=int(driveData['driveHours']),\
-            power_cycle=int(driveData['driveCycles']),\
-            smart_error_count=int(driveData['driveErrors']),\
-            shred_timestamp=int(driveData['driveShredTimestamp']),\
-            shred_duration=int(driveData['driveShredDuration']))
-
-            while(not os.path.exists(printer_path)):
-                print("Printer not found, waiting ...")
-                time.sleep(30) #sleep 30
-
-            layouter.generate_image(drive, rehdd_info, file_name)
-            qlr = BrotherQLRaster("QL-570")
-            create_label(qlr, file_name, '62')
-
-            with open(printer_path, 'wb') as file:
-                file.write(qlr.data)
-            os.remove(file_name)
-    except sysv_ipc.ExistentialError:
-        print("ERROR: message queue creation failed")
 
 if __name__ == "__main__":
     main()