)->None:
```
------------------------------------------------------------------------
source
### sync_display
``` python
def sync_display(
cfg:Config, l:str
)->None:
```
``` python
!cat ../loghound/core.py
```
"""Fill in a module description here"""
# AUTOGENERATED! DO NOT EDIT! File to edit: ../nbs/00_core.ipynb.
# %% auto #0
__all__ = ['console', 'Styles', 'load_yaml', 'parse_line', 'Config', 'build_display_string', 'get_styles', 'display_line',
'read_lines', 'ConcurrentInfo', 'generate_row', 'sync_display', 'main']
# %% ../nbs/00_core.ipynb #915f694b
import yaml, re, sys
# %% ../nbs/00_core.ipynb #9a7a29b0
from dataclasses import dataclass, field
# %% ../nbs/00_core.ipynb #299e5a84
@dataclass
class Styles:
style: str;
match_identifiers: list[str] = field(default_factory=set);
# %% ../nbs/00_core.ipynb #a9c0af2a
def load_yaml(path: str) -> dict[str, str]:
contents = None
with open(path) as f: contents = f.read()
return yaml.safe_load(contents)
# %% ../nbs/00_core.ipynb #db902471
def parse_line(pattern: str, line: str) -> dict[str, str]:
if pattern is None: return {"_line": line}
m = re.match(pattern, line)
if m == None: return {"_line": f"unparseable line::{line}"}
return m.groupdict()
# %% ../nbs/00_core.ipynb #d0db9729
@dataclass
class Config:
timestamp_field: str;
concurrent: bool;
default_style: str;
match_identifier_field: str;
pattern: str | None = None;
format_str: str | None = None;
source_field: str | None = None;
styles: list[Styles] = field(default_factory=list);
def __post_init__(self):
self.match_mode = "exact" if self.pattern != None else "contains"
self.match_identifier_field = "_line" if self.pattern == None else self.match_identifier_field
if self.pattern and self.match_identifier_field not in re.compile(self.pattern).groupindex: raise ValueError("invalid config: match identifier not contained in pattern regex group")
if self.pattern and not self.format_str: raise ValueError("invalid config: pattern and format_str are needed")
# %% ../nbs/00_core.ipynb #4172d758
from fastcore.basics import patch
@patch(cls_method=True)
def from_yaml(cls: Config, path: str):
cnf_dict = load_yaml(path)
main = cnf_dict.get("parse")
styles = cnf_dict.get("styles")
rules = styles.get("rules", [])
res = Config(
pattern = main.get("pattern", None),
format_str= main.get("format_str", None),
source_field=main.get("source_field", ""),
timestamp_field= main.get("timestamp_field"),
concurrent = main.get("concurrent", False),
default_style=styles.get("default_style", "black"),
match_identifier_field= styles.get("match_field"),
styles = list(map(lambda d: Styles(match_identifiers=set(d.get("values", [])), style=d.get("style")), rules))
)
return res
# %% ../nbs/00_core.ipynb #c5ca36c4
@patch
def __repr__(self: Config):
return (f"Config(\n"
f" pattern: {self.pattern}\n"
f" timestamp_field: {self.timestamp_field}\n"
f" source_field: {self.source_field}\n"
f" match_field: {self.match_identifier_field}\n"
f" styles: {len(self.styles)} rules\n"
f")")
# %% ../nbs/00_core.ipynb #ec97589b
def build_display_string(line_metadata:dict[str, str], format: str ) -> str:
return format.format_map(line_metadata)
# %% ../nbs/00_core.ipynb #c298e716
def get_styles(cfg: Config, line_metadata: dict[str, str]) -> str:
match_identifier_value = line_metadata[cfg.match_identifier_field] if "_line" not in line_metadata else "_line"
if cfg.match_mode == "exact": matching_style: Styles = next((s for s in cfg.styles if match_identifier_value in s.match_identifiers), None)
else: matching_style: Styles = next((s for s in cfg.styles if any(ident in line_metadata["_line"] for ident in s.match_identifiers)), None)
return matching_style.style if matching_style else cfg.default_style
# %% ../nbs/00_core.ipynb #b34c192c
def display_line(cfg: Config, line_metadata: dict[str, str])-> None:
matching_style = get_styles(cfg, line_metadata)
text = line_metadata.get("_line", line) if cfg.match_mode == "contains" or "_line" in line_metadata else build_display_string(line_metadata, cfg.format_str)
console.print(text, style=matching_style, highlight=False)
return
# %% ../nbs/00_core.ipynb #28c48fba
def read_lines(file_name: str | None = None):
if not file_name and not sys.stdin.isatty():
for line in sys.stdin: yield line.strip()
elif file_name:
with open(file_name) as f:
for line in f: yield line.strip()
else:
raise Error("failure: unable to read input")
# %% ../nbs/00_core.ipynb #338fc5a5
from rich.console import Console
console = Console()
# %% ../nbs/00_core.ipynb #bdc932c9
@dataclass
class ConcurrentInfo:
source: str;
display_text: str;
ts: int;
style: list[str] = field(default_factory=list);
# %% ../nbs/00_core.ipynb #b512fb44
def generate_row(n_cols: int, source_col: int, log: ConcurrentInfo) -> list[str]:
row = ["" for _ in range(n_cols)]
row[source_col] = Text(log.display_text, style=log.style)
return row
# %% ../nbs/00_core.ipynb #99b41427
from rich.text import Text
from rich.table import Table
from fastcore.script import call_parse
def sync_display(cfg: Config, l: str) -> None:
line_metadata = parse_line(cfg.pattern, l)
display_line(cfg, line_metadata)
@dataclass
class ConcurrentInfo:
source: str;
display_text: str;
ts: int;
style: list[str] = field(default_factory=list);
@call_parse
def main(config:str=None, # Path to config YAML file
logfile:str=None,
concurrent:bool=False,
): # Path to log file (or pipe via stdin)
"Pretty print styled logs"
cfg = Config.from_yaml(config)
if not cfg.concurrent and not concurrent:
for l in read_lines(logfile): sync_display(cfg, l)
else:
processes = dict()
current_idx = 0
logs = []
for l in read_lines(logfile):
line_metadata = parse_line(cfg.pattern, l)
source = line_metadata[cfg.source_field]
if source not in processes:
processes[source] = current_idx
current_idx += 1
logs.append(
ConcurrentInfo(
source=line_metadata.get(cfg.source_field),
style=get_styles(cfg, line_metadata),
ts=line_metadata.get(cfg.timestamp_field),
display_text=build_display_string(line_metadata, cfg.format_str)))
t = Table()
t.add_column("time")
for k in processes: t.add_column(k)
for log in sorted(logs, key=lambda l: l.ts):
row = generate_row(len(processes), processes[log.source], log)
t.add_row(log.ts, *row)
console.print(t)
How can I test this nbdev package cli?
##### π€Replyπ€
π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§
There are a couple of approaches:
1. **Call
[`main`](https://jjmaturino.github.io/loghound/core.html#main)
directly in the notebook** β `@call_parse` lets you call it with
keyword args like a normal function:
``` python
main(config="path/to/your/config.yaml", logfile="path/to/some.log")
```
2. **As a CLI command** β if youβve set up `console_scripts` in your
`settings.ini`, you can run it from the terminal after
`pip install -e .`. Check your `settings.ini` β do you have a
`console_scripts` line pointing to `bloodhound.core:main`?
Do you have a sample YAML config and log file you can test with? That
would be the first thing to get in place if not.
``` python
main(config="../example_configs/raft_errors.yaml", logfile="../example_logs/raft.txt", concurrent=True)
```
ββββββββββ³βββββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββββ³βββββββββββββββββββββββββββββββββββ
β time β S0 β S1 β S2 β
β‘ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ©
β 000000 β 000000::S0::Starting up β β β
β 000000 β β 000000::S1::Starting up β β
β 000000 β β β 000000::S2::Starting up β
β 000001 β 000001::S0::Election timer β β β
β β started β β β
β 000001 β β 000001::S1::Election timer β β
β β β started β β
β 000001 β β β 000001::S2::Election timer β
β β β β started β
β 000045 β β β 000045::S2::Election timeout β
β β β β elapsed β
β 000046 β β β 000046::S2::Becoming candidate β
β β β β for term 1 β
β 000047 β β β 000047::S2::Requesting vote from β
β β β β S0 β
β 000047 β β β 000047::S2::Requesting vote from β
β β β β S1 β
β 000052 β 000052::S0::Granting vote to S2 β β β
β β for term 1 β β β
β 000053 β β 000053::S1::Granting vote to S2 β β
β β β for term 1 β β
β 000054 β β β 000054::S2::Became leader for β
β β β β term 1 β
β 000055 β β β 000055::S2::Leader, checking β
β β β β heartbeats β
β 000056 β β β 000056::S2::Sending β
β β β β AppendEntries to S0 β
β 000056 β β β 000056::S2::Sending β
β β β β AppendEntries to S1 β
β 000060 β 000060::S0::Received β β β
β β AppendEntries from S2 β β β
β 000060 β β 000060::S1::Received β β
β β β AppendEntries from S2 β β
β 000100 β β β 000100::S2::Received client β
β β β β request: Put(x, 1) β
β 000101 β β β 000101::S2::Appended entry at β
β β β β index 1 β
β 000102 β β β 000102::S2::Replicating entry 1 β
β β β β to S0 β
β 000102 β β β 000102::S2::Replicating entry 1 β
β β β β to S1 β
β 000108 β 000108::S0::Accepted entry 1 β β β
β β from S2 β β β
β 000109 β β 000109::S1::Accepted entry 1 β β
β β β from S2 β β
β 000110 β β β 000110::S2::Entry 1 committed β
β β β β (majority reached) β
β 000111 β β β 000111::S2::Applied command β
β β β β Put(x, 1) β
β 000155 β β β 000155::S2::Leader, checking β
β β β β heartbeats β
β 000156 β β β 000156::S2::Sending β
β β β β AppendEntries to S0 β
β 000156 β β β 000156::S2::Sending β
β β β β AppendEntries to S1 β
β 000200 β β β 000200::S2::Received client β
β β β β request: Put(y, 2) β
β 000201 β β β 000201::S2::Appended entry at β
β β β β index 2 β
β 000202 β β β 000202::S2::Replicating entry 2 β
β β β β to S0 β
β 000202 β β β 000202::S2::Replicating entry 2 β
β β β β to S1 β
β 000210 β 000210::S0::Accepted entry 2 β β β
β β from S2 β β β
β 000212 β β 000212::S1::Dropped β β
β β β AppendEntries (simulated network β β
β β β loss) β β
β 000213 β β β 000213::S2::Entry 2 committed β
β β β β (majority reached) β
β 000214 β β β 000214::S2::Applied command β
β β β β Put(y, 2) β
β 000255 β β β 000255::S2::Leader, checking β
β β β β heartbeats β
β 000256 β β β 000256::S2::Sending β
β β β β AppendEntries to S0 β
β 000256 β β β 000256::S2::Sending β
β β β β AppendEntries to S1 β
β 000260 β 000260::S0::Accepted heartbeat β β β
β β from S2 β β β
β 000262 β β 000262::S1::Accepted heartbeat β β
β β β from S2 β β
β 000300 β β 000300::S1::Failed to persist β β
β β β state: disk full β β
β 000301 β β 000301::S1::Retrying persist in β β
β β β 50ms β β
β 000350 β β 000350::S1::State persisted β β
β β β successfully β β
β 000400 β β β 000400::S2::Leader, checking β
β β β β heartbeats β
β 000500 β 000500::S0::Dropped heartbeat β β β
β β (simulated network loss) β β β
β 000501 β β 000501::S1::Dropped heartbeat β β
β β β (simulated network loss) β β
β 000600 β 000600::S0::Election timeout β β β
β β elapsed β β β
β 000601 β 000601::S0::Becoming candidate β β β
β β for term 2 β β β
β 000602 β 000602::S0::Requesting vote from β β β
β β S1 β β β
β 000602 β 000602::S0::Requesting vote from β β β
β β S2 β β β
β 000605 β β β 000605::S2::Stepping down, β
β β β β discovered higher term 2 β
β 000606 β β 000606::S1::Granting vote to S0 β β
β β β for term 2 β β
β 000607 β β β 000607::S2::Granting vote to S0 β
β β β β for term 2 β
β 000608 β 000608::S0::Became leader for β β β
β β term 2 β β β
β 000609 β 000609::S0::Leader, checking β β β
β β heartbeats β β β
β 000610 β 000610::S0::Sending β β β
β β AppendEntries to S1 β β β
β 000610 β 000610::S0::Sending β β β
β β AppendEntries to S2 β β β
β 000700 β 000700::S0::Received client β β β
β β request: Get(x) β β β
β 000701 β 000701::S0::Reading key x = 1 β β β
β 000702 β 000702::S0::Responded to client: β β β
β β x = 1 β β β
β 000800 β 000800::S0::Starting snapshot at β β β
β β index 2 β β β
β 000801 β 000801::S0::Snapshot complete, β β β
β β 128 bytes β β β
β 000802 β 000802::S0::Sending snapshot to β β β
β β S1 β β β
β 000810 β β 000810::S1::Installed snapshot β β
β β β from S0 β β
ββββββββββ΄βββββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββββ΄βββββββββββββββββββββββββββββββββββ
In order to include the 3rd party packages for nbdev, what must I do?
##### π€Replyπ€
π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§ π§
In your `settings.ini` file (at the root of your nbdev project), add the
third-party packages to the `requirements` field. For your project, that
would look something like:
``` ini
requirements = pyyaml rich fastcore
```
Each package name is space-separated. This ensures that when someone
installs your package (e.g.Β via `pip install`), those dependencies get
pulled in automatically.
Do you have a `settings.ini` already? If so, can you show it so we can
check whatβs there?