Switch from pendulum to arrow (#492, #482)

In order to have Python 3.13 support

CONTRIBUTING.md

@@ -49,7 +49,7 @@ Please read their documentations on how to use them the best possible.
 - [DuniterPy](https://clients.pages.duniter.org/python/duniterpy/index.html): Autogenerated documentation.
   - Feel free to contribute upstream to share the code with other Python programs
 - [Click](https://click.palletsprojects.com/#documentation)
-- [Pendulum](https://pendulum.eustace.io/docs/)
+- [Arrow](https://arrow.readthedocs.io/)
 - [texttable](https://github.com/foutaise/texttable/#documentation)
 
 ## Pre-commit hooks

README.md

@@ -119,7 +119,7 @@ Silkaj is based on following Python modules:
 
 - [Click](https://click.palletsprojects.com/): Composable command line interface toolkit
 - [DuniterPy](https://git.duniter.org/clients/python/duniterpy/): Most complete client oriented Python library for Duniter/Ğ1 ecosystem
-- [Pendulum](https://pendulum.eustace.io/): Datetimes made easy
+- [Arrow](https://arrow.readthedocs.io/): Better dates & times for Python
 - [texttable](https://github.com/foutaise/texttable/): Creation of simple ASCII tables
 
 ### Names

pyproject.toml

@@ -18,7 +18,7 @@ python = "^3.9.0"
 duniterpy = "~1.2.0"
 rich-click = "^1.8.3"
 texttable = "^1.7.0"
-pendulum = "^3.0.0"
+arrow = "^1.3.0"
 pydiscourse = { version = "^1.7.0", optional = true }
 
 [tool.poetry.group.test.dependencies]

silkaj/blockchain/blocks.py

@@ -17,9 +17,9 @@ import time
 from operator import itemgetter
 from urllib.error import HTTPError
 
+import arrow
 import rich_click as click
 from duniterpy.api import bma
-from pendulum import from_timestamp
 
 from silkaj import tui
 from silkaj.blockchain.tools import get_head_block
@@ -49,11 +49,11 @@ def list_blocks(number: int, detailed: bool) -> None:
         issuer = {}
         issuer["pubkey"] = block["issuer"]
         if detailed or number <= 30:
-            gentime = from_timestamp(block["time"], tz="local").format(ALL)
-            mediantime = from_timestamp(block["medianTime"], tz="local").format(ALL)
             issuer["block"] = block["number"]
-            issuer["gentime"] = gentime
-            issuer["mediantime"] = mediantime
+            issuer["gentime"] = arrow.get(block["time"]).to("local").format(ALL)
+            issuer["mediantime"] = (
+                arrow.get(block["medianTime"]).to("local").format(ALL)
+            )
             issuer["hash"] = block["hash"][:10]
             issuer["powMin"] = block["powMin"]
         issuers_dict[issuer["pubkey"]] = issuer

silkaj/blockchain/difficulty.py

@@ -16,11 +16,11 @@
 from operator import itemgetter
 from os import system
 
+import arrow
 import jsonschema
 import rich_click as click
 from duniterpy.api import bma
 from duniterpy.api.client import WSConnection
-from pendulum import from_timestamp
 from websocket._exceptions import WebSocketConnectionClosedException
 
 from silkaj import network, tui
@@ -54,7 +54,7 @@ def display_diffi(current: WSConnection, diffi: dict) -> None:
         d["Π diffi"] = compute_power(match_pattern(d["level"])[1])
         d["Σ diffi"] = d.pop("level")
     system("cls||clear")
-    block_gen = from_timestamp(current["time"], tz="local").format(ALL)
+    block_gen = arrow.get(current["time"]).to("local").format(ALL)
     match = match_pattern(int(current["powMin"]))[0]
 
     table = tui.Table(style="columns").set_cols_dtype(["t", "t", "t", "i"])

silkaj/blockchain/information.py

@@ -13,8 +13,8 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
 
+import arrow
 import rich_click as click
-from pendulum import from_timestamp
 
 from silkaj.blockchain.tools import get_head_block
 from silkaj.constants import ALL
@@ -26,8 +26,8 @@ from silkaj.tools import get_currency_symbol
 def currency_info() -> None:
     head_block = get_head_block()
     ep = determine_endpoint()
-    current_time = from_timestamp(head_block["time"], tz="local")
-    mediantime = from_timestamp(head_block["medianTime"], tz="local")
+    current_time = arrow.get(head_block["time"]).to("local")
+    mediantime = arrow.get(head_block["medianTime"]).to("local")
     print(
         "Connected to node:",
         ep.host,
@@ -45,5 +45,5 @@ def currency_info() -> None:
         "\nMedian time:",
         mediantime.format(ALL),
         "\nDifference time:",
-        current_time.diff_for_humans(mediantime, True),
+        current_time - mediantime,
     )

silkaj/constants.py

@@ -39,10 +39,9 @@ MINIMAL_RELATIVE_TX_AMOUNT = 1e-6
 CENT_MULT_TO_UNIT = 100
 SHORT_PUBKEY_SIZE = 8
 
-# pendulum constants
-# see https://pendulum.eustace.io/docs/#localized-formats
-DATE = "LL"
-HOUR = "LTS"
-ALL = "LLL"
+# Arrow date time formats
+# https://arrow.readthedocs.io/en/latest/guide.html#supported-tokens
+DATE = "MMMM D, YYYY"
+ALL = "MMMM D, YYYY hh:mm A ZZZ"
 # Not ISO 8601 compliant but common
 ALL_DIGITAL = "YYYY-MM-DD HH:mm:ss"

silkaj/money/history.py

@@ -19,12 +19,12 @@ from pathlib import Path
 from typing import Any, Optional
 from urllib.error import HTTPError
 
+import arrow
 import rich_click as click
 from duniterpy.api.bma.tx import history
 from duniterpy.api.client import Client
 from duniterpy.documents.transaction import OutputSource, Transaction
 from duniterpy.grammars.output import Condition
-from pendulum import from_timestamp, now
 
 from silkaj.constants import ALL, ALL_DIGITAL
 from silkaj.money.tools import (
@@ -109,7 +109,7 @@ def generate_header(pubkey: str, currency_symbol: str, ud_value: int) -> str:
         idty = {"uid": ""}
     balance = get_amount_from_pubkey(pubkey)
     balance_ud = round(balance[1] / ud_value, 2)
-    date = now().format(ALL)
+    date = arrow.now().format(ALL)
     return f'Transactions history from: {idty["uid"]} {gen_pubkey_checksum(pubkey)}\n\
 Current balance: {balance[1] / 100} {currency_symbol}, {balance_ud} UD {currency_symbol} on {date}\n'
 
@@ -202,7 +202,7 @@ def parse_received_tx(
     identities = wt.identities_from_pubkeys(issuers, uids)
     for received_tx in received_txs:
         tx_list = []
-        tx_list.append(from_timestamp(received_tx.time, tz="local").format(ALL_DIGITAL))
+        tx_list.append(arrow.get(received_tx.time).to("local").format(ALL_DIGITAL))
         tx_list.append("")
         for i, issuer in enumerate(received_tx.issuers):
             tx_list[1] += prefix(None, None, i) + assign_idty_from_pubkey(
@@ -243,7 +243,7 @@ def parse_sent_tx(
     identities = wt.identities_from_pubkeys(pubkeys, uids)
     for sent_tx in sent_txs:
         tx_list = []
-        tx_list.append(from_timestamp(sent_tx.time, tz="local").format(ALL_DIGITAL))
+        tx_list.append(arrow.get(sent_tx.time).to("local").format(ALL_DIGITAL))
 
         total_amount, outputs = tx_amount(sent_tx, pubkey, sent_func)
         if len(outputs) > 1:

silkaj/wot/certification.py

@@ -15,12 +15,12 @@
 
 import sys
 
+import arrow
 import rich_click as click
 from duniterpy.api import bma
 from duniterpy.api.client import Client
 from duniterpy.documents import Block, BlockID, Certification, Identity, get_block_id
 from duniterpy.key import SigningKey
-from pendulum import from_timestamp, now
 
 from silkaj import tui
 from silkaj.auth import auth_method
@@ -103,7 +103,7 @@ def pre_checks(client: Client, issuer_pubkey: str, pubkey_to_certify: str) -> di
             # ĞT: 0<->4.8m - 4.8m + 12.5d
             renewable = cert["expiresIn"] - params["sigValidity"] + params["sigReplay"]
             if renewable > 0:
-                renewable_date = now().add(seconds=renewable).format(DATE)
+                renewable_date = arrow.now().shift(seconds=renewable).format(DATE)
                 sys.exit(f"Certification renewable from {renewable_date}")
 
     # Check if the certification is already in the pending certifications
@@ -125,7 +125,7 @@ def certification_confirmation(
     idty_timestamp = idty_to_certify["meta"]["timestamp"]
     block_id_idty = get_block_id(idty_timestamp)
     block = client(bma.blockchain.block, block_id_idty.number)
-    timestamp_date = from_timestamp(block["time"], tz="local").format(ALL)
+    timestamp_date = arrow.get(block["time"]).to("local").format(ALL)
     block_id_date = f": #{idty_timestamp[:15]}… {timestamp_date}"
     cert.append(["ID", issuer["uid"], "->", idty_to_certify["uid"] + block_id_date])
     cert.append(
@@ -137,8 +137,8 @@ def certification_confirmation(
         ],
     )
     params = bc_tools.get_blockchain_parameters()
-    cert_ends = now().add(seconds=params["sigValidity"]).format(DATE)
-    cert.append(["Valid", now().format(DATE), "—>", cert_ends])
+    cert_ends = arrow.now().shift(seconds=params["sigValidity"]).format(DATE)
+    cert.append(["Valid", arrow.now().format(DATE), "—>", cert_ends])
 
     table = tui.Table()
     table.fill_rows(

silkaj/wot/exclusions.py

@@ -19,7 +19,7 @@ import sys
 import time
 import urllib
 
-import pendulum
+import arrow
 import rich_click as click
 from duniterpy import constants as dp_const
 from duniterpy.api.bma import blockchain
@@ -222,9 +222,10 @@ def generate_identity_info(lookup, block, params):
     for i, certified in enumerate(lookup["signed"]):
         info += elements_inbetween_list(i, lookup["signed"])
         info += "@" + certified["uid"]
-    dt = pendulum.from_timestamp(block.mediantime + constants.ONE_HOUR, tz="local")
-    info += ".\n- **Exclu·e le** " + dt.format("LLLL", locale="fr")
-    info += " CET\n- **Raison de l'exclusion** : "
+    dt = arrow.get(block.mediantime).shift(hours=1).to(tz="local")
+    arrow_format = "dddd D MMMM YYYY HH:mm ZZZ"
+    info += ".\n- **Exclu·e le** " + dt.format(arrow_format, locale="fr")
+    info += "\n- **Raison de l'exclusion** : "
     if nbr_different_certifiers < params["sigQty"]:
         info += "manque de certifications"
     else:

silkaj/wot/idty_tools.py

@@ -18,7 +18,7 @@ import sys
 import urllib
 from typing import Union
 
-import pendulum
+import arrow
 import rich_click as click
 from duniterpy.api import bma
 from duniterpy.documents import BlockID, Identity, Revocation
@@ -40,9 +40,7 @@ def display_identity(idty: Identity) -> Texttable:
     id_table.append(["User ID", idty.uid])
     id_table.append(["Blockstamp", str(idty.block_id)])
     creation_block = client(bma.blockchain.block, idty.block_id.number)
-    creation_date = pendulum.from_timestamp(creation_block["time"], tz="local").format(
-        ALL,
-    )
+    creation_date = arrow.get(creation_block["time"]).to("local").format(ALL)
     id_table.append(["Created on", creation_date])
     # display infos
     table = Texttable(max_width=shutil.get_terminal_size().columns)

silkaj/wot/membership.py

@@ -16,7 +16,7 @@
 import logging
 import sys
 
-import pendulum
+import arrow
 import rich_click as click
 from duniterpy.api import bma
 from duniterpy.documents import BlockID, Membership, get_block_id
@@ -109,7 +109,7 @@ def display_confirmation_table(
 
     table = []
     if membership_expires:
-        expires = pendulum.now().add(seconds=membership_expires).diff_for_humans()
+        expires = arrow.now().shift(seconds=membership_expires).humanize()
         table.append(["Expiration date of current membership", expires])
 
     if pending_memberships:
@@ -123,7 +123,7 @@ def display_confirmation_table(
         table.append(
             [
                 "Pending membership documents will expire",
-                pendulum.now().add(seconds=pending_expires).diff_for_humans(),
+                arrow.now().shift(seconds=pending_expires).humanize(),
             ],
         )
 
@@ -136,7 +136,7 @@ def display_confirmation_table(
     table.append(
         [
             "Identity published",
-            pendulum.from_timestamp(block["time"], tz="local").format(DATE),
+            arrow.get(block["time"]).to("local").format(DATE),
         ],
     )
 
@@ -144,14 +144,14 @@ def display_confirmation_table(
     table.append(
         [
             "Expiration date of new membership",
-            pendulum.now().add(seconds=params["msValidity"]).diff_for_humans(),
+            arrow.now().shift(seconds=params["msValidity"]).humanize(),
         ],
     )
 
     table.append(
         [
             "Expiration date of new membership from the mempool",
-            pendulum.now().add(seconds=params["msPeriod"]).diff_for_humans(),
+            arrow.now().shift(seconds=params["msPeriod"]).humanize(),
         ],
     )
 

silkaj/wot/status.py

@@ -13,9 +13,9 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
 
+import arrow
 import rich_click as click
 from duniterpy.api.bma import blockchain, wot
-from pendulum import from_timestamp, now
 
 from silkaj.blockchain.tools import get_blockchain_parameters
 from silkaj.constants import DATE
@@ -58,7 +58,7 @@ def status(uid_pubkey: str) -> None:
         for req_cert in req["certifications"]:
             if req_cert["from"] == lookup_cert["pubkey"]:
                 certifications["received_expire"].append(
-                    now().add(seconds=req_cert["expiresIn"]).format(DATE),
+                    arrow.now().shift(seconds=req_cert["expiresIn"]).format(DATE),
                 )
                 certifications["received"].append(f'{lookup_cert["uids"][0]} ✔')
                 break
@@ -67,7 +67,7 @@ def status(uid_pubkey: str) -> None:
             f'{(wt.identity_of(pending_cert["from"]))["uid"]} ✘',
         )
         certifications["received_expire"].append(
-            from_timestamp(pending_cert["expires_on"], tz="local").format(DATE),
+            arrow.get(pending_cert["expires_on"]).to("local").format(DATE),
         )
     certifications["sent"], certifications["sent_expire"] = get_sent_certifications(
         signed,
@@ -103,12 +103,14 @@ def membership_status(certifications: dict, pubkey: str, req: dict) -> None:
     is_member = bool(member_lookup)
     print("member:", is_member)
     if req["revoked"]:
-        revoke_date = from_timestamp(req["revoked_on"], tz="local").format(DATE)
+        revoke_date = arrow.get(req["revoked_on"]).to("local").format(DATE)
         print(f"revoked: {req['revoked']}\nrevoked on: {revoke_date}")
     if not is_member and req["wasMember"]:
         print("expired:", req["expired"], "\nwasMember:", req["wasMember"])
     elif is_member:
-        expiration_date = now().add(seconds=req["membershipExpiresIn"]).format(DATE)
+        expiration_date = (
+            arrow.now().shift(seconds=req["membershipExpiresIn"]).format(DATE)
+        )
         print(f"Membership document expiration: {expiration_date}")
         print("Sentry:", req["isSentry"])
     print("outdistanced:", req["outdistanced"])
@@ -143,7 +145,7 @@ def expiration_date_from_block_id(
         date_approximation(block_id, time_first_block, params["avgGenTime"])
         + params["sigValidity"]
     )
-    return from_timestamp(expir_timestamp, tz="local").format(DATE)
+    return arrow.get(expir_timestamp).to("local").format(DATE)
 
 
 def date_approximation(block_id, time_first_block, avgentime):

tests/unit/wot/test_idty_tools.py

@@ -16,7 +16,7 @@
 import re
 import urllib
 
-import pendulum
+import arrow
 import pytest
 from duniterpy.api import bma
 from duniterpy.documents.block_id import BlockID
@@ -292,7 +292,7 @@ def test_display_identity(idty, monkeypatch, capsys):
     assert "| User ID    | Claude" in result
     assert "| Blockstamp | 597334" in result
     # idty_block returns a block at timestamp 1594980185,.
-    creation_time = pendulum.from_timestamp(idty_block["time"], tz="local").format(ALL)
+    creation_time = arrow.get(idty_block["time"]).to("local").format(ALL)
     assert creation_time in result
 
 

tests/unit/wot/test_membership.py

@@ -13,7 +13,7 @@
 # You should have received a copy of the GNU Affero General Public License
 # along with Silkaj. If not, see <https://www.gnu.org/licenses/>.
 
-import pendulum
+import arrow
 import pytest
 from duniterpy.api import bma
 from duniterpy.documents import Membership, get_block_id
@@ -73,7 +73,7 @@ def test_display_confirmation_table(patched_wot_requirements, monkeypatch, capsy
         table.append(
             [
                 "Expiration date of current membership",
-                pendulum.now().add(seconds=membership_expires).diff_for_humans(),
+                arrow.now().shift(seconds=membership_expires).humanize(),
             ],
         )
 
@@ -87,7 +87,7 @@ def test_display_confirmation_table(patched_wot_requirements, monkeypatch, capsy
         table.append(
             [
                 "Pending membership documents will expire",
-                pendulum.now().add(seconds=pending_expires).diff_for_humans(),
+                arrow.now().shift(seconds=pending_expires).humanize(),
             ],
         )
 
@@ -100,7 +100,7 @@ def test_display_confirmation_table(patched_wot_requirements, monkeypatch, capsy
     table.append(
         [
             "Identity published",
-            pendulum.from_timestamp(block["time"], tz="local").format(DATE),
+            arrow.get(block["time"]).to("local").format(DATE),
         ],
     )
 
@@ -108,14 +108,14 @@ def test_display_confirmation_table(patched_wot_requirements, monkeypatch, capsy
     table.append(
         [
             "Expiration date of new membership",
-            pendulum.now().add(seconds=params["msValidity"]).diff_for_humans(),
+            arrow.now().shift(seconds=params["msValidity"]).humanize(),
         ],
     )
 
     table.append(
         [
             "Expiration date of new membership from the mempool",
-            pendulum.now().add(seconds=params["msPeriod"]).diff_for_humans(),
+            arrow.now().shift(seconds=params["msPeriod"]).humanize(),
         ],
     )