Coverage for hledger_lots/hl.py: 93%

1import json 

2import subprocess 

3import sys 

4from typing import List, Optional, Tuple 

5 

6from .lib import AdjustedTxn, Txn, get_avg_fifo, get_files_comm, get_xirr 

7 

8 

9def adjust_txn(txn: Txn) -> AdjustedTxn: 

10 price = txn.price if txn.type == "UnitPrice" else txn.price / txn.qtty 

11 

12 result = AdjustedTxn(txn.date, price, txn.base_cur, txn.qtty, txn.acct) 

13 return result 

14 

15 

16def prices_items2txn(date: str, prices_items: dict, account: str) -> Txn: 

17 price = prices_items["aprice"]["contents"]["aquantity"]["floatingPoint"] 

18 base_cur = prices_items["aprice"]["contents"]["acommodity"] 

19 qtty = prices_items["aquantity"]["floatingPoint"] 

20 price_type = prices_items["aprice"]["tag"] 

21 

22 txn = Txn(date, price, base_cur, qtty, account, price_type) 

23 return txn 

24 

25 

26def hledger2txn( 

27 file_path: Tuple[str, ...], cur: str, no_desc: Optional[str] = None 

28) -> List[AdjustedTxn]: 

29 files_comm = get_files_comm(file_path) 

30 comm = ["hledger", *files_comm, "print", f"cur:{cur}", "--output-format=json"] 

31 if no_desc: 

32 comm.append(f"not:desc:{no_desc}") 

33 

34 hl_proc = subprocess.run(comm, stdin=sys.stdin, capture_output=True) 

35 if hl_proc.returncode != 0: 

36 raise ValueError(hl_proc.stderr.decode("utf8")) 

37 

38 hl_data = hl_proc.stdout.decode("utf8") 

39 txns_list = json.loads(hl_data) 

40 

41 txns = [ 

42 prices_items2txn(txn["tdate"], prices_items, posting_items["paccount"]) 

43 for txn in txns_list 

44 for posting_items in txn["tpostings"] 

45 for prices_items in posting_items["pamount"] 

46 if prices_items["acommodity"].upper() == cur.upper() and prices_items["aprice"] 

47 ] 

48 

49 adjusted_txns = [adjust_txn(txn) for txn in txns] 

50 return adjusted_txns