Add entry points

.gitignore

@@ -69,3 +69,5 @@ lockfiles/
 
 # ruff cache
 .ruff_cache/
+
+src/aa_remove_data/P:2021.pb

pyproject.toml

@@ -12,7 +12,9 @@ classifiers = [
     "Programming Language :: Python :: 3.12",
 ]
 description = "A tool to remove some data from the Epics Archiver Appliance"
-dependencies = ["protobuf"] # Add project dependencies here, e.g. ["click", "numpy"]
+dependencies = [
+    "protobuf",
+] # Add project dependencies here, e.g. ["click", "numpy"]
 dynamic = ["version"]
 license.file = "LICENSE"
 readme = "README.md"
@@ -35,6 +37,8 @@ dev = [
 
 [project.scripts]
 aa-remove-data = "aa_remove_data.__main__:main"
+aa-print-header = "aa_remove_data.pb_utils:print_header"
+PB_2_TXT = "aa_remove_data.pb_utils:pb_2_txt"
 
 [project.urls]
 GitHub = "https://github.com/DiamondLightSource/aa-remove-data"
@@ -49,7 +53,7 @@ version_file = "src/aa_remove_data/_version.py"
 
 [tool.pyright]
 typeCheckingMode = "standard"
-reportMissingImports = false # Ignore missing stubs in imported modules
+reportMissingImports = false                # Ignore missing stubs in imported modules
 exclude = ["src/aa_remove_data/generated/"]
 
 [tool.pytest.ini_options]

src/aa_remove_data/pb_utils.py

@@ -1,6 +1,8 @@
+import argparse
 from collections.abc import Callable
 from datetime import datetime, timedelta
 from os import PathLike
+from pathlib import Path
 
 from aa_remove_data.generated import EPICSEvent_pb2
 
@@ -141,7 +143,7 @@ def read_pb(self, filepath: PathLike):
             lines = f.readlines()
         self.header.ParseFromString(first_line)
         proto_class = self.get_proto_class()
-        self.samples = [proto_class() for n in range(len(lines))]
+        self.samples = [proto_class() for _ in range(len(lines))]
         for i, sample in enumerate(self.samples):
             line = self._restore_newline_chars(lines[i].strip())
             sample.ParseFromString(line)
@@ -160,3 +162,42 @@ def write_pb(self, filepath: PathLike):
         ]
         with open(filepath, "wb") as f:
             f.writelines([header_b] + samples_b)
+
+
+def pb_2_txt():
+    """Convert a .pb file to a human-readable .txt file"""
+    parser = argparse.ArgumentParser()
+    parser.add_argument("pb_filename", type=str)
+    parser.add_argument("txt_filename", type=str)
+    args = parser.parse_args()
+    pb_file = Path(args.pb_filename)
+    txt_file = Path(args.txt_filename)
+    if not pb_file.is_file():
+        raise FileNotFoundError(f"The file {pb_file} does not exist.")
+    if pb_file.suffix != ".pb":
+        raise ValueError(f"Invalid file extension: '{pb_file.suffix}'. Expected '.pb'.")
+    pb = PBUtils(pb_file)
+    pb.write_to_txt(txt_file)
+
+
+def print_header():
+    parser = argparse.ArgumentParser()
+    parser.add_argument("pb_filename", type=str)
+    parser.add_argument("--lines", type=int, default=0)
+    args = parser.parse_args()
+    pb_file = Path(args.pb_filename)
+    lines = args.lines
+    if not pb_file.is_file():
+        raise FileNotFoundError(f"The file {pb_file} does not exist.")
+    if pb_file.suffix != ".pb":
+        raise ValueError(f"Invalid file extension: '{pb_file.suffix}'. Expected '.pb'.")
+    if lines < 0:
+        raise ValueError(f"Cannot have a negative number of lines ({lines}).")
+    pb = PBUtils(pb_file)
+    pvname = pb.header.pvname
+    year = pb.header.year
+    print(f"Name: {pvname}, Type: {pb.pv_type}, Year: {year}")
+    if lines > 0:
+        print(f"DATE{' ' * 19}SECONDS{' ' * 5}NANO{' ' * 9}VAL")
+        for i in range(lines):
+            print(pb.format_datastr(pb.samples[i], year).strip())