Skip to content

Commit

Permalink
Simplify FH properties
Browse files Browse the repository at this point in the history
  • Loading branch information
amc-corey-cox committed Nov 8, 2024
1 parent 5a76b66 commit d8fb3b4
Show file tree
Hide file tree
Showing 2 changed files with 8 additions and 13 deletions.
4 changes: 1 addition & 3 deletions src/koza/io/writer/jsonl_writer.py
Original file line number Diff line number Diff line change
@@ -1,13 +1,11 @@
import json
import os
from typing import List, Optional, TextIO
from typing import Optional, TextIO

from koza.io.writer.writer import KozaWriter


class JSONLWriter(KozaWriter):
node_properties: List[str]
edge_properties: List[str]
nodeFH: Optional[TextIO]
edgeFH: Optional[TextIO]

Expand Down
17 changes: 7 additions & 10 deletions src/koza/io/writer/tsv_writer.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
#### TSV Writer ####

from pathlib import Path
from typing import Dict, List, Literal, Set, TextIO
from typing import Dict, List, Literal, Optional, Set, TextIO

from ordered_set import OrderedSet

Expand All @@ -13,27 +13,24 @@ class TSVWriter(KozaWriter):
delimiter: str = "\t"
list_delimiter: str = "|"

nodes_file_name: Path
edges_file_name: Path

nodeFH: TextIO
edgeFH: TextIO
nodeFH: Optional[TextIO]
edgeFH: Optional[TextIO]

def init(self):
Path(self.output_dir).mkdir(parents=True, exist_ok=True)

if self.node_properties: # Make node file
self.node_properties = TSVWriter._order_columns(set(self.node_properties), "node")
self.nodes_file_name = Path(self.output_dir if self.output_dir else "", f"{self.source_name}_nodes.tsv")
self.nodeFH = open(self.nodes_file_name, "w")
nodes_file_name = Path(self.output_dir if self.output_dir else "", f"{self.source_name}_nodes.tsv")
self.nodeFH = open(nodes_file_name, "w")
self.nodeFH.write(self.delimiter.join(self.node_properties) + "\n")

if self.edge_properties: # Make edge file
if self.sssom_config:
self.edge_properties = self.add_sssom_columns(self.edge_properties)
self.edge_properties = TSVWriter._order_columns(set(self.edge_properties), "edge")
self.edges_file_name = Path(self.output_dir if self.output_dir else "", f"{self.source_name}_edges.tsv")
self.edgeFH = open(self.edges_file_name, "w")
edges_file_name = Path(self.output_dir if self.output_dir else "", f"{self.source_name}_edges.tsv")
self.edgeFH = open(edges_file_name, "w")
self.edgeFH.write(self.delimiter.join(self.edge_properties) + "\n")

def write_edge(self, edge: dict):
Expand Down

0 comments on commit d8fb3b4

Please sign in to comment.